wip

Author: BrennanConroy

src/SignalR/server/Core/src/HubConnectionHandler.cs

@@ -329,11 +329,11 @@ private async Task DispatchMessagesAsync(HubConnectionContext connection)
                     input.AdvanceTo(buffer.Start, buffer.End);
                 }
 
-                static async Task DispatchMessage(HubConnectionContext connection, HubDispatcher<THub> dispatcher, HubMessage message)
+                static Task DispatchMessage(HubConnectionContext connection, HubDispatcher<THub> dispatcher, HubMessage message)
                 {
                     connection.StopClientTimeout();
                     _ = ProcessTask(connection, dispatcher.DispatchMessageAsync(connection, message));
-                    await connection.ActiveInvocationLimit.WaitAsync();
+                    return connection.ActiveInvocationLimit.WaitAsync();
                 }
 
                 static async Task ProcessTask(HubConnectionContext connection, Task task)

src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs

@@ -349,6 +349,10 @@ await SendInvocationError(hubMethodInvocationMessage.InvocationId, connection,
 
                     if (isStreamResponse)
                     {
+                        // TODO: Need to sync this with CancelInvocationMessage
+                        cts = cts ?? CancellationTokenSource.CreateLinkedTokenSource(connection.ConnectionAborted);
+                        connection.ActiveRequestCancellationSources.TryAdd(hubMethodInvocationMessage.InvocationId, cts);
+
                         var result = await ExecuteHubMethod(methodExecutor, hub, arguments, connection, scope.ServiceProvider);
 
                         if (result == null)
@@ -359,8 +363,6 @@ await SendInvocationError(hubMethodInvocationMessage.InvocationId, connection,
                             return;
                         }
 
-                        cts = cts ?? CancellationTokenSource.CreateLinkedTokenSource(connection.ConnectionAborted);
-                        connection.ActiveRequestCancellationSources.TryAdd(hubMethodInvocationMessage.InvocationId, cts);
                         var enumerable = descriptor.FromReturnedStream(result, cts.Token);
 
                         Log.StreamingResult(_logger, hubMethodInvocationMessage.InvocationId, methodExecutor);

src/SignalR/server/SignalR/test/HubConnectionHandlerTests.cs

@@ -2976,6 +2976,8 @@ public async Task HubMethodInvokeCountsTowardsClientTimeoutIfParallelNotMaxed()
 
                     // Connection is closed
                     await connectionHandlerTask.OrTimeout();
+
+                    tcsService.EndMethod.SetResult(null);
                 }
             }
         }
@@ -3223,8 +3225,8 @@ public async Task PendingInvocationUnblockedWhenBlockingMethodCompletesWithParal
                     await tcsService.StartedMethod.Task.OrTimeout();
                     // Grab the tcs before resetting to use in the second long running method
                     var endTcs = tcsService.EndMethod;
-
                     tcsService.Reset();
+
                     // Long running hub invocation to test that other invocations will not run until it is completed
                     await client.SendInvocationAsync(nameof(LongRunningHub.LongRunningMethod), nonBlocking: false).OrTimeout();
                     // Wait for the long running method to start
@@ -3325,6 +3327,63 @@ public async Task StreamInvocationsBlockOtherInvocationsUntilTheyStartStreaming(
             }
         }
 
+        [Fact]
+        public async Task StreamInvocationsDoNotBlockOtherInvocationsWithParallelInvokes()
+        {
+            using (StartVerifiableLog())
+            {
+                var tcsService = new TcsService();
+                var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
+                {
+                    builder.AddSingleton(tcsService);
+                    builder.AddSingleton(typeof(IHubActivator<>), typeof(CustomHubActivator<>));
+
+                    builder.AddSignalR(options =>
+                    {
+                        options.MaxParallelInvocationsPerClient = 2;
+                    });
+                }, LoggerFactory);
+                var connectionHandler = serviceProvider.GetService<HubConnectionHandler<LongRunningHub>>();
+
+                // Because we use PipeScheduler.Inline the hub invocations will run inline until they wait, which happens inside the LongRunningMethod call
+                using (var client = new TestClient())
+                {
+                    var connectionHandlerTask = await client.ConnectAsync(connectionHandler).OrTimeout();
+
+                    // Long running hub invocation to test that other invocations will not run until it is completed
+                    var streamInvocationId = await client.SendStreamInvocationAsync(nameof(LongRunningHub.LongRunningStream), null).OrTimeout();
+                    // Wait for the long running method to start
+                    await tcsService.StartedMethod.Task.OrTimeout();
+
+                    // Invoke another hub method which will wait for the first method to finish
+                    await client.SendInvocationAsync(nameof(LongRunningHub.SimpleMethod), nonBlocking: false).OrTimeout();
+
+                    // simple hub method result
+                    var result = await client.ReadAsync().OrTimeout();
+                    var simpleCompletion = Assert.IsType<CompletionMessage>(result);
+                    Assert.Equal(21L, simpleCompletion.Result);
+
+                    // Release the long running hub method
+                    tcsService.EndMethod.TrySetResult(null);
+
+                    var hubActivator = serviceProvider.GetService<IHubActivator<LongRunningHub>>() as CustomHubActivator<LongRunningHub>;
+
+                    await client.SendHubMessageAsync(new CancelInvocationMessage(streamInvocationId)).OrTimeout();
+
+                    // Completion message for canceled Stream
+                    await client.ReadAsync().OrTimeout();
+
+                    // Shut down
+                    client.Dispose();
+
+                    await connectionHandlerTask.OrTimeout();
+
+                    // OnConnectedAsync, SimpleMethod, LongRunningStream, OnDisconnectedAsync
+                    Assert.Equal(4, hubActivator.ReleaseCount);
+                }
+            }
+        }
+
         [Fact]
         public async Task ServerSendsCloseWithErrorWhenConnectionClosedWithPartialMessage()
         {