Always subscribe to all ACKs on all servers

Prior to this fix, Groups.Add would fail to complete when there was no
subscription configured to receive the ACK on the server Groups.Add was
called on.

Author: halter73

src/Microsoft.AspNet.SignalR.Core/DefaultDependencyResolver.cs

@@ -63,6 +63,9 @@ private void RegisterDefaultServices()
             var ackHandler = new Lazy<AckHandler>();
             Register(typeof(IAckHandler), () => ackHandler.Value);
 
+            var serverMessageHandler = new Lazy<AckSubscriber>(() => new AckSubscriber(this));
+            Register(typeof(AckSubscriber), () => serverMessageHandler.Value);
+
             var perfCounterWriter = new Lazy<PerformanceCounterManager>(() => new PerformanceCounterManager(this));
             Register(typeof(IPerformanceCounterManager), () => perfCounterWriter.Value);
 

src/Microsoft.AspNet.SignalR.Core/Hubs/HubDispatcher.cs

@@ -408,8 +408,7 @@ protected override IList<string> GetSignals(string userId, string connectionId)
             })
             .Concat(new[] 
             { 
-                PrefixHelper.GetConnectionId(connectionId), 
-                PrefixHelper.GetAck(connectionId) 
+                PrefixHelper.GetConnectionId(connectionId)
             });
 
             return signals.ToList();

src/Microsoft.AspNet.SignalR.Core/Infrastructure/AckSubscriber.cs

@@ -0,0 +1,91 @@
+// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.md in the project root for license information.
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Threading.Tasks;
+using Microsoft.AspNet.SignalR.Messaging;
+
+namespace Microsoft.AspNet.SignalR.Infrastructure
+{
+    /// <summary>
+    /// A singleton that subscribes to all ACKs sent over the
+    /// <see cref="Microsoft.AspNet.SignalR.Messaging.IMessageBus"/> and
+    /// triggers any corresponding ACKs on the <see cref="IAckHandler"/>.
+    /// </summary>
+    internal class AckSubscriber : ISubscriber, IDisposable
+    {
+        private readonly IMessageBus _messageBus;
+        private readonly IAckHandler _ackHandler;
+        private IDisposable _subscription;
+
+        private const int MaxMessages = 10;
+
+        private static readonly string[] ServerSignals = new[] { Signal };
+
+        public AckSubscriber(IDependencyResolver resolver) :
+            this(resolver.Resolve<IMessageBus>(),
+                 resolver.Resolve<IAckHandler>())
+        {
+        }
+
+        public AckSubscriber(IMessageBus messageBus, IAckHandler ackHandler)
+        {
+            _messageBus = messageBus;
+            _ackHandler = ackHandler;
+
+            Identity = Guid.NewGuid().ToString();
+
+            ProcessMessages();
+        }
+
+        // The signal for all signalr servers
+        public const string Signal = "__SIGNALR__SERVER__";
+
+        public IList<string> EventKeys
+        {
+            get { return ServerSignals; }
+        }
+
+        public event Action<ISubscriber, string> EventKeyAdded
+        {
+            add { }
+            remove { }
+        }
+
+        public event Action<ISubscriber, string> EventKeyRemoved
+        {
+            add { }
+            remove { }
+        }
+
+        public Action<TextWriter> WriteCursor { get; set; }
+
+        public string Identity { get; private set; }
+
+        public Subscription Subscription { get; set; }
+
+        public void Dispose()
+        {
+            if (_subscription != null)
+            {
+                _subscription.Dispose();
+            }
+        }
+
+        private void ProcessMessages()
+        {
+            // Process messages that come from the bus for servers
+            _subscription = _messageBus.Subscribe(this, cursor: null, callback: TriggerAcks, maxMessages: MaxMessages, state: null);
+        }
+
+        private Task<bool> TriggerAcks(MessageResult result, object state)
+        {
+            result.Messages.Enumerate<object>(m => m.IsAck,
+                                              (s, m) => ((IAckHandler)s).TriggerAck(m.CommandId),
+                                              state: _ackHandler);
+
+            return TaskAsyncHelper.True;
+        }
+   }
+}

src/Microsoft.AspNet.SignalR.Core/Infrastructure/Connection.cs

@@ -94,7 +94,6 @@ public string Identity
             }
         }
 
-        [SuppressMessage("Microsoft.Performance", "CA1811:AvoidUncalledPrivateCode", Justification = "Used for debugging purposes.")]
         private TraceSource Trace
         {
             get
@@ -299,34 +298,27 @@ private bool ExcludeMessage(Message message)
 
         private void ProcessResults(MessageResult result)
         {
-            result.Messages.Enumerate<Connection>(message => message.IsAck || message.IsCommand,
+            result.Messages.Enumerate<Connection>(message => message.IsCommand,
                 (connection, message) => ProcessResultsCore(connection, message), this);
         }
 
         private static void ProcessResultsCore(Connection connection, Message message)
         {
             if (message.IsAck)
             {
-                connection._ackHandler.TriggerAck(message.CommandId);
+                connection.Trace.TraceError("Connection {0} received an unexpected ACK message.", connection.Identity);
+                return;
             }
-            else if (message.IsCommand)
-            {
-                var command = connection._serializer.Parse<Command>(message.Value, message.Encoding);
-                connection.ProcessCommand(command);
 
-                // Only send the ack if this command is waiting for it
-                if (message.WaitForAck)
-                {
-                    // If we're on the same box and there's a pending ack for this command then
-                    // just trip it
-                    if (!connection._ackHandler.TriggerAck(message.CommandId))
-                    {
-                        connection._bus.Ack(
-                            acker: connection._connectionId,
-                            waiter: message.Source,
-                            commandId: message.CommandId).Catch(connection._traceSource);
-                    }
-                }
+            var command = connection._serializer.Parse<Command>(message.Value, message.Encoding);
+            connection.ProcessCommand(command);
+
+            // Only send the ack if this command is waiting for it
+            if (message.WaitForAck)
+            {
+                connection._bus.Ack(
+                    acker: connection._connectionId,
+                    commandId: message.CommandId).Catch(connection._traceSource);
             }
         }
 

src/Microsoft.AspNet.SignalR.Core/Infrastructure/ConnectionManager.cs

@@ -120,6 +120,10 @@ internal Connection GetConnectionCore(string connectionName)
         {
             IList<string> signals = connectionName == null ? ListHelper<string>.Empty : new[] { connectionName };
 
+            // Ensure that this server is listening for any ACKs sent over the bus.
+            // This is important in case there are any calls to Groups.Add on a context.
+            _resolver.Resolve<AckSubscriber>();
+
             // Give this a unique id
             var connectionId = Guid.NewGuid().ToString();
             return new Connection(_resolver.Resolve<IMessageBus>(),

src/Microsoft.AspNet.SignalR.Core/Infrastructure/PrefixHelper.cs

@@ -21,7 +21,6 @@ internal static class PrefixHelper
 
         // Both
         internal const string ConnectionIdPrefix = "c-";
-        internal const string AckPrefix = "ack-";
 
         public static bool HasGroupPrefix(string value)
         {
@@ -64,11 +63,6 @@ public static string GetPersistentConnectionName(string connectionName)
             return PersistentConnectionPrefix + connectionName;
         }
 
-        public static string GetAck(string connectionId)
-        {
-            return AckPrefix + connectionId;
-        }
-
         public static IList<string> GetPrefixedConnectionIds(IList<string> connectionIds)
         {
             if (connectionIds.Count == 0)

src/Microsoft.AspNet.SignalR.Core/Messaging/MessageBusExtensions.cs

@@ -31,10 +31,10 @@ public static Task Publish(this IMessageBus bus, string source, string key, stri
             return bus.Publish(new Message(source, key, value));
         }
 
-        internal static Task Ack(this IMessageBus bus, string acker, string waiter, string commandId)
+        internal static Task Ack(this IMessageBus bus, string acker, string commandId)
         {
             // Prepare the ack
-            var message = new Message(acker, PrefixHelper.GetAck(waiter), null);
+            var message = new Message(acker, AckSubscriber.Signal, null);
             message.CommandId = commandId;
             message.IsAck = true;
             return bus.Publish(message);

src/Microsoft.AspNet.SignalR.Core/Microsoft.AspNet.SignalR.Core.csproj

@@ -270,6 +270,7 @@
     <Compile Include="Transports\ITransportConnection.cs" />
     <Compile Include="Json\JsonUtility.cs" />
     <Compile Include="GroupManager.cs" />
+    <Compile Include="Infrastructure\AckSubscriber.cs" />
     <Compile Include="Messaging\Command.cs" />
     <Compile Include="Resources.Designer.cs">
       <AutoGen>True</AutoGen>

src/Microsoft.AspNet.SignalR.Core/PersistentConnection.cs

@@ -59,6 +59,9 @@ public virtual void Initialize(IDependencyResolver resolver)
             _configurationManager = resolver.Resolve<IConfigurationManager>();
             _transportManager = resolver.Resolve<ITransportManager>();
 
+            // Ensure that this server is listening for any ACKs sent over the bus.
+            resolver.Resolve<AckSubscriber>();
+
             _initialized = true;
         }
 
@@ -393,12 +396,10 @@ private IList<string> GetDefaultSignals(string userId, string connectionId)
             // The list of default signals this connection cares about:
             // 1. The default signal (the type name)
             // 2. The connection id (so we can message this particular connection)
-            // 3. Ack signal
 
             return new string[] {
                 DefaultSignal,
-                PrefixHelper.GetConnectionId(connectionId),
-                PrefixHelper.GetAck(connectionId)
+                PrefixHelper.GetConnectionId(connectionId)
             };
         }
 

tests/Microsoft.AspNet.SignalR.FunctionalTests/Microsoft.AspNet.SignalR.FunctionalTests.csproj

@@ -90,6 +90,7 @@
     <Compile Include="Properties\AssemblyInfo.cs" />
     <Compile Include="Server\Connections\ConnectionFacts.cs" />
     <Compile Include="Server\Connections\DisconnectFacts.cs" />
+    <Compile Include="Server\Connections\FarmFacts.cs" />
     <Compile Include="Server\Connections\PersistentConnectionFacts.cs" />
     <Compile Include="Server\Connections\ReconnectFacts.cs" />
     <Compile Include="Server\Hubs\GetHubContextFacts.cs" />