Added a simple stress test

Author: ascott18

demo/DemoServer/ChatHub.cs

@@ -5,6 +5,14 @@
 
 namespace DemoServer
 {
+    public class UserIdProvider : IUserIdProvider
+    {
+        public virtual string GetUserId(HubConnectionContext connection)
+        {
+            return "staticUserid";
+        }
+    }
+
     public class ChatHubA : Hub
     {
         private readonly ILogger<ChatHubA> _logger;
@@ -53,7 +61,11 @@ public ChatHubB(ILogger<ChatHubB> logger)
         public async Task SendMessage(string name, string message)
         {
           //  _logger.LogInformation($"{nameof(SendMessage)} called. ConnectionId:{Context.ConnectionId}, Name:{name}, Message:{message}");
-            await Clients.Group(groupId).SendAsync("BroadcastMessage", name, message);
+            await Clients.Group(groupId).SendAsync("BroadcastMessage", name, "group");
+            await Clients.Caller.SendAsync("BroadcastMessage", name, "caller");
+            await Clients.User(Context.UserIdentifier).SendAsync("BroadcastMessage", name, "user");
+            await Clients.All.SendAsync("BroadcastMessage", name, "all");
+            await Clients.AllExcept(Context.ConnectionId).SendAsync("BroadcastMessage", name, "allExcept");
         }
 
         public override async Task OnConnectedAsync()

demo/DemoServer/Startup.cs

@@ -2,6 +2,7 @@
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.AspNetCore.HttpsPolicy;
+using Microsoft.AspNetCore.SignalR;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
@@ -24,6 +25,8 @@ public Startup(IConfiguration configuration)
         // This method gets called by the runtime. Use this method to add services to the container.
         public void ConfigureServices(IServiceCollection services)
         {
+            services.AddSingleton<IUserIdProvider, UserIdProvider>();
+
             services.AddRazorPages();
             services.AddSignalR()
                 .AddSqlServer(o =>

src/IntelliTect.AspNetCore.SignalR.SqlServer/SqlServerHubLifetimeManager.cs

@@ -403,9 +403,9 @@ private async Task EnsureSqlServerConnection()
 
                             _streams.Add(stream);
 
-                            StartReceiving(streamIndex);
+                            StartReceiving();
 
-                            void StartReceiving(int streamIndex)
+                            void StartReceiving()
                             {
                                 stream.StartReceiving((id, message) => OnReceived(message))
                                     .ContinueWith(async t =>
@@ -416,7 +416,7 @@ void StartReceiving(int streamIndex)
 
                                         // Try again in a little bit
                                         await Task.Delay(2000);
-                                        StartReceiving(streamIndex);
+                                        StartReceiving();
                                     }, TaskContinuationOptions.OnlyOnFaulted);
                             }
 

test/IntelliTect.AspNetCore.SignalR.SqlServer.Tests/SqlServerEndToEndTests.cs

@@ -49,7 +49,71 @@ public async Task CanSendAndReceivePayloads_WithPolling()
             await RunCore(options, prefix);
         }
 
-        private static async Task RunCore(SqlServerOptions options, string prefix)
+
+        [SkippableFact]
+        public async Task CanSendAndReceivePayloads_WithServiceBroker_UnderHeavyLoad()
+        {
+            await CreateDatabaseAsync();
+
+            var options = new SqlServerOptions
+            {
+                ConnectionString = connectionString,
+                AutoEnableServiceBroker = true,
+                Mode = SqlServerMessageMode.ServiceBroker
+            };
+
+            var prefix = nameof(CanSendAndReceivePayloads_WithServiceBroker_UnderHeavyLoad);
+            var installer = new SqlInstaller(options, NullLogger.Instance, prefix);
+            var receiver = new SqlReceiver(options, NullLogger.Instance, prefix + "_0", "");
+
+            var receivedMessages = new ConcurrentBag<byte[]>();
+            await installer.Install();
+            var receiverTask = receiver.Start((_, message) =>
+            {
+                receivedMessages.Add(message);
+                return Task.CompletedTask;
+            });
+            // Give the receiver time to reach a steady state (waiting).
+            await Task.Delay(150);
+
+            var cts = new CancellationTokenSource();
+
+            // This is roughly analagous to number of connections, not number of servers.
+            // The reasoning is that each connected client to the hub could be triggering
+            // the hub to be sending messages.
+            int numSenders = 100;
+            int numSent = 0;
+            var sender = new SqlSender(options, NullLogger.Instance, prefix + "_0");
+            for (int i = 0; i < numSenders; i++)
+            {
+                _ = Task.Run(async () =>
+                  {
+                      var random = new Random();
+                      while (!cts.IsCancellationRequested)
+                      {
+                          var payload = new byte[255];
+                          random.NextBytes(payload);
+                          await sender.Send(payload);
+                          Interlocked.Increment(ref numSent);
+                      }
+                  }, cts.Token);
+            }
+
+            var payload = new byte[255];
+            new Random().NextBytes(payload);
+            await Task.Delay(10000);
+            cts.Cancel();
+
+            // Give the receiver time to reach a steady state (waiting).
+            await Task.Delay(1000);
+
+            Assert.Equal(numSent, receivedMessages.Count);
+
+            receiver.Dispose();
+            await receiverTask;
+        }
+
+        private async Task RunCore(SqlServerOptions options, string prefix)
         {
             var installer = new SqlInstaller(options, NullLogger.Instance, prefix);
             var sender = new SqlSender(options, NullLogger.Instance, prefix + "_0");