* Standardize on ConsumerDispatchConcurrency name

Author: lukebakken

projects/Benchmarks/ConsumerDispatching/ConsumerDispatcher.cs

@@ -35,7 +35,7 @@ public class BasicDeliverConsumerDispatching : ConsumerDispatcherBase
         public int Count { get; set; }
 
         [Params(1, 2)]
-        public int Concurrency { get; set; }
+        public ushort Concurrency { get; set; }
 
         [GlobalSetup(Target = nameof(AsyncConsumerDispatcher))]
         public async Task SetUpAsyncConsumer()

projects/RabbitMQ.Client/PublicAPI.Shipped.txt

@@ -700,7 +700,6 @@ readonly RabbitMQ.Client.ConnectionConfig.AuthMechanisms -> System.Collections.G
 readonly RabbitMQ.Client.ConnectionConfig.ClientProperties -> System.Collections.Generic.IDictionary<string, object>
 readonly RabbitMQ.Client.ConnectionConfig.ClientProvidedName -> string
 readonly RabbitMQ.Client.ConnectionConfig.ContinuationTimeout -> System.TimeSpan
-readonly RabbitMQ.Client.ConnectionConfig.DispatchConsumerConcurrency -> int
 readonly RabbitMQ.Client.ConnectionConfig.HandshakeContinuationTimeout -> System.TimeSpan
 readonly RabbitMQ.Client.ConnectionConfig.HeartbeatInterval -> System.TimeSpan
 readonly RabbitMQ.Client.ConnectionConfig.MaxChannelCount -> ushort
@@ -896,3 +895,4 @@ RabbitMQ.Client.IChannel.ConfirmSelectAsync(bool trackConfirmations = true, Syst
 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!>!
+readonly RabbitMQ.Client.ConnectionConfig.ConsumerDispatchConcurrency -> ushort

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

@@ -139,7 +139,7 @@ public sealed class ConnectionConfig
         /// 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.
         /// </summary>
-        public readonly int DispatchConsumerConcurrency;
+        public readonly ushort ConsumerDispatchConcurrency;
 
         internal readonly Func<AmqpTcpEndpoint, CancellationToken, Task<IFrameHandler>> FrameHandlerFactoryAsync;
 
@@ -150,7 +150,7 @@ internal ConnectionConfig(string virtualHost, string userName, string password,
             ushort maxChannelCount, uint maxFrameSize, uint maxInboundMessageBodySize, bool topologyRecoveryEnabled,
             TopologyRecoveryFilter topologyRecoveryFilter, TopologyRecoveryExceptionHandler topologyRecoveryExceptionHandler,
             TimeSpan networkRecoveryInterval, TimeSpan heartbeatInterval, TimeSpan continuationTimeout, TimeSpan handshakeContinuationTimeout, TimeSpan requestedConnectionTimeout,
-            ushort dispatchConsumerConcurrency, Func<AmqpTcpEndpoint, CancellationToken, Task<IFrameHandler>> frameHandlerFactoryAsync)
+            ushort consumerDispatchConcurrency, Func<AmqpTcpEndpoint, CancellationToken, Task<IFrameHandler>> frameHandlerFactoryAsync)
         {
             VirtualHost = virtualHost;
             UserName = userName;
@@ -170,7 +170,7 @@ internal ConnectionConfig(string virtualHost, string userName, string password,
             ContinuationTimeout = continuationTimeout;
             HandshakeContinuationTimeout = handshakeContinuationTimeout;
             RequestedConnectionTimeout = requestedConnectionTimeout;
-            DispatchConsumerConcurrency = dispatchConsumerConcurrency;
+            ConsumerDispatchConcurrency = consumerDispatchConcurrency;
             FrameHandlerFactoryAsync = frameHandlerFactoryAsync;
         }
     }

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, int? dispatchConsumerConcurrency = null) : base(config, session, dispatchConsumerConcurrency)
+        public Channel(ConnectionConfig config, ISession session, ushort consumerDispatchConcurrency)
+            : base(config, session, consumerDispatchConcurrency)
         {
         }
 

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

@@ -52,6 +52,7 @@ internal sealed class AutorecoveringChannel : IChannel, IRecoverable
         private bool _usesPublisherConfirms;
         private bool _tracksPublisherConfirmations;
         private bool _usesTransactions;
+        private ushort _consumerDispatchConcurrency;
 
         internal IConsumerDispatcher ConsumerDispatcher => InnerChannel.ConsumerDispatcher;
 
@@ -70,10 +71,12 @@ public TimeSpan ContinuationTimeout
             set => InnerChannel.ContinuationTimeout = value;
         }
 
-        public AutorecoveringChannel(AutorecoveringConnection conn, RecoveryAwareChannel innerChannel)
+        public AutorecoveringChannel(AutorecoveringConnection conn, RecoveryAwareChannel innerChannel,
+            ushort consumerDispatchConcurrency)
         {
             _connection = conn;
             _innerChannel = innerChannel;
+            _consumerDispatchConcurrency = consumerDispatchConcurrency;
         }
 
         public event EventHandler<BasicAckEventArgs> BasicAcks
@@ -160,7 +163,8 @@ internal async Task<bool> AutomaticallyRecoverAsync(AutorecoveringConnection con
 
             _connection = conn;
 
-            RecoveryAwareChannel newChannel = await conn.CreateNonRecoveringChannelAsync(cancellationToken: cancellationToken)
+            RecoveryAwareChannel newChannel = await conn.CreateNonRecoveringChannelAsync(_consumerDispatchConcurrency,
+                cancellationToken: cancellationToken)
                 .ConfigureAwait(false);
             newChannel.TakeOver(_innerChannel);
 

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

@@ -173,10 +173,11 @@ public event EventHandler<RecoveringConsumerEventArgs> RecoveringConsumer
 
         public IProtocol Protocol => Endpoint.Protocol;
 
-        public async ValueTask<RecoveryAwareChannel> CreateNonRecoveringChannelAsync(int? dispatchConsumerConcurrency = null, CancellationToken cancellationToken = default)
+        public async ValueTask<RecoveryAwareChannel> CreateNonRecoveringChannelAsync(ushort consumerDispatchConcurrency,
+            CancellationToken cancellationToken = default)
         {
             ISession session = InnerConnection.CreateSession();
-            var result = new RecoveryAwareChannel(_config, session, dispatchConsumerConcurrency);
+            var result = new RecoveryAwareChannel(_config, session, consumerDispatchConcurrency);
             return (RecoveryAwareChannel)await result.OpenAsync(cancellationToken).ConfigureAwait(false);
         }
 
@@ -239,12 +240,12 @@ await CloseInnerConnectionAsync()
             }
         }
 
-        public async Task<IChannel> CreateChannelAsync(ushort dispatchConsumerConcurrency, CancellationToken cancellationToken = default)
+        public async Task<IChannel> CreateChannelAsync(ushort consumerDispatchConcurrency, CancellationToken cancellationToken = default)
         {
             EnsureIsOpen();
-            RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(dispatchConsumerConcurrency, cancellationToken)
+            RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(consumerDispatchConcurrency, cancellationToken)
                 .ConfigureAwait(false);
-            AutorecoveringChannel channel = new AutorecoveringChannel(this, recoveryAwareChannel);
+            AutorecoveringChannel channel = new AutorecoveringChannel(this, recoveryAwareChannel, consumerDispatchConcurrency);
             await RecordChannelAsync(channel, channelsSemaphoreHeld: false, cancellationToken: cancellationToken)
                 .ConfigureAwait(false);
             return channel;

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

@@ -73,10 +73,10 @@ internal abstract class ChannelBase : IChannel, IRecoverable
 
         internal readonly IConsumerDispatcher ConsumerDispatcher;
 
-        protected ChannelBase(ConnectionConfig config, ISession session, int? dispatchConsumerConcurrency = null)
+        protected ChannelBase(ConnectionConfig config, ISession session, ushort consumerDispatchConcurrency)
         {
             ContinuationTimeout = config.ContinuationTimeout;
-            ConsumerDispatcher = new AsyncConsumerDispatcher(this, dispatchConsumerConcurrency.GetValueOrDefault(config.DispatchConsumerConcurrency));
+            ConsumerDispatcher = new AsyncConsumerDispatcher(this, consumerDispatchConcurrency);
             Action<Exception, string> onException = (exception, context) =>
                 OnCallbackException(CallbackExceptionEventArgs.Build(exception, context));
             _basicAcksWrapper = new EventingWrapper<BasicAckEventArgs>("OnBasicAck", onException);

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); ;
+            _channel0 = new Channel(_config, _session0, ConnectionFactory.DefaultConsumerDispatchConcurrency); ;
 
             ClientProperties = new Dictionary<string, object?>(_config.ClientProperties)
             {
@@ -253,11 +253,11 @@ await CloseAsync(ea, true,
             }
         }
 
-        public Task<IChannel> CreateChannelAsync(ushort dispatchConsumerConcurrency, CancellationToken cancellationToken = default)
+        public Task<IChannel> CreateChannelAsync(ushort consumerDispatchConcurrency, CancellationToken cancellationToken = default)
         {
             EnsureIsOpen();
             ISession session = CreateSession();
-            var channel = new Channel(_config, session, dispatchConsumerConcurrency);
+            var channel = new Channel(_config, session, consumerDispatchConcurrency);
             return channel.OpenAsync(cancellationToken);
         }
 

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

@@ -8,7 +8,7 @@ namespace RabbitMQ.Client.ConsumerDispatching
 {
     internal sealed class AsyncConsumerDispatcher : ConsumerDispatcherChannelBase
     {
-        internal AsyncConsumerDispatcher(ChannelBase channel, int concurrency)
+        internal AsyncConsumerDispatcher(ChannelBase channel, ushort concurrency)
             : base(channel, concurrency)
         {
         }

projects/RabbitMQ.Client/client/impl/ConsumerDispatching/ConsumerDispatcherChannelBase.cs

@@ -17,7 +17,7 @@ internal abstract class ConsumerDispatcherChannelBase : ConsumerDispatcherBase,
         private bool _quiesce = false;
         private bool _disposed;
 
-        internal ConsumerDispatcherChannelBase(ChannelBase channel, int concurrency)
+        internal ConsumerDispatcherChannelBase(ChannelBase channel, ushort concurrency)
         {
             _channel = channel;
             var workChannel = Channel.CreateUnbounded<WorkStruct>(new UnboundedChannelOptions