Invoke Multiple Groups (#1254)

samples/ChatSample/PresenceHubLifetimeManager.cs

@@ -162,6 +162,11 @@ namespace ChatSample
             return _wrappedHubLifetimeManager.InvokeGroupAsync(groupName, methodName, args);
         }
 
+        public override Task InvokeGroupsAsync(IReadOnlyList<string> groupNames, string methodName, object[] args)
+        {
+            return _wrappedHubLifetimeManager.InvokeGroupsAsync(groupNames, methodName, args);
+        }
+
         public override Task InvokeUserAsync(string userId, string methodName, object[] args)
         {
             return _wrappedHubLifetimeManager.InvokeUserAsync(userId, methodName, args);

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

@@ -127,6 +127,29 @@ namespace Microsoft.AspNetCore.SignalR
             return Task.CompletedTask;
         }
 
+        public override Task InvokeGroupsAsync(IReadOnlyList<string> groupNames, string methodName, object[] args)
+        {
+            // Each task represents the list of tasks for each of the writes within a group
+            var tasks = new List<Task>();
+            var message = CreateInvocationMessage(methodName, args);
+
+            foreach (var groupName in groupNames)
+            {
+                if (string.IsNullOrEmpty(groupName))
+                {
+                    throw new ArgumentException(nameof(groupName));
+                }
+
+                var group = _groups[groupName];
+                if (group != null)
+                {
+                    tasks.Add(Task.WhenAll(group.Values.Select(c => c.WriteAsync(message))));
+                }
+            }
+
+            return Task.WhenAll(tasks);
+        }
+
         public override Task InvokeGroupExceptAsync(string groupName, string methodName, object[] args, IReadOnlyList<string> excludedIds)
         {
             if (groupName == null)

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

@@ -20,6 +20,7 @@ namespace Microsoft.AspNetCore.SignalR
         public dynamic Client(string connectionId) => new DynamicClientProxy(_clients.Client(connectionId));
         public dynamic Clients(IReadOnlyList<string> connectionIds) => new DynamicClientProxy(_clients.Clients(connectionIds));
         public dynamic Group(string groupName) => new DynamicClientProxy(_clients.Group(groupName));
+        public dynamic Groups(IReadOnlyList<string> groupNames) => new DynamicClientProxy(_clients.Groups(groupNames));
         public dynamic GroupExcept(string groupName, IReadOnlyList<string> excludedIds) => new DynamicClientProxy(_clients.GroupExcept(groupName, excludedIds));
         public dynamic OthersInGroup(string groupName) => new DynamicClientProxy(_clients.OthersInGroup(groupName));
         public dynamic Others => new DynamicClientProxy(_clients.Others);

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

@@ -39,6 +39,11 @@ namespace Microsoft.AspNetCore.SignalR
             return _hubClients.Group(groupName);
         }
 
+        public IClientProxy Groups(IReadOnlyList<string> groupNames)
+        {
+            return _hubClients.Groups(groupNames);
+        }
+
         public IClientProxy OthersInGroup(string groupName)
         {
             return _hubClients.GroupExcept(groupName, _currentConnectionId);

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

@@ -42,6 +42,11 @@ namespace Microsoft.AspNetCore.SignalR
             return new MultipleClientProxy<THub>(_lifetimeManager, connectionIds);
         }
 
+        public IClientProxy Groups(IReadOnlyList<string> groupNames)
+        {
+            return new MultipleGroupProxy<THub>(_lifetimeManager, groupNames);
+        }
+
         public IClientProxy User(string userId)
         {
             return new UserProxy<THub>(_lifetimeManager, userId);

src/Microsoft.AspNetCore.SignalR.Core/HubClients`T.cs

@@ -44,6 +44,11 @@ namespace Microsoft.AspNetCore.SignalR
             return TypedClientBuilder<T>.Build(new GroupExceptProxy<THub>(_lifetimeManager, groupName, excludeIds));
         }
 
+        public T Groups(IReadOnlyList<string> groupNames)
+        {
+            return TypedClientBuilder<T>.Build(new MultipleGroupProxy<THub>(_lifetimeManager, groupNames));
+        }
+
         public virtual T User(string userId)
         {
             return TypedClientBuilder<T>.Build(new UserProxy<THub>(_lifetimeManager, userId));

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

@@ -22,6 +22,8 @@ namespace Microsoft.AspNetCore.SignalR
 
         public abstract Task InvokeGroupAsync(string groupName, string methodName, object[] args);
 
+        public abstract Task InvokeGroupsAsync(IReadOnlyList<string> groupNames, string methodName, object[] args);
+
         public abstract Task InvokeGroupExceptAsync(string groupName, string methodName, object[] args, IReadOnlyList<string> excludedIds);
 
         public abstract Task InvokeUserAsync(string userId, string methodName, object[] args);

src/Microsoft.AspNetCore.SignalR.Core/IHubClients`T.cs

@@ -17,6 +17,8 @@ namespace Microsoft.AspNetCore.SignalR
 
         T Group(string groupName);
 
+        T Groups(IReadOnlyList<string> groupNames);
+
         T GroupExcept(string groupName, IReadOnlyList<string> excludeIds);
 
         T User(string userId);

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

@@ -40,6 +40,23 @@ namespace Microsoft.AspNetCore.SignalR
         }
     }
 
+    public class MultipleGroupProxy<THub> : IClientProxy
+    {
+        private readonly HubLifetimeManager<THub> _lifetimeManager;
+        private IReadOnlyList<string> _groupNames;
+
+        public MultipleGroupProxy(HubLifetimeManager<THub> lifetimeManager, IReadOnlyList<string> groupNames)
+        {
+            _lifetimeManager = lifetimeManager;
+            _groupNames = groupNames;
+        }
+
+        public Task InvokeAsync(string method, params object[] args)
+        {
+            return _lifetimeManager.InvokeGroupsAsync(_groupNames, method, args);
+        }
+    }
+
     public class GroupExceptProxy<THub> : IClientProxy
     {
         private readonly string _groupName;

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

@@ -42,6 +42,11 @@ namespace Microsoft.AspNetCore.SignalR
             return TypedClientBuilder<T>.Build(_hubClients.Clients(connectionIds));
         }
 
+        public T Groups(IReadOnlyList<string> groupNames)
+        {
+            return TypedClientBuilder<T>.Build(_hubClients.Groups(groupNames));
+        }
+
         public T OthersInGroup(string groupName)
         {
             return TypedClientBuilder<T>.Build(_hubClients.OthersInGroup(groupName));

src/Microsoft.AspNetCore.SignalR.Redis/RedisHubLifetimeManager.cs

@@ -601,6 +601,26 @@ namespace Microsoft.AspNetCore.SignalR.Redis
             return Task.WhenAll(publishTasks);
         }
 
+        public override Task InvokeGroupsAsync(IReadOnlyList<string> groupNames, string methodName, object[] args)
+        {
+            if (groupNames == null)
+            {
+                throw new ArgumentNullException(nameof(groupNames));
+            }
+            var publishTasks = new List<Task>(groupNames.Count);
+            var message = new RedisInvocationMessage(target: methodName, arguments: args);
+
+            foreach (var groupName in groupNames)
+            {
+                if (!string.IsNullOrEmpty(groupName))
+                {
+                    publishTasks.Add(PublishAsync(_channelNamePrefix + "." + groupName, message));
+                }
+            }
+
+            return Task.WhenAll(publishTasks);
+        }
+
         private class LoggerTextWriter : TextWriter
         {
             private readonly ILogger _logger;

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

@@ -47,6 +47,11 @@ namespace Microsoft.AspNetCore.SignalR.Tests.HubEndpointTestUtils
             return Clients.GroupExcept(groupName, excludedIds).InvokeAsync("Send", message);
         }
 
+        public Task SendToMultipleGroups(string message, IReadOnlyList<string> groupNames)
+        {
+            return Clients.Groups(groupNames).InvokeAsync("Send", message);
+        }
+
         public Task SendToOthersInGroup(string groupName, string message)
         {
             return Clients.OthersInGroup(groupName).InvokeAsync("Send", message);
@@ -206,6 +211,11 @@ namespace Microsoft.AspNetCore.SignalR.Tests.HubEndpointTestUtils
             return Clients.OthersInGroup(groupName).Send(message);
         }
 
+        public Task SendToMultipleGroups(string message, IReadOnlyList<string> groupNames)
+        {
+            return Clients.Groups(groupNames).Send(message);
+        }
+
         public Task BroadcastMethod(string message)
         {
             return Clients.All.Broadcast(message);
@@ -277,6 +287,11 @@ namespace Microsoft.AspNetCore.SignalR.Tests.HubEndpointTestUtils
             return Clients.GroupExcept(groupName, excludedIds).Send(message);
         }
 
+        public Task SendToMultipleGroups(string message, IReadOnlyList<string> groupNames)
+        {
+            return Clients.Groups(groupNames).Send(message);
+        }
+
         public Task SendToOthersInGroup(string groupName, string message)
         {
             return Clients.OthersInGroup(groupName).Send(message);

test/Microsoft.AspNetCore.SignalR.Tests/HubEndpointTests.cs

@@ -1123,6 +1123,46 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             }
         }
 
+        [Theory]
+        [MemberData(nameof(HubTypes))]
+        public async Task InvokeMultipleGroups(Type hubType)
+        {
+            dynamic endPoint = HubEndPointTestUtils.GetHubEndpoint(hubType);
+
+            using (var firstClient = new TestClient())
+            using (var secondClient = new TestClient())
+            {
+                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+
+                await Task.WhenAll(firstClient.Connected, secondClient.Connected).OrTimeout();
+
+                await secondClient.InvokeAsync(nameof(MethodHub.GroupAddMethod), "GroupA").OrTimeout();
+                await firstClient.InvokeAsync(nameof(MethodHub.GroupAddMethod), "GroupB").OrTimeout(); ;
+
+                var groupNames = new List<string> { "GroupA", "GroupB" };
+                await firstClient.SendInvocationAsync(nameof(MethodHub.SendToMultipleGroups), "test", groupNames).OrTimeout();
+
+                var hubMessage = await secondClient.ReadAsync().OrTimeout();
+                var invocation = Assert.IsType<InvocationMessage>(hubMessage);
+                Assert.Equal("Send", invocation.Target);
+                Assert.Single(invocation.Arguments);
+                Assert.Equal("test", invocation.Arguments[0]);
+
+                hubMessage = await firstClient.ReadAsync().OrTimeout();
+                invocation = Assert.IsType<InvocationMessage>(hubMessage);
+                Assert.Equal("Send", invocation.Target);
+                Assert.Single(invocation.Arguments);
+                Assert.Equal("test", invocation.Arguments[0]);
+
+                // kill the connections
+                firstClient.Dispose();
+                secondClient.Dispose();
+
+                await Task.WhenAll(firstEndPointTask, secondEndPointTask).OrTimeout();
+            }
+        }
+
         [Fact]
         public async Task RemoveFromGroupWhenNotInGroupDoesNotFail()
         {