Simplify error handling in HubConnectionHandler (#2162)

* Simplify error handling in HubConnectionHandler
- Since we execute hub methods inline, there's no need to abort the connection on unexpected exceptions.
- Don't pass the cancellation token to ReadAsync, instead just use CancelPendingRead.
- Don't treat OperationCancelledException errors as errors.

src/Microsoft.AspNetCore.SignalR.Core/HubConnectionContext.cs

@@ -59,8 +59,6 @@ namespace Microsoft.AspNetCore.SignalR
 
         public virtual IHubProtocol Protocol { get; internal set; }
 
-        internal ExceptionDispatchInfo AbortException { get; private set; }
-
         // Currently used only for streaming methods
         internal ConcurrentDictionary<string, CancellationTokenSource> ActiveRequestCancellationSources { get; } = new ConcurrentDictionary<string, CancellationTokenSource>(StringComparer.Ordinal);
 
@@ -270,6 +268,8 @@ namespace Microsoft.AspNetCore.SignalR
                 return;
             }
 
+            Input.CancelPendingRead();
+
             // We fire and forget since this can trigger user code to run
             Task.Factory.StartNew(_abortedCallback, this);
         }
@@ -295,6 +295,12 @@ namespace Microsoft.AspNetCore.SignalR
 
                         try
                         {
+                            if (result.IsCanceled)
+                            {
+                                Log.HandshakeCanceled(_logger);
+                                return false;
+                            }
+
                             if (!buffer.IsEmpty)
                             {
                                 if (HandshakeProtocol.TryParseRequestMessage(ref buffer, out var handshakeRequestMessage))
@@ -386,12 +392,6 @@ namespace Microsoft.AspNetCore.SignalR
             }
         }
 
-        internal void Abort(Exception exception)
-        {
-            AbortException = ExceptionDispatchInfo.Capture(exception);
-            Abort();
-        }
-
         // Used by the HubConnectionHandler only
         internal Task AbortAsync()
         {

src/Microsoft.AspNetCore.SignalR.Core/HubConnectionHandler.cs

@@ -100,6 +100,11 @@ namespace Microsoft.AspNetCore.SignalR
             {
                 await DispatchMessagesAsync(connection);
             }
+            catch (OperationCanceledException)
+            {
+                // Don't treat OperationCanceledException as an error, it's basically a "control flow"
+                // exception to stop things from running
+            }
             catch (Exception ex)
             {
                 Log.ErrorProcessingRequest(_logger, ex);
@@ -163,24 +168,25 @@ namespace Microsoft.AspNetCore.SignalR
         }
 
         private async Task DispatchMessagesAsync(HubConnectionContext connection)
-        {
-            try
         {
             var input = connection.Input;
             var protocol = connection.Protocol;
             while (true)
             {
-                    var result = await input.ReadAsync(connection.ConnectionAborted);
+                var result = await input.ReadAsync();
                 var buffer = result.Buffer;
 
                 try
                 {
+                    if (result.IsCanceled)
+                    {
+                        break;
+                    }
+
                     if (!buffer.IsEmpty)
                     {
                         while (protocol.TryParseMessage(ref buffer, _dispatcher, out var message))
                         {
-                                // Messages are dispatched sequentially and will block other messages from being processed until they complete.
-                                // Streaming methods will run sequentially until they start streaming, then they will fire-and-forget allowing other messages to run.
                             await _dispatcher.DispatchMessageAsync(connection, message);
                         }
                     }
@@ -198,12 +204,6 @@ namespace Microsoft.AspNetCore.SignalR
                 }
             }
         }
-            catch (OperationCanceledException)
-            {
-                // If there's an exception, bubble it to the caller
-                connection.AbortException?.Throw();
-            }
-        }
 
         private static class Log
         {

src/Microsoft.AspNetCore.SignalR.Core/Internal/DefaultHubDispatcher.cs

@@ -75,6 +75,9 @@ namespace Microsoft.AspNetCore.SignalR.Internal
 
         public override async Task DispatchMessageAsync(HubConnectionContext connection, HubMessage hubMessage)
         {
+            // Messages are dispatched sequentially and will stop other messages from being processed until they complete.
+            // Streaming methods will run sequentially until they start streaming, then they will fire-and-forget allowing other messages to run.
+
             switch (hubMessage)
             {
                 case InvocationBindingFailureMessage bindingFailureMessage:
@@ -142,10 +145,6 @@ namespace Microsoft.AspNetCore.SignalR.Internal
         private async Task ProcessInvocation(HubConnectionContext connection,
             HubMethodInvocationMessage hubMethodInvocationMessage, bool isStreamedInvocation)
         {
-            try
-            {
-                // If an unexpected exception occurs then we want to kill the entire connection
-                // by ending the processing loop
             if (!_methods.TryGetValue(hubMethodInvocationMessage.Target, out var descriptor))
             {
                 // Send an error to the client. Then let the normal completion process occur
@@ -158,12 +157,6 @@ namespace Microsoft.AspNetCore.SignalR.Internal
                 await Invoke(descriptor, connection, hubMethodInvocationMessage, isStreamedInvocation);
             }
         }
-            catch (Exception ex)
-            {
-                // Abort the entire connection if the invocation fails in an unexpected way
-                connection.Abort(ex);
-            }
-        }
 
         private async Task Invoke(HubMethodDescriptor descriptor, HubConnectionContext connection,
             HubMethodInvocationMessage hubMethodInvocationMessage, bool isStreamedInvocation)