Add streaming benchmarks (#1854)

benchmarks/Microsoft.AspNetCore.SignalR.Microbenchmarks/DefaultHubDispatcherBenchmark.cs

@@ -157,6 +157,27 @@ namespace Microsoft.AspNetCore.SignalR.Microbenchmarks
 
                 return new ValueTask<ChannelReader<int>>(channel);
             }
+
+            public ChannelReader<int> StreamChannelReaderCount(int count)
+            {
+                var channel = Channel.CreateUnbounded<int>();
+
+                _ = Task.Run(async () =>
+                {
+                    for (var i = 0; i < count; i++)
+                    {
+                        await channel.Writer.WriteAsync(i);
+                    }
+                    channel.Writer.Complete();
+                });
+
+                return channel.Reader;
+            }
+
+            public IObservable<int> StreamObservableCount(int count)
+            {
+                return Observable.Range(0, count);
+            }
         }
 
         [Benchmark]
@@ -224,5 +245,41 @@ namespace Microsoft.AspNetCore.SignalR.Microbenchmarks
         {
             return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamChannelReaderValueTaskAsync", null));
         }
+
+        [Benchmark]
+        public Task StreamChannelReaderCount_Zero()
+        {
+            return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamChannelReaderCount", argumentBindingException: null, new object[] { 0 }));
+        }
+
+        [Benchmark]
+        public Task StreamChannelReaderCount_One()
+        {
+            return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamChannelReaderCount", argumentBindingException: null, new object[] { 1 }));
+        }
+
+        [Benchmark]
+        public Task StreamChannelReaderCount_Thousand()
+        {
+            return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamChannelReaderCount", argumentBindingException: null, new object[] { 1000 }));
+        }
+
+        [Benchmark]
+        public Task StreamObservableCount_Zero()
+        {
+            return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamObservableCount", argumentBindingException: null, new object[] { 0 }));
+        }
+
+        [Benchmark]
+        public Task StreamObservableCount_One()
+        {
+            return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamObservableCount", argumentBindingException: null, new object[] { 1 }));
+        }
+
+        [Benchmark]
+        public Task StreamObservableCount_Thousand()
+        {
+            return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamObservableCount", argumentBindingException: null, new object[] { 1000 }));
+        }
     }
 }

test/Microsoft.AspNetCore.SignalR.Tests/HubEndpointTestUtils/Hubs.cs

@@ -529,7 +529,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
         {
             var channel = Channel.CreateUnbounded<string>();
 
-            var task = Task.Run(async () =>
+            _ = Task.Run(async () =>
             {
                 for (int i = 0; i < count; i++)
                 {