Invoke Users (#1257)

samples/ChatSample/PresenceHubLifetimeManager.cs

@@ -186,5 +186,10 @@ namespace ChatSample
         {
             return _wrappedHubLifetimeManager.InvokeGroupExceptAsync(groupName, methodName, args, excludedIds);
         }
+
+        public override Task InvokeUsersAsync(IReadOnlyList<string> userIds, string methodName, object[] args)
+        {
+            return _wrappedHubLifetimeManager.InvokeUsersAsync(userIds, methodName, args);
+        }
     }
 }

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

@@ -208,5 +208,13 @@ namespace Microsoft.AspNetCore.SignalR
                 return connectionIds.Contains(connection.ConnectionId);
             });
         }
+
+        public override Task InvokeUsersAsync(IReadOnlyList<string> userIds, string methodName, object[] args)
+        {
+            return InvokeAllWhere(methodName, args, connection =>
+            {
+                return userIds.Contains(connection.UserIdentifier);
+            });
+        }
     }
 }

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

@@ -25,5 +25,6 @@ namespace Microsoft.AspNetCore.SignalR
         public dynamic OthersInGroup(string groupName) => new DynamicClientProxy(_clients.OthersInGroup(groupName));
         public dynamic Others => new DynamicClientProxy(_clients.Others);
         public dynamic User(string userId) => new DynamicClientProxy(_clients.User(userId));
+        public dynamic Users(IReadOnlyList<string> users) => new DynamicClientProxy(_clients.Users(users));
     }
 }

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

@@ -63,5 +63,10 @@ namespace Microsoft.AspNetCore.SignalR
         {
             return _hubClients.Clients(connectionIds);
         }
+
+        public IClientProxy Users(IReadOnlyList<string> userIds)
+        {
+            return _hubClients.Users(userIds);
+        }
     }
 }

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

@@ -51,5 +51,10 @@ namespace Microsoft.AspNetCore.SignalR
         {
             return new UserProxy<THub>(_lifetimeManager, userId);
         }
+
+        public IClientProxy Users(IReadOnlyList<string> userIds)
+        {
+            return new MultipleUserProxy<THub>(_lifetimeManager, userIds);
+        }
     }
 }

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

@@ -1,9 +1,7 @@
 // Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
-using System;
 using System.Collections.Generic;
-using System.Text;
 
 namespace Microsoft.AspNetCore.SignalR
 {
@@ -53,5 +51,10 @@ namespace Microsoft.AspNetCore.SignalR
         {
             return TypedClientBuilder<T>.Build(new UserProxy<THub>(_lifetimeManager, userId));
         }
+
+        public virtual T Users(IReadOnlyList<string> userIds)
+        {
+            return TypedClientBuilder<T>.Build(new MultipleUserProxy<THub>(_lifetimeManager, userIds));
+        }
     }
 }

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

@@ -28,6 +28,8 @@ namespace Microsoft.AspNetCore.SignalR
 
         public abstract Task InvokeUserAsync(string userId, string methodName, object[] args);
 
+        public abstract Task InvokeUsersAsync(IReadOnlyList<string> userIds, string methodName, object[] args);
+
         public abstract Task AddGroupAsync(string connectionId, string groupName);
 
         public abstract Task RemoveGroupAsync(string connectionId, string groupName);

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

@@ -22,6 +22,8 @@ namespace Microsoft.AspNetCore.SignalR
         T GroupExcept(string groupName, IReadOnlyList<string> excludeIds);
 
         T User(string userId);
+
+        T Users(IReadOnlyList<string> userIds);
     }
 }
 

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

@@ -23,6 +23,23 @@ namespace Microsoft.AspNetCore.SignalR
         }
     }
 
+    public class MultipleUserProxy<THub> : IClientProxy
+    {
+        private readonly IReadOnlyList<string> _userIds;
+        private readonly HubLifetimeManager<THub> _lifetimeManager;
+
+        public MultipleUserProxy(HubLifetimeManager<THub> lifetimeManager, IReadOnlyList<string> userIds)
+        {
+            _lifetimeManager = lifetimeManager;
+            _userIds = userIds;
+        }
+
+        public Task InvokeAsync(string method, params object[] args)
+        {
+            return _lifetimeManager.InvokeUsersAsync(_userIds, method, args);
+        }
+    }
+
     public class GroupProxy<THub> : IClientProxy
     {
         private readonly string _groupName;

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

@@ -56,5 +56,10 @@ namespace Microsoft.AspNetCore.SignalR
         {
             return TypedClientBuilder<T>.Build(_hubClients.User(userId));
         }
+
+        public T Users(IReadOnlyList<string> userIds)
+        {
+            return TypedClientBuilder<T>.Build(_hubClients.Users(userIds));
+        }
     }
 }

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

@@ -621,6 +621,26 @@ namespace Microsoft.AspNetCore.SignalR.Redis
             return Task.WhenAll(publishTasks);
         }
 
+        public override Task InvokeUsersAsync(IReadOnlyList<string> userIds, string methodName, object[] args)
+        {
+            if (userIds.Count > 0)
+            {
+                var message = new RedisInvocationMessage(methodName, args);
+                var publishTasks = new List<Task>(userIds.Count);
+                foreach (var userId in userIds)
+                {
+                    if (!string.IsNullOrEmpty(userId))
+                    {
+                        publishTasks.Add(PublishAsync(_channelNamePrefix + ".user." + userId, message));
+                    }
+                }
+
+                return Task.WhenAll(publishTasks);
+            }
+
+            return Task.CompletedTask;
+        }
+
         private class LoggerTextWriter : TextWriter
         {
             private readonly ILogger _logger;

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

@@ -22,6 +22,11 @@ namespace Microsoft.AspNetCore.SignalR.Tests.HubEndpointTestUtils
             return Clients.User(userId).InvokeAsync("Send", message);
         }
 
+        public Task SendToMultipleUsers(IReadOnlyList<string> userIds, string message)
+        {
+            return Clients.Users(userIds).InvokeAsync("Send", message);
+        }
+
         public Task ConnectionSendMethod(string connectionId, string message)
         {
             return Clients.Client(connectionId).InvokeAsync("Send", message);
@@ -181,6 +186,11 @@ namespace Microsoft.AspNetCore.SignalR.Tests.HubEndpointTestUtils
             return Clients.User(userId).Send(message);
         }
 
+        public Task SendToMultipleUsers(IReadOnlyList<string> userIds, string message)
+        {
+            return Clients.Users(userIds).Send(message);
+        }
+
         public Task ConnectionSendMethod(string connectionId, string message)
         {
             return Clients.Client(connectionId).Send(message);
@@ -256,6 +266,11 @@ namespace Microsoft.AspNetCore.SignalR.Tests.HubEndpointTestUtils
             return Clients.User(userId).Send(message);
         }
 
+        public Task SendToMultipleUsers(IReadOnlyList<string> userIds, string message)
+        {
+            return Clients.Users(userIds).Send(message);
+        }
+
         public Task ConnectionSendMethod(string connectionId, string message)
         {
             return Clients.Client(connectionId).Send(message);

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

@@ -956,8 +956,6 @@ namespace Microsoft.AspNetCore.SignalR.Tests
                 var secondAndThirdClients = new HashSet<string> {secondClient.Connection.ConnectionId,
                     thirdClient.Connection.ConnectionId };
 
-                secondAndThirdClients.Add(secondClient.Connection.ConnectionId);
-
                 await firstClient.SendInvocationAsync("SendToMultipleClients", "Second and Third", secondAndThirdClients).OrTimeout();
 
                 var secondClientResult = await secondClient.ReadAsync().OrTimeout();
@@ -984,6 +982,51 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             }
         }
 
+        [Theory]
+        [MemberData(nameof(HubTypes))]
+        public async Task SendToMultipleUsers(Type hubType)
+        {
+            dynamic endPoint = HubEndPointTestUtils.GetHubEndpoint(hubType);
+
+            using (var firstClient = new TestClient(addClaimId: true))
+            using (var secondClient = new TestClient(addClaimId: true))
+            using (var thirdClient = new TestClient(addClaimId: true))
+            {
+                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+                Task thirdEndPointTask = endPoint.OnConnectedAsync(thirdClient.Connection);
+
+                await Task.WhenAll(firstClient.Connected, secondClient.Connected, thirdClient.Connected).OrTimeout();
+
+                var secondAndThirdClients = new HashSet<string> {secondClient.Connection.User.FindFirst(ClaimTypes.NameIdentifier)?.Value,
+                    thirdClient.Connection.User.FindFirst(ClaimTypes.NameIdentifier)?.Value };
+
+                await firstClient.SendInvocationAsync(nameof(MethodHub.SendToMultipleUsers), secondAndThirdClients, "Second and Third").OrTimeout();
+
+                var secondClientResult = await secondClient.ReadAsync().OrTimeout();
+                var invocation = Assert.IsType<InvocationMessage>(secondClientResult);
+                Assert.Equal("Send", invocation.Target);
+                Assert.Equal("Second and Third", invocation.Arguments[0]);
+
+                var thirdClientResult = await thirdClient.ReadAsync().OrTimeout();
+                invocation = Assert.IsType<InvocationMessage>(thirdClientResult);
+                Assert.Equal("Send", invocation.Target);
+                Assert.Equal("Second and Third", invocation.Arguments[0]);
+
+                // Check that first client only got the completion message
+                var hubMessage = await firstClient.ReadAsync().OrTimeout();
+                Assert.IsType<CompletionMessage>(hubMessage);
+                Assert.Null(firstClient.TryRead());
+
+                // kill the connections
+                firstClient.Dispose();
+                secondClient.Dispose();
+                thirdClient.Dispose();
+
+                await Task.WhenAll(firstEndPointTask, secondEndPointTask, thirdEndPointTask).OrTimeout();
+            }
+        }
+
         [Theory]
         [MemberData(nameof(HubTypes))]
         public async Task HubsCanAddAndSendToGroup(Type hubType)