Follow-up to #1611

* Name new property `DispatchConsumersAsyncEnabled`.
* Add a check on `BasicConsume` for when a regular dispatcher is used, and an async consumer passed.

Author: lukebakken

projects/RabbitMQ.Client/PublicAPI.Shipped.txt

@@ -511,7 +511,6 @@ RabbitMQ.Client.IConnection.Endpoint.get -> RabbitMQ.Client.AmqpTcpEndpoint
 RabbitMQ.Client.IConnection.FrameMax.get -> uint
 RabbitMQ.Client.IConnection.Heartbeat.get -> System.TimeSpan
 RabbitMQ.Client.IConnection.IsOpen.get -> bool
-RabbitMQ.Client.IConnection.DispatchConsumersAsync.get -> bool
 RabbitMQ.Client.IConnection.Protocol.get -> RabbitMQ.Client.IProtocol
 RabbitMQ.Client.IConnection.QueueNameChangedAfterRecovery -> System.EventHandler<RabbitMQ.Client.Events.QueueNameChangedAfterRecoveryEventArgs>
 RabbitMQ.Client.IConnection.RecoveringConsumer -> System.EventHandler<RabbitMQ.Client.Events.RecoveringConsumerEventArgs>

projects/RabbitMQ.Client/PublicAPI.Unshipped.txt

@@ -1 +1,2 @@
-RabbitMQ.Client.BasicProperties.BasicProperties(RabbitMQ.Client.ReadOnlyBasicProperties! input) -> void
+RabbitMQ.Client.BasicProperties.BasicProperties(RabbitMQ.Client.ReadOnlyBasicProperties! input) -> void
+RabbitMQ.Client.IConnection.DispatchConsumersAsyncEnabled.get -> bool

projects/RabbitMQ.Client/client/api/IConnection.cs

@@ -127,9 +127,9 @@ public interface IConnection : INetworkConnection, IDisposable
         IEnumerable<ShutdownReportEntry> ShutdownReport { get; }
 
         /// <summary>
-        /// Returns if connection is set to use asynchronous consumer dispatcher/>.
+        /// Returns <c>true</c> if the connection is set to use asynchronous consumer dispatchers.
         /// </summary>
-        public bool DispatchConsumersAsync { get; }
+        public bool DispatchConsumersAsyncEnabled { get; }
 
         /// <summary>
         /// Application-specific connection name, will be displayed in the management UI

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

@@ -176,7 +176,7 @@ public event EventHandler<RecoveringConsumerEventArgs> RecoveringConsumer
 
         public IProtocol Protocol => Endpoint.Protocol;
 
-        public bool DispatchConsumersAsync => _config.DispatchConsumersAsync;
+        public bool DispatchConsumersAsyncEnabled => _config.DispatchConsumersAsync;
 
         public async ValueTask<RecoveryAwareChannel> CreateNonRecoveringChannelAsync(CancellationToken cancellationToken)
         {

projects/RabbitMQ.Client/client/impl/ChannelBase.cs

@@ -975,12 +975,20 @@ public async Task<string> BasicConsumeAsync(string queue, bool autoAck, string c
         {
             if (ConsumerDispatcher is AsyncConsumerDispatcher)
             {
-                if (!(consumer is IAsyncBasicConsumer))
+                if (false == (consumer is IAsyncBasicConsumer))
                 {
                     throw new InvalidOperationException("When using an AsyncConsumerDispatcher, the consumer must implement IAsyncBasicConsumer");
                 }
             }
 
+            if (ConsumerDispatcher is ConsumerDispatcher)
+            {
+                if (consumer is IAsyncBasicConsumer)
+                {
+                    throw new InvalidOperationException("When using an ConsumerDispatcher, the consumer must not implement IAsyncBasicConsumer");
+                }
+            }
+
             // NOTE:
             // Maybe don't dispose this instance because the CancellationToken must remain
             // valid for processing the response.

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

@@ -101,7 +101,7 @@ internal Connection(ConnectionConfig config, IFrameHandler frameHandler)
         public int LocalPort => _frameHandler.LocalPort;
         public int RemotePort => _frameHandler.RemotePort;
 
-        public bool DispatchConsumersAsync => _config.DispatchConsumersAsync;
+        public bool DispatchConsumersAsyncEnabled => _config.DispatchConsumersAsync;
 
         public IDictionary<string, object?>? ServerProperties { get; private set; }
 

projects/Test/Integration/TestConsumer.cs

@@ -49,6 +49,24 @@ public TestConsumer(ITestOutputHelper output) : base(output)
         {
         }
 
+        [Fact]
+        public async Task AsyncConsumerShouldThrowInvalidOperationException()
+        {
+            bool sawException = false;
+            QueueDeclareOk q = await _channel.QueueDeclareAsync(string.Empty, false, false, false);
+            await _channel.BasicPublishAsync(string.Empty, q.QueueName, GetRandomBody(1024));
+            var consumer = new AsyncEventingBasicConsumer(_channel);
+            try
+            {
+                string consumerTag = await _channel.BasicConsumeAsync(q.QueueName, false, string.Empty, false, false, null, consumer);
+            }
+            catch (InvalidOperationException)
+            {
+                sawException = true;
+            }
+            Assert.True(sawException, "did not see expected InvalidOperationException");
+        }
+
         [Fact]
         public async Task TestBasicRoundtrip()
         {