AsyncConsumerDispatcher uses ConcurrentConsumerDispatcher as shim

Author: danielmarbach

projects/client/RabbitMQ.Client/src/client/api/IAsyncConnection.cs

@@ -20,13 +20,13 @@ public void Quiesce()
         public void Shutdown()
         {
             // necessary evil
-            this.workService.StopWork().GetAwaiter().GetResult();
+            this.workService.Stop().GetAwaiter().GetResult();
         }
 
         public void Shutdown(IModel model)
         {
             // necessary evil
-            this.workService.StopWork(model).GetAwaiter().GetResult();
+            this.workService.Stop(model).GetAwaiter().GetResult();
         }
 
         public bool IsShutdown

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

@@ -6,7 +6,7 @@
 
 namespace RabbitMQ.Client
 {
-    internal class AsyncConsumerWorkService
+    internal class AsyncConsumerWorkService : ConsumerWorkService
     {
         readonly ConcurrentDictionary<IModel, WorkPool> workPools = new ConcurrentDictionary<IModel, WorkPool>();
 
@@ -32,7 +32,7 @@ public void Schedule<TWork>(ModelBase model, TWork work)
             workPool.Enqueue(work);
         }
 
-        public async Task StopWork(IModel model)
+        public async Task Stop(IModel model)
         {
             WorkPool workPool;
             if (workPools.TryRemove(model, out workPool))
@@ -41,11 +41,11 @@ public async Task StopWork(IModel model)
             }
         }
 
-        public async Task StopWork()
+        public async Task Stop()
         {
             foreach (var model in workPools.Keys)
             {
-                await StopWork(model).ConfigureAwait(false);
+                await Stop(model).ConfigureAwait(false);
             }
         }
 

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

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

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

@@ -120,7 +120,16 @@ public Connection(IConnectionFactory factory, bool insist, IFrameHandler frameHa
             FrameMax = 0;
             m_factory = factory;
             m_frameHandler = frameHandler;
-            ConsumerWorkService = new ConsumerWorkService();
+
+            var asyncConnectionFactory = factory as IAsyncConnectionFactory;
+            if (asyncConnectionFactory != null && asyncConnectionFactory.DispatchConsumersAsync)
+            {
+                ConsumerWorkService = new AsyncConsumerWorkService();
+            }
+            else
+            {
+                ConsumerWorkService = new ConsumerWorkService();
+            }
 
             m_sessionManager = new SessionManager(this, 0);
             m_session0 = new MainSession(this) { Handler = NotifyReceivedCloseOk };

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

@@ -96,10 +96,10 @@ public ModelBase(ISession session)
 
         public ModelBase(ISession session, ConsumerWorkService workService)
         {
-            var asyncConnection = session.Connection as IAsyncConnection;
-            if (asyncConnection != null)
+            var asyncConsumerWorkService = workService as AsyncConsumerWorkService;
+            if (asyncConsumerWorkService != null)
             {
-                ConsumerDispatcher = new AsyncConsumerDispatcher(this, asyncConnection.AsyncConsumerWorkService);
+                ConsumerDispatcher = new AsyncConsumerDispatcher(this, asyncConsumerWorkService);
             }
             else
             {
@@ -1168,8 +1168,9 @@ public string BasicConsume(string queue,
             IDictionary<string, object> arguments,
             IBasicConsumer consumer)
         {
-            var sessionConnection = Session.Connection as IAsyncConnection;
-            if (sessionConnection != null)
+            // TODO: Replace with flag
+            var asyncDispatcher = ConsumerDispatcher as AsyncConsumerDispatcher;
+            if (asyncDispatcher != null)
             {
                 var asyncConsumer = consumer as IAsyncBasicConsumer;
                 if (asyncConsumer == null)

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

@@ -131,11 +131,6 @@ 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);
         }
 
@@ -145,11 +140,6 @@ 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);
         }
 

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

@@ -12,15 +12,15 @@ protected Work(IBasicConsumer consumer)
             asyncConsumer = (IAsyncBasicConsumer)consumer;
         }
 
-        public Task Execute(ModelBase model)
+        public async Task Execute(ModelBase model)
         {
             try
             {
-                return Execute(model, asyncConsumer);
+                await Execute(model, asyncConsumer).ConfigureAwait(false);
             }
             catch (Exception)
             {
-                return TaskExtensions.CompletedTask;
+               // intentionally caught
             }
         }