Merge pull request #1671 from rabbitmq/rabbitmq-dotnet-client-1669-followup

lukebakken · web-flow · commit 735bbca388b2 · 2024-09-12T11:10:30.000-07:00
Follow-up to #1669 - per-channel dispatch concurrency
diff --git a/projects/RabbitMQ.Client/PublicAPI.Shipped.txt b/projects/RabbitMQ.Client/PublicAPI.Shipped.txt
@@ -892,7 +892,6 @@ static RabbitMQ.Client.IChannelExtensions.BasicPublishAsync(this RabbitMQ.Client
 static RabbitMQ.Client.IChannelExtensions.BasicPublishAsync(this RabbitMQ.Client.IChannel! channel, string! exchange, string! routingKey, bool mandatory, System.ReadOnlyMemory<byte> body, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
 static RabbitMQ.Client.IChannelExtensions.BasicPublishAsync(this RabbitMQ.Client.IChannel! channel, string! exchange, string! routingKey, System.ReadOnlyMemory<byte> body, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
 RabbitMQ.Client.IChannel.ConfirmSelectAsync(bool trackConfirmations = true, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
-const RabbitMQ.Client.ConnectionFactory.DefaultConsumerDispatchConcurrency = 1 -> ushort
-RabbitMQ.Client.IConnection.CreateChannelAsync(ushort consumerDispatchConcurrency, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.Client.IChannel!>!
-static RabbitMQ.Client.IConnectionExtensions.CreateChannelAsync(this RabbitMQ.Client.IConnection! connection, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.Client.IChannel!>!
+const RabbitMQ.Client.Constants.DefaultConsumerDispatchConcurrency = 1 -> ushort
 readonly RabbitMQ.Client.ConnectionConfig.ConsumerDispatchConcurrency -> ushort
+RabbitMQ.Client.IConnection.CreateChannelAsync(ushort? consumerDispatchConcurrency = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.Client.IChannel!>!
diff --git a/projects/RabbitMQ.Client/client/api/ConnectionFactory.cs b/projects/RabbitMQ.Client/client/api/ConnectionFactory.cs
@@ -92,11 +92,6 @@ namespace RabbitMQ.Client
     ///hosts with an empty name are not addressable. </para></remarks>
     public sealed class ConnectionFactory : ConnectionFactoryBase, IConnectionFactory
     {
-        /// <summary>
-        /// Default value for consumer dispatch concurrency.
-        /// </summary>
-        public const ushort DefaultConsumerDispatchConcurrency = 1;
-
         /// <summary>
         /// Default value for the desired maximum channel number. Default: 2047.
         /// </summary>
@@ -180,7 +175,7 @@ public sealed class ConnectionFactory : ConnectionFactoryBase, IConnectionFactor
         /// </summary>
         /// <remarks>For concurrency greater than one this removes the guarantee that consumers handle messages in the order they receive them.
         /// In addition to that consumers need to be thread/concurrency safe.</remarks>
-        public ushort ConsumerDispatchConcurrency { get; set; } = DefaultConsumerDispatchConcurrency;
+        public ushort ConsumerDispatchConcurrency { get; set; } = Constants.DefaultConsumerDispatchConcurrency;
 
         /// <summary>The host to connect to.</summary>
         public string HostName { get; set; } = "localhost";
diff --git a/projects/RabbitMQ.Client/client/api/IConnection.cs b/projects/RabbitMQ.Client/client/api/IConnection.cs
@@ -239,12 +239,13 @@ Task CloseAsync(ushort reasonCode, string reasonText, TimeSpan timeout, bool abo
         /// will be offloaded to the worker thread pool so it is important to choose the value for the concurrency wisely to avoid thread pool overloading.
         /// <see cref="IAsyncBasicConsumer"/> can handle concurrency much more efficiently due to the non-blocking nature of the consumer.
         ///
-        /// Defaults to <see cref="IConnectionFactory.ConsumerDispatchConcurrency"/>.
+        /// Defaults to <c>null</c>, which will use the value from <see cref="IConnectionFactory.ConsumerDispatchConcurrency"/>
         /// 
         /// For concurrency greater than one this removes the guarantee that consumers handle messages in the order they receive them.
         /// In addition to that consumers need to be thread/concurrency safe.
         /// </param>
         /// <param name="cancellationToken">Cancellation token</param>
-        Task<IChannel> CreateChannelAsync(ushort consumerDispatchConcurrency, CancellationToken cancellationToken = default);
+        Task<IChannel> CreateChannelAsync(ushort? consumerDispatchConcurrency = null,
+            CancellationToken cancellationToken = default);
     }
 }
diff --git a/projects/RabbitMQ.Client/client/api/IConnectionExtensions.cs b/projects/RabbitMQ.Client/client/api/IConnectionExtensions.cs
@@ -7,12 +7,6 @@ namespace RabbitMQ.Client
 {
     public static class IConnectionExtensions
     {
-        /// <summary>
-        /// Asynchronously create and return a fresh channel, session, and channel.
-        /// </summary>
-        public static Task<IChannel> CreateChannelAsync(this IConnection connection, CancellationToken cancellationToken = default) =>
-            connection.CreateChannelAsync(ConnectionFactory.DefaultConsumerDispatchConcurrency, cancellationToken);
-
         /// <summary>
         /// Asynchronously close this connection and all its channels.
         /// </summary>
diff --git a/projects/RabbitMQ.Client/client/framing/Channel.cs b/projects/RabbitMQ.Client/client/framing/Channel.cs
@@ -38,7 +38,7 @@ namespace RabbitMQ.Client.Framing.Impl
 {
     internal class Channel : ChannelBase
     {
-        public Channel(ConnectionConfig config, ISession session, ushort consumerDispatchConcurrency)
+        public Channel(ConnectionConfig config, ISession session, ushort? consumerDispatchConcurrency = null)
             : base(config, session, consumerDispatchConcurrency)
         {
         }
diff --git a/projects/RabbitMQ.Client/client/framing/Constants.cs b/projects/RabbitMQ.Client/client/framing/Constants.cs
@@ -83,5 +83,13 @@ public static class Constants
         public const int NotImplemented = 540;
         ///<summary>(= 541)</summary>
         public const int InternalError = 541;
+
+        /// <summary>
+        /// The default consumer dispatch concurrency. See <see cref="IConnectionFactory.ConsumerDispatchConcurrency"/>
+        /// to set this value for every channel created on a connection,
+        /// and <see cref="IConnection.CreateChannelAsync(ushort?, System.Threading.CancellationToken)"/>
+        /// for setting this value for a particular channel.
+        /// </summary>
+        public const ushort DefaultConsumerDispatchConcurrency = 1;
     }
 }
diff --git a/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs b/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs
@@ -240,12 +240,14 @@ await CloseInnerConnectionAsync()
             }
         }
 
-        public async Task<IChannel> CreateChannelAsync(ushort consumerDispatchConcurrency, CancellationToken cancellationToken = default)
+        public async Task<IChannel> CreateChannelAsync(ushort? consumerDispatchConcurrency = null,
+            CancellationToken cancellationToken = default)
         {
             EnsureIsOpen();
-            RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(consumerDispatchConcurrency, cancellationToken)
+            ushort cdc = consumerDispatchConcurrency.GetValueOrDefault(_config.ConsumerDispatchConcurrency);
+            RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(cdc, cancellationToken)
                 .ConfigureAwait(false);
-            AutorecoveringChannel channel = new AutorecoveringChannel(this, recoveryAwareChannel, consumerDispatchConcurrency);
+            AutorecoveringChannel channel = new AutorecoveringChannel(this, recoveryAwareChannel, cdc);
             await RecordChannelAsync(channel, channelsSemaphoreHeld: false, cancellationToken: cancellationToken)
                 .ConfigureAwait(false);
             return channel;
diff --git a/projects/RabbitMQ.Client/client/impl/ChannelBase.cs b/projects/RabbitMQ.Client/client/impl/ChannelBase.cs
@@ -73,10 +73,12 @@ internal abstract class ChannelBase : IChannel, IRecoverable
 
         internal readonly IConsumerDispatcher ConsumerDispatcher;
 
-        protected ChannelBase(ConnectionConfig config, ISession session, ushort consumerDispatchConcurrency)
+        protected ChannelBase(ConnectionConfig config, ISession session,
+            ushort? perChannelConsumerDispatchConcurrency = null)
         {
             ContinuationTimeout = config.ContinuationTimeout;
-            ConsumerDispatcher = new AsyncConsumerDispatcher(this, consumerDispatchConcurrency);
+            ConsumerDispatcher = new AsyncConsumerDispatcher(this,
+                perChannelConsumerDispatchConcurrency.GetValueOrDefault(config.ConsumerDispatchConcurrency));
             Action<Exception, string> onException = (exception, context) =>
                 OnCallbackException(CallbackExceptionEventArgs.Build(exception, context));
             _basicAcksWrapper = new EventingWrapper<BasicAckEventArgs>("OnBasicAck", onException);
diff --git a/projects/RabbitMQ.Client/client/impl/Connection.cs b/projects/RabbitMQ.Client/client/impl/Connection.cs
@@ -72,7 +72,7 @@ internal Connection(ConnectionConfig config, IFrameHandler frameHandler)
 
             _sessionManager = new SessionManager(this, 0, config.MaxInboundMessageBodySize);
             _session0 = new MainSession(this, config.MaxInboundMessageBodySize);
-            _channel0 = new Channel(_config, _session0, ConnectionFactory.DefaultConsumerDispatchConcurrency); ;
+            _channel0 = new Channel(_config, _session0);
 
             ClientProperties = new Dictionary<string, object?>(_config.ClientProperties)
             {
@@ -253,7 +253,8 @@ await CloseAsync(ea, true,
             }
         }
 
-        public Task<IChannel> CreateChannelAsync(ushort consumerDispatchConcurrency, CancellationToken cancellationToken = default)
+        public Task<IChannel> CreateChannelAsync(ushort? consumerDispatchConcurrency = null,
+            CancellationToken cancellationToken = default)
         {
             EnsureIsOpen();
             ISession session = CreateSession();
diff --git a/projects/RabbitMQ.Client/client/impl/ConsumerDispatching/ConsumerDispatcherChannelBase.cs b/projects/RabbitMQ.Client/client/impl/ConsumerDispatching/ConsumerDispatcherChannelBase.cs
@@ -14,44 +14,42 @@ internal abstract class ConsumerDispatcherChannelBase : ConsumerDispatcherBase,
         protected readonly ChannelReader<WorkStruct> _reader;
         private readonly ChannelWriter<WorkStruct> _writer;
         private readonly Task _worker;
+        private readonly ushort _concurrency;
         private bool _quiesce = false;
         private bool _disposed;
 
         internal ConsumerDispatcherChannelBase(ChannelBase channel, ushort concurrency)
         {
             _channel = channel;
+            _concurrency = concurrency;
             var workChannel = Channel.CreateUnbounded<WorkStruct>(new UnboundedChannelOptions
             {
-                SingleReader = concurrency == 1,
+                SingleReader = _concurrency == 1,
                 SingleWriter = false,
                 AllowSynchronousContinuations = false
             });
             _reader = workChannel.Reader;
             _writer = workChannel.Writer;
 
             Func<Task> loopStart = ProcessChannelAsync;
-            if (concurrency == 1)
+            if (_concurrency == 1)
             {
                 _worker = Task.Run(loopStart);
             }
             else
             {
-                var tasks = new Task[concurrency];
-                for (int i = 0; i < concurrency; i++)
+                var tasks = new Task[_concurrency];
+                for (int i = 0; i < _concurrency; i++)
                 {
                     tasks[i] = Task.Run(loopStart);
                 }
                 _worker = Task.WhenAll(tasks);
             }
         }
 
-        public bool IsShutdown
-        {
-            get
-            {
-                return _quiesce;
-            }
-        }
+        public bool IsShutdown => _quiesce;
+
+        public ushort Concurrency => _concurrency;
 
         public ValueTask HandleBasicConsumeOkAsync(IAsyncBasicConsumer consumer, string consumerTag, CancellationToken cancellationToken)
         {
diff --git a/projects/RabbitMQ.Client/client/impl/ConsumerDispatching/IConsumerDispatcher.cs b/projects/RabbitMQ.Client/client/impl/ConsumerDispatching/IConsumerDispatcher.cs
@@ -41,6 +41,8 @@ internal interface IConsumerDispatcher : IDisposable
 
         bool IsShutdown { get; }
 
+        ushort Concurrency { get; }
+
         IAsyncBasicConsumer GetAndRemoveConsumer(string tag);
 
         ValueTask HandleBasicConsumeOkAsync(IAsyncBasicConsumer consumer, string consumerTag, CancellationToken cancellationToken);
diff --git a/projects/Test/Common/IntegrationFixture.cs b/projects/Test/Common/IntegrationFixture.cs
@@ -71,7 +71,7 @@ public abstract class IntegrationFixture : IAsyncLifetime
         protected readonly ITestOutputHelper _output;
         protected readonly string _testDisplayName;
 
-        protected readonly ushort _consumerDispatchConcurrency = 1;
+        protected readonly ushort _consumerDispatchConcurrency = Constants.DefaultConsumerDispatchConcurrency;
         protected readonly bool _openChannel = true;
 
         public static readonly TimeSpan ShortSpan;
@@ -109,7 +109,7 @@ static IntegrationFixture()
         }
 
         public IntegrationFixture(ITestOutputHelper output,
-            ushort consumerDispatchConcurrency = 1,
+            ushort consumerDispatchConcurrency = Constants.DefaultConsumerDispatchConcurrency,
             bool openChannel = true)
         {
             _consumerDispatchConcurrency = consumerDispatchConcurrency;
@@ -143,8 +143,7 @@ public virtual async Task InitializeAsync()
              */
             if (_connFactory == null)
             {
-                _connFactory = CreateConnectionFactory();
-                _connFactory.ConsumerDispatchConcurrency = _consumerDispatchConcurrency;
+                _connFactory = CreateConnectionFactory(_consumerDispatchConcurrency);
             }
 
             if (_conn == null)
@@ -517,13 +516,15 @@ protected static async Task WaitAsync(TaskCompletionSource<bool> tcs, TimeSpan t
             }
         }
 
-        protected ConnectionFactory CreateConnectionFactory()
+        protected ConnectionFactory CreateConnectionFactory(
+            ushort consumerDispatchConcurrency = Constants.DefaultConsumerDispatchConcurrency)
         {
             return new ConnectionFactory
             {
                 ClientProvidedName = $"{_testDisplayName}:{Util.Now}:{GetConnectionIdx()}",
                 ContinuationTimeout = WaitSpan,
                 HandshakeContinuationTimeout = WaitSpan,
+                ConsumerDispatchConcurrency = consumerDispatchConcurrency
             };
         }
 
diff --git a/projects/Test/Integration/TestAsyncConsumer.cs b/projects/Test/Integration/TestAsyncConsumer.cs
@@ -36,23 +36,28 @@
 using System.Threading.Tasks;
 using RabbitMQ.Client;
 using RabbitMQ.Client.Events;
+using RabbitMQ.Client.Impl;
 using Xunit;
 using Xunit.Abstractions;
 
 namespace Test.Integration
 {
     public class TestAsyncConsumer : IntegrationFixture
     {
+        private const ushort ConsumerDispatchConcurrency = 2;
+
         private readonly ShutdownEventArgs _closeArgs = new ShutdownEventArgs(ShutdownInitiator.Application, Constants.ReplySuccess, "normal shutdown");
 
         public TestAsyncConsumer(ITestOutputHelper output)
-            : base(output, consumerDispatchConcurrency: 2)
+            : base(output, consumerDispatchConcurrency: ConsumerDispatchConcurrency)
         {
         }
 
         [Fact]
         public async Task TestBasicRoundtripConcurrent()
         {
+            await ValidateConsumerDispatchConcurrency();
+
             AddCallbackExceptionHandlers();
             _channel.DefaultConsumer = new DefaultAsyncConsumer(_channel, "_channel,", _output);
 
@@ -146,6 +151,8 @@ public async Task TestBasicRoundtripConcurrent()
         [Fact]
         public async Task TestBasicRoundtripConcurrentManyMessages()
         {
+            await ValidateConsumerDispatchConcurrency();
+
             AddCallbackExceptionHandlers();
             _channel.DefaultConsumer = new DefaultAsyncConsumer(_channel, "_channel,", _output);
 
@@ -323,6 +330,8 @@ public async Task TestBasicRoundtripConcurrentManyMessages()
         [Fact]
         public async Task TestBasicRejectAsync()
         {
+            await ValidateConsumerDispatchConcurrency();
+
             string queueName = GenerateQueueName();
 
             var publishSyncSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -421,6 +430,8 @@ await _channel.BasicConsumeAsync(queue: queueName, autoAck: false,
         [Fact]
         public async Task TestBasicAckAsync()
         {
+            await ValidateConsumerDispatchConcurrency();
+
             string queueName = GenerateQueueName();
 
             const int messageCount = 1024;
@@ -488,6 +499,8 @@ await _channel.BasicConsumeAsync(queue: queueName, autoAck: false,
         [Fact]
         public async Task TestBasicNackAsync()
         {
+            await ValidateConsumerDispatchConcurrency();
+
             var publishSyncSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
 
             _conn.ConnectionShutdown += (o, ea) =>
@@ -561,6 +574,8 @@ await _channel.BasicConsumeAsync(queue: queueName, autoAck: false,
         [Fact]
         public async Task TestDeclarationOfManyAutoDeleteQueuesWithTransientConsumer()
         {
+            await ValidateConsumerDispatchConcurrency();
+
             AssertRecordedQueues((RabbitMQ.Client.Framing.Impl.AutorecoveringConnection)_conn, 0);
             var tasks = new List<Task>();
             for (int i = 0; i < 256; i++)
@@ -581,6 +596,8 @@ public async Task TestDeclarationOfManyAutoDeleteQueuesWithTransientConsumer()
         [Fact]
         public async Task TestCreateChannelWithinAsyncConsumerCallback_GH650()
         {
+            await ValidateConsumerDispatchConcurrency();
+
             string exchangeName = GenerateExchangeName();
             string queue1Name = GenerateQueueName();
             string queue2Name = GenerateQueueName();
@@ -650,6 +667,8 @@ await innerChannel.BasicPublishAsync(exchangeName, queue2Name,
         [Fact]
         public async Task TestCloseWithinEventHandler_GH1567()
         {
+            await ValidateConsumerDispatchConcurrency();
+
             var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
 
             QueueDeclareOk q = await _channel.QueueDeclareAsync();
@@ -679,6 +698,20 @@ await _channel.BasicPublishAsync(exchange: string.Empty, routingKey: queueName,
             Assert.True(await tcs.Task);
         }
 
+        private async Task ValidateConsumerDispatchConcurrency()
+        {
+            ushort expectedConsumerDispatchConcurrency = (ushort)S_Random.Next(3, 10);
+            AutorecoveringChannel autorecoveringChannel = (AutorecoveringChannel)_channel;
+            Assert.Equal(ConsumerDispatchConcurrency, autorecoveringChannel.ConsumerDispatcher.Concurrency);
+            Assert.Equal(_consumerDispatchConcurrency, autorecoveringChannel.ConsumerDispatcher.Concurrency);
+            using (IChannel ch = await _conn.CreateChannelAsync(
+                consumerDispatchConcurrency: expectedConsumerDispatchConcurrency))
+            {
+                AutorecoveringChannel ach = (AutorecoveringChannel)ch;
+                Assert.Equal(expectedConsumerDispatchConcurrency, ach.ConsumerDispatcher.Concurrency);
+            }
+        }
+
         private static void SetException(Exception ex, params TaskCompletionSource<bool>[] tcsAry)
         {
             foreach (TaskCompletionSource<bool> tcs in tcsAry)
diff --git a/projects/Test/Integration/TestAsyncEventingBasicConsumer.cs b/projects/Test/Integration/TestAsyncEventingBasicConsumer.cs
@@ -34,13 +34,16 @@
 using System.Threading.Tasks;
 using RabbitMQ.Client;
 using RabbitMQ.Client.Events;
+using RabbitMQ.Client.Impl;
 using Xunit;
 using Xunit.Abstractions;
 
 namespace Test.Integration
 {
     public class TestAsyncEventingBasicConsumer : IntegrationFixture
     {
+        private const ushort ConsumerDispatchConcurrency = 2;
+
         private readonly CancellationTokenSource _cts = new CancellationTokenSource(ShortSpan);
         private readonly CancellationTokenRegistration _ctr;
         private readonly TaskCompletionSource<bool> _onCallbackExceptionTcs =
@@ -49,7 +52,7 @@ public class TestAsyncEventingBasicConsumer : IntegrationFixture
             new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
 
         public TestAsyncEventingBasicConsumer(ITestOutputHelper output)
-            : base(output, consumerDispatchConcurrency: 2)
+            : base(output, consumerDispatchConcurrency: ConsumerDispatchConcurrency)
         {
             _ctr = _cts.Token.Register(OnTokenCanceled);
         }
@@ -81,6 +84,10 @@ private Task AsyncConsumerOnReceived(object sender, BasicDeliverEventArgs @event
         [Fact]
         public async Task TestAsyncEventingBasicConsumer_GH1038()
         {
+            AutorecoveringChannel autorecoveringChannel = (AutorecoveringChannel)_channel;
+            Assert.Equal(ConsumerDispatchConcurrency, autorecoveringChannel.ConsumerDispatcher.Concurrency);
+            Assert.Equal(_consumerDispatchConcurrency, autorecoveringChannel.ConsumerDispatcher.Concurrency);
+
             string exchangeName = GenerateExchangeName();
             string queueName = GenerateQueueName();
             string routingKey = string.Empty;
diff --git a/projects/Test/Integration/TestConcurrentAccessBase.cs.cs b/projects/Test/Integration/TestConcurrentAccessBase.cs.cs

Original file line number	Diff line number	Diff line change
`@@ -7,12 +7,6 @@ namespace RabbitMQ.Client`
`7`	`7`	`{`
`8`	`8`	`public static class IConnectionExtensions`
`9`	`9`	`{`
`10`		`- /// <summary>`
`11`		`- /// Asynchronously create and return a fresh channel, session, and channel.`
`12`		`- /// </summary>`
`13`		`- public static Task<IChannel> CreateChannelAsync(this IConnection connection, CancellationToken cancellationToken = default) =>`
`14`		`- connection.CreateChannelAsync(ConnectionFactory.DefaultConsumerDispatchConcurrency, cancellationToken);`
`15`		`-`
`16`	`10`	`/// <summary>`
`17`	`11`	`/// Asynchronously close this connection and all its channels.`
`18`	`12`	`/// </summary>`
Original file line number	Diff line number	Diff line change
`@@ -38,7 +38,7 @@ namespace RabbitMQ.Client.Framing.Impl`
`38`	`38`	`{`
`39`	`39`	`internal class Channel : ChannelBase`
`40`	`40`	`{`
`41`		`- public Channel(ConnectionConfig config, ISession session, ushort consumerDispatchConcurrency)`
	`41`	`+ public Channel(ConnectionConfig config, ISession session, ushort? consumerDispatchConcurrency = null)`
`42`	`42`	`: base(config, session, consumerDispatchConcurrency)`
`43`	`43`	`{`
`44`	`44`	`}`
Original file line number	Diff line number	Diff line change
`@@ -83,5 +83,13 @@ public static class Constants`
`83`	`83`	`public const int NotImplemented = 540;`
`84`	`84`	`///<summary>(= 541)</summary>`
`85`	`85`	`public const int InternalError = 541;`
	`86`	`+`
	`87`	`+ /// <summary>`
	`88`	`+ /// The default consumer dispatch concurrency. See <see cref="IConnectionFactory.ConsumerDispatchConcurrency"/>`
	`89`	`+ /// to set this value for every channel created on a connection,`
	`90`	`+ /// and <see cref="IConnection.CreateChannelAsync(ushort?, System.Threading.CancellationToken)"/>`
	`91`	`+ /// for setting this value for a particular channel.`
	`92`	`+ /// </summary>`
	`93`	`+ public const ushort DefaultConsumerDispatchConcurrency = 1;`
`86`	`94`	`}`
`87`	`95`	`}`
Original file line number	Diff line number	Diff line change
`@@ -240,12 +240,14 @@ await CloseInnerConnectionAsync()`
`240`	`240`	`}`
`241`	`241`	`}`
`242`	`242`
`243`		`- public async Task<IChannel> CreateChannelAsync(ushort consumerDispatchConcurrency, CancellationToken cancellationToken = default)`
	`243`	`+ public async Task<IChannel> CreateChannelAsync(ushort? consumerDispatchConcurrency = null,`
	`244`	`+ CancellationToken cancellationToken = default)`
`244`	`245`	`{`
`245`	`246`	`EnsureIsOpen();`
`246`		`- RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(consumerDispatchConcurrency, cancellationToken)`
	`247`	`+ ushort cdc = consumerDispatchConcurrency.GetValueOrDefault(_config.ConsumerDispatchConcurrency);`
	`248`	`+ RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(cdc, cancellationToken)`
`247`	`249`	`.ConfigureAwait(false);`
`248`		`- AutorecoveringChannel channel = new AutorecoveringChannel(this, recoveryAwareChannel, consumerDispatchConcurrency);`
	`250`	`+ AutorecoveringChannel channel = new AutorecoveringChannel(this, recoveryAwareChannel, cdc);`
`249`	`251`	`await RecordChannelAsync(channel, channelsSemaphoreHeld: false, cancellationToken: cancellationToken)`
`250`	`252`	`.ConfigureAwait(false);`
`251`	`253`	`return channel;`
Original file line number	Diff line number	Diff line change
`@@ -72,7 +72,7 @@ internal Connection(ConnectionConfig config, IFrameHandler frameHandler)`
`72`	`72`
`73`	`73`	`_sessionManager = new SessionManager(this, 0, config.MaxInboundMessageBodySize);`
`74`	`74`	`_session0 = new MainSession(this, config.MaxInboundMessageBodySize);`
`75`		`- _channel0 = new Channel(_config, _session0, ConnectionFactory.DefaultConsumerDispatchConcurrency); ;`
	`75`	`+ _channel0 = new Channel(_config, _session0);`
`76`	`76`
`77`	`77`	`ClientProperties = new Dictionary<string, object?>(_config.ClientProperties)`
`78`	`78`	`{`
`@@ -253,7 +253,8 @@ await CloseAsync(ea, true,`
`253`	`253`	`}`
`254`	`254`	`}`
`255`	`255`
`256`		`- public Task<IChannel> CreateChannelAsync(ushort consumerDispatchConcurrency, CancellationToken cancellationToken = default)`
	`256`	`+ public Task<IChannel> CreateChannelAsync(ushort? consumerDispatchConcurrency = null,`
	`257`	`+ CancellationToken cancellationToken = default)`
`257`	`258`	`{`
`258`	`259`	`EnsureIsOpen();`
`259`	`260`	`ISession session = CreateSession();`
Original file line number	Diff line number	Diff line change
`@@ -14,44 +14,42 @@ internal abstract class ConsumerDispatcherChannelBase : ConsumerDispatcherBase,`
`14`	`14`	`protected readonly ChannelReader<WorkStruct> _reader;`
`15`	`15`	`private readonly ChannelWriter<WorkStruct> _writer;`
`16`	`16`	`private readonly Task _worker;`
	`17`	`+ private readonly ushort _concurrency;`
`17`	`18`	`private bool _quiesce = false;`
`18`	`19`	`private bool _disposed;`
`19`	`20`
`20`	`21`	`internal ConsumerDispatcherChannelBase(ChannelBase channel, ushort concurrency)`
`21`	`22`	`{`
`22`	`23`	`_channel = channel;`
	`24`	`+ _concurrency = concurrency;`
`23`	`25`	`var workChannel = Channel.CreateUnbounded<WorkStruct>(new UnboundedChannelOptions`
`24`	`26`	`{`
`25`		`- SingleReader = concurrency == 1,`
	`27`	`+ SingleReader = _concurrency == 1,`
`26`	`28`	`SingleWriter = false,`
`27`	`29`	`AllowSynchronousContinuations = false`
`28`	`30`	`});`
`29`	`31`	`_reader = workChannel.Reader;`
`30`	`32`	`_writer = workChannel.Writer;`
`31`	`33`
`32`	`34`	`Func<Task> loopStart = ProcessChannelAsync;`
`33`		`- if (concurrency == 1)`
	`35`	`+ if (_concurrency == 1)`
`34`	`36`	`{`
`35`	`37`	`_worker = Task.Run(loopStart);`
`36`	`38`	`}`
`37`	`39`	`else`
`38`	`40`	`{`
`39`		`- var tasks = new Task[concurrency];`
`40`		`- for (int i = 0; i < concurrency; i++)`
	`41`	`+ var tasks = new Task[_concurrency];`
	`42`	`+ for (int i = 0; i < _concurrency; i++)`
`41`	`43`	`{`
`42`	`44`	`tasks[i] = Task.Run(loopStart);`
`43`	`45`	`}`
`44`	`46`	`_worker = Task.WhenAll(tasks);`
`45`	`47`	`}`
`46`	`48`	`}`
`47`	`49`
`48`		`- public bool IsShutdown`
`49`		`- {`
`50`		`- get`
`51`		`- {`
`52`		`- return _quiesce;`
`53`		`- }`
`54`		`- }`
	`50`	`+ public bool IsShutdown => _quiesce;`
	`51`	`+`
	`52`	`+ public ushort Concurrency => _concurrency;`
`55`	`53`
`56`	`54`	`public ValueTask HandleBasicConsumeOkAsync(IAsyncBasicConsumer consumer, string consumerTag, CancellationToken cancellationToken)`
`57`	`55`	`{`