AsyncConnectionDecorator renamed to AsyncConnection, AsyncConnection now inherits from Connection

AutorecoveringConnection and ProtocolBase create correct connection depending on settings

Author: danielmarbach

projects/client/RabbitMQ.Client/src/client/impl/AsyncConnection.cs

@@ -0,0 +1,15 @@
+using RabbitMQ.Client.Impl;
+
+namespace RabbitMQ.Client.Framing.Impl
+{
+    internal class AsyncConnection : Connection, IAsyncConnection
+    {
+        public AsyncConnection(IConnectionFactory factory, bool insist, IFrameHandler frameHandler, string clientProvidedName = null) : 
+            base(factory, insist, frameHandler, clientProvidedName)
+        {
+            AsyncConsumerWorkService = new AsyncConsumerWorkService();
+        }
+
+        public AsyncConsumerWorkService AsyncConsumerWorkService { get; }
+    }
+}

projects/client/RabbitMQ.Client/src/client/impl/AsyncConnectionDecorator.cs

@@ -626,8 +626,8 @@ public void Init(IEndpointResolver endpoints)
 
         private void Init(IFrameHandler fh)
         {
-            m_delegate = new Connection(m_factory, false,
-                fh, this.ClientProvidedName);
+            m_delegate = m_factory.DispatchConsumersAsync ? new AsyncConnection(m_factory, false, fh, this.ClientProvidedName) 
+                : new Connection(m_factory, false, fh, this.ClientProvidedName);
 
             AutorecoveringConnection self = this;
             EventHandler<ShutdownEventArgs> recoveryListener = (_, args) =>
@@ -831,7 +831,8 @@ protected bool RecoverConnectionDelegate()
                 try
                 {
                     var fh = endpoints.SelectOne(m_factory.CreateFrameHandler);
-                    m_delegate = new Connection(m_factory, false, fh, this.ClientProvidedName);
+                    m_delegate = m_factory.DispatchConsumersAsync ? new AsyncConnection(m_factory, false, fh, this.ClientProvidedName)
+                        : new Connection(m_factory, false, fh, this.ClientProvidedName);
                     return true;
                 }
                 catch (Exception e)

projects/client/RabbitMQ.Client/src/client/impl/AutorecoveringConnection.cs

@@ -131,6 +131,11 @@ public IConnection CreateConnection(IConnectionFactory factory,
             bool insist,
             IFrameHandler frameHandler)
         {
+            var connectionFactory = factory as IAsyncConnectionFactory;
+            if (connectionFactory != null && connectionFactory.DispatchConsumersAsync)
+            {
+                return new AsyncConnection(factory, insist, frameHandler, null);
+            } 
             return new Connection(factory, insist, frameHandler, null);
         }
 
@@ -140,6 +145,11 @@ public IConnection CreateConnection(IConnectionFactory factory,
             IFrameHandler frameHandler,
             string clientProvidedName)
         {
+            var connectionFactory = factory as IAsyncConnectionFactory;
+            if (connectionFactory != null && connectionFactory.DispatchConsumersAsync)
+            {
+                return new AsyncConnection(factory, insist, frameHandler, clientProvidedName);
+            }
             return new Connection(factory, insist, frameHandler, clientProvidedName);
         }