Follow-up to #1669 - per-channel dispatch concurrency

lukebakken · lukebakken · commit 7770fd863a46 · 2024-09-12T06:57:55.000-07:00
PR #1669 by @danielmarbach adds the ability to configure consumer dispatch on a per-channel basis. * Test that consumer dispatch concurrency is set on the dispatcher.
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 = 1, 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 <see cref="Constants.DefaultConsumerDispatchConcurrency"/>.
         /// 
         /// 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 = Constants.DefaultConsumerDispatchConcurrency,
+            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,8 @@ 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 = Constants.DefaultConsumerDispatchConcurrency)
             : 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
@@ -37,6 +37,7 @@
 using RabbitMQ.Client.Events;
 using RabbitMQ.Client.Exceptions;
 using RabbitMQ.Client.Impl;
+using RabbitMQ.Util;
 
 namespace RabbitMQ.Client.Framing.Impl
 {
@@ -240,12 +241,14 @@ await CloseInnerConnectionAsync()
             }
         }
 
-        public async Task<IChannel> CreateChannelAsync(ushort consumerDispatchConcurrency, CancellationToken cancellationToken = default)
+        public async Task<IChannel> CreateChannelAsync(ushort consumerDispatchConcurrency = Constants.DefaultConsumerDispatchConcurrency,
+            CancellationToken cancellationToken = default)
         {
             EnsureIsOpen();
-            RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(consumerDispatchConcurrency, cancellationToken)
+            ushort cdc = Misc.DetermineConsumerDispatchConcurrency(_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
@@ -45,6 +45,7 @@
 using RabbitMQ.Client.Events;
 using RabbitMQ.Client.Exceptions;
 using RabbitMQ.Client.Framing.Impl;
+using RabbitMQ.Util;
 
 namespace RabbitMQ.Client.Impl
 {
@@ -73,10 +74,11 @@ 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 consumerDispatchConcurrency = Constants.DefaultConsumerDispatchConcurrency)
         {
             ContinuationTimeout = config.ContinuationTimeout;
-            ConsumerDispatcher = new AsyncConsumerDispatcher(this, consumerDispatchConcurrency);
+            ConsumerDispatcher = BuildConsumerDispatcher(config, consumerDispatchConcurrency);
             Action<Exception, string> onException = (exception, context) =>
                 OnCallbackException(CallbackExceptionEventArgs.Build(exception, context));
             _basicAcksWrapper = new EventingWrapper<BasicAckEventArgs>("OnBasicAck", onException);
@@ -92,6 +94,12 @@ protected ChannelBase(ConnectionConfig config, ISession session, ushort consumer
             Session = session;
         }
 
+        private IConsumerDispatcher BuildConsumerDispatcher(ConnectionConfig config, ushort perChannelConsumerDispatchConcurrency)
+        {
+            ushort cdc = Misc.DetermineConsumerDispatchConcurrency(config, perChannelConsumerDispatchConcurrency);
+            return new AsyncConsumerDispatcher(this, cdc);
+        }
+
         internal TimeSpan HandshakeContinuationTimeout { get; set; } = TimeSpan.FromSeconds(10);
         public TimeSpan ContinuationTimeout { get; set; }
 
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 = Constants.DefaultConsumerDispatchConcurrency,
+            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/RabbitMQ.Client/util/Misc.cs b/projects/RabbitMQ.Client/util/Misc.cs
@@ -0,0 +1,48 @@
+// This source code is dual-licensed under the Apache License, version
+// 2.0, and the Mozilla Public License, version 2.0.
+//
+// The APL v2.0:
+//
+//---------------------------------------------------------------------------
+//   Copyright (c) 2007-2024 Broadcom. All Rights Reserved.
+//
+//   Licensed under the Apache License, Version 2.0 (the "License");
+//   you may not use this file except in compliance with the License.
+//   You may obtain a copy of the License at
+//
+//       https://www.apache.org/licenses/LICENSE-2.0
+//
+//   Unless required by applicable law or agreed to in writing, software
+//   distributed under the License is distributed on an "AS IS" BASIS,
+//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//   See the License for the specific language governing permissions and
+//   limitations under the License.
+//---------------------------------------------------------------------------
+//
+// The MPL v2.0:
+//
+//---------------------------------------------------------------------------
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at https://mozilla.org/MPL/2.0/.
+//
+//  Copyright (c) 2007-2024 Broadcom. All Rights Reserved.
+//---------------------------------------------------------------------------
+
+using RabbitMQ.Client;
+
+namespace RabbitMQ.Util
+{
+    internal static class Misc
+    {
+        internal static ushort DetermineConsumerDispatchConcurrency(ConnectionConfig config, ushort perChannelConsumerDispatchConcurrency)
+        {
+            ushort cdc = config.ConsumerDispatchConcurrency;
+            if (perChannelConsumerDispatchConcurrency > Constants.DefaultConsumerDispatchConcurrency)
+            {
+                cdc = perChannelConsumerDispatchConcurrency;
+            }
+            return cdc;
+        }
+    }
+}
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
diff --git a/projects/Test/Integration/TestAsyncEventingBasicConsumer.cs b/projects/Test/Integration/TestAsyncEventingBasicConsumer.cs
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,8 @@ 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,`
	`42`	`+ ushort consumerDispatchConcurrency = Constants.DefaultConsumerDispatchConcurrency)`
`42`	`43`	`: base(config, session, consumerDispatchConcurrency)`
`43`	`44`	`{`
`44`	`45`	`}`
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
`@@ -37,6 +37,7 @@`
`37`	`37`	`using RabbitMQ.Client.Events;`
`38`	`38`	`using RabbitMQ.Client.Exceptions;`
`39`	`39`	`using RabbitMQ.Client.Impl;`
	`40`	`+using RabbitMQ.Util;`
`40`	`41`
`41`	`42`	`namespace RabbitMQ.Client.Framing.Impl`
`42`	`43`	`{`
`@@ -240,12 +241,14 @@ await CloseInnerConnectionAsync()`
`240`	`241`	`}`
`241`	`242`	`}`
`242`	`243`
`243`		`- public async Task<IChannel> CreateChannelAsync(ushort consumerDispatchConcurrency, CancellationToken cancellationToken = default)`
	`244`	`+ public async Task<IChannel> CreateChannelAsync(ushort consumerDispatchConcurrency = Constants.DefaultConsumerDispatchConcurrency,`
	`245`	`+ CancellationToken cancellationToken = default)`
`244`	`246`	`{`
`245`	`247`	`EnsureIsOpen();`
`246`		`- RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(consumerDispatchConcurrency, cancellationToken)`
	`248`	`+ ushort cdc = Misc.DetermineConsumerDispatchConcurrency(_config, consumerDispatchConcurrency);`
	`249`	`+ RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(cdc, cancellationToken)`
`247`	`250`	`.ConfigureAwait(false);`
`248`		`- AutorecoveringChannel channel = new AutorecoveringChannel(this, recoveryAwareChannel, consumerDispatchConcurrency);`
	`251`	`+ AutorecoveringChannel channel = new AutorecoveringChannel(this, recoveryAwareChannel, cdc);`
`249`	`252`	`await RecordChannelAsync(channel, channelsSemaphoreHeld: false, cancellationToken: cancellationToken)`
`250`	`253`	`.ConfigureAwait(false);`
`251`	`254`	`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 = Constants.DefaultConsumerDispatchConcurrency,`
	`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`	`{`