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

As @danielmarbach points out, it's kind of silly to have the dispatch concurrency set in the connection factory as PR #1669 makes this value per-channel.

So, we'll remove it from `IConnectionFactory` and use a default parameter value of `1` in `CreateChannelAsync`

Author: lukebakken

projects/Benchmarks/Networking/Networking_BasicDeliver_Commons.cs

@@ -9,9 +9,10 @@ namespace Benchmarks.Networking
     [MemoryDiagnoser]
     public class Networking_BasicDeliver_Commons
     {
-        public static async Task Publish_Hello_World(IConnection connection, uint messageCount, byte[] body)
+        public static async Task Publish_Hello_World(IConnection connection,
+            uint messageCount, byte[] body, ushort consumerDispatchConcurrency = 1)
         {
-            using (IChannel channel = await connection.CreateChannelAsync())
+            using (IChannel channel = await connection.CreateChannelAsync(consumerDispatchConcurrency))
             {
                 QueueDeclareOk queue = await channel.QueueDeclareAsync();
                 var consumer = new CountingConsumer(channel, messageCount);

projects/Benchmarks/Networking/Networking_BasicDeliver_ConnectionChurn.cs

@@ -29,16 +29,17 @@ public void GlobalCleanup()
         [Benchmark(Baseline = true)]
         public async Task Publish_Hello_World()
         {
-            var cf = new ConnectionFactory { ConsumerDispatchConcurrency = 2 };
+            var cf = new ConnectionFactory();
             using (IConnection connection = await cf.CreateConnectionAsync())
             {
-                await Publish_Hello_World(connection);
+                await Publish_Hello_World(connection, 2);
             }
         }
 
-        public static async Task Publish_Hello_World(IConnection connection)
+        public static async Task Publish_Hello_World(IConnection connection, ushort consumerDispatchConcurrency)
         {
-            await Networking_BasicDeliver_Commons.Publish_Hello_World(connection, messageCount, _body);
+            await Networking_BasicDeliver_Commons.Publish_Hello_World(connection, messageCount, _body,
+                consumerDispatchConcurrency);
         }
     }
 }

projects/Benchmarks/Networking/Networking_BasicDeliver_LongLivedConnection.cs

@@ -19,8 +19,7 @@ public class Networking_BasicDeliver_LongLivedConnection
         public void GlobalSetup()
         {
             _container = RabbitMQBroker.Start();
-
-            var cf = new ConnectionFactory { ConsumerDispatchConcurrency = 2 };
+            var cf = new ConnectionFactory();
             // NOTE: https://github.com/dotnet/BenchmarkDotNet/issues/1738
             _connection = EnsureCompleted(cf.CreateConnectionAsync());
         }
@@ -35,7 +34,8 @@ public void GlobalCleanup()
         [Benchmark(Baseline = true)]
         public Task Publish_Hello_World()
         {
-            return Networking_BasicDeliver_Commons.Publish_Hello_World(_connection, messageCount, _body);
+            return Networking_BasicDeliver_Commons.Publish_Hello_World(_connection, messageCount, _body,
+                consumerDispatchConcurrency: 2);
         }
 
         private static T EnsureCompleted<T>(Task<T> task) => task.GetAwaiter().GetResult();

projects/RabbitMQ.Client/PublicAPI.Shipped.txt

@@ -189,8 +189,6 @@ RabbitMQ.Client.ConnectionFactory.ClientProperties.set -> void
 RabbitMQ.Client.ConnectionFactory.ClientProvidedName.get -> string
 RabbitMQ.Client.ConnectionFactory.ClientProvidedName.set -> void
 RabbitMQ.Client.ConnectionFactory.ConnectionFactory() -> void
-RabbitMQ.Client.ConnectionFactory.ConsumerDispatchConcurrency.get -> ushort
-RabbitMQ.Client.ConnectionFactory.ConsumerDispatchConcurrency.set -> void
 RabbitMQ.Client.ConnectionFactory.ContinuationTimeout.get -> System.TimeSpan
 RabbitMQ.Client.ConnectionFactory.ContinuationTimeout.set -> void
 RabbitMQ.Client.ConnectionFactory.Endpoint.get -> RabbitMQ.Client.AmqpTcpEndpoint
@@ -472,8 +470,6 @@ RabbitMQ.Client.IConnectionFactory.ClientProperties.get -> System.Collections.Ge
 RabbitMQ.Client.IConnectionFactory.ClientProperties.set -> void
 RabbitMQ.Client.IConnectionFactory.ClientProvidedName.get -> string
 RabbitMQ.Client.IConnectionFactory.ClientProvidedName.set -> void
-RabbitMQ.Client.IConnectionFactory.ConsumerDispatchConcurrency.get -> ushort
-RabbitMQ.Client.IConnectionFactory.ConsumerDispatchConcurrency.set -> void
 RabbitMQ.Client.IConnectionFactory.ContinuationTimeout.get -> System.TimeSpan
 RabbitMQ.Client.IConnectionFactory.ContinuationTimeout.set -> void
 RabbitMQ.Client.IConnectionFactory.HandshakeContinuationTimeout.get -> System.TimeSpan
@@ -892,7 +888,4 @@ 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!>!
-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!>!

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

@@ -134,13 +134,6 @@ public sealed class ConnectionConfig
         /// </summary>
         public readonly TimeSpan RequestedConnectionTimeout;
 
-        /// <summary>
-        /// Set to a value greater than one to enable concurrent processing. For a concurrency greater than one <see cref="IAsyncBasicConsumer"/>
-        /// 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 ushort ConsumerDispatchConcurrency;
-
         internal readonly Func<AmqpTcpEndpoint, CancellationToken, Task<IFrameHandler>> FrameHandlerFactoryAsync;
 
         internal ConnectionConfig(string virtualHost, string userName, string password,
@@ -150,7 +143,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 consumerDispatchConcurrency, Func<AmqpTcpEndpoint, CancellationToken, Task<IFrameHandler>> frameHandlerFactoryAsync)
+            Func<AmqpTcpEndpoint, CancellationToken, Task<IFrameHandler>> frameHandlerFactoryAsync)
         {
             VirtualHost = virtualHost;
             UserName = userName;
@@ -170,7 +163,6 @@ internal ConnectionConfig(string virtualHost, string userName, string password,
             ContinuationTimeout = continuationTimeout;
             HandshakeContinuationTimeout = handshakeContinuationTimeout;
             RequestedConnectionTimeout = requestedConnectionTimeout;
-            ConsumerDispatchConcurrency = consumerDispatchConcurrency;
             FrameHandlerFactoryAsync = frameHandlerFactoryAsync;
         }
     }

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

@@ -92,10 +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.
@@ -172,16 +168,6 @@ public sealed class ConnectionFactory : ConnectionFactoryBase, IConnectionFactor
         /// </summary>
         public bool AutomaticRecoveryEnabled { get; set; } = true;
 
-        /// <summary>
-        /// Set to a value greater than one to enable concurrent processing. For a concurrency greater than one <see cref="IAsyncBasicConsumer"/>
-        /// 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 1.
-        /// </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;
-
         /// <summary>The host to connect to.</summary>
         public string HostName { get; set; } = "localhost";
 
@@ -603,7 +589,6 @@ private ConnectionConfig CreateConfig(string? clientProvidedName)
                 ContinuationTimeout,
                 HandshakeContinuationTimeout,
                 RequestedConnectionTimeout,
-                ConsumerDispatchConcurrency,
                 CreateFrameHandlerAsync);
         }
 

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"/>.
+        /// The default value is 1.
         /// 
         /// 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);
     }
 }

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>

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

@@ -185,15 +185,5 @@ Task<IConnection> CreateConnectionAsync(IEnumerable<AmqpTcpEndpoint> endpoints,
         /// timing out.
         /// </summary>
         TimeSpan ContinuationTimeout { get; set; }
-
-        /// <summary>
-        /// Set to a value greater than one to enable concurrent processing. For a concurrency greater than one <see cref="IAsyncBasicConsumer"/>
-        /// 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 1.
-        /// </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>
-        ushort ConsumerDispatchConcurrency { get; set; }
     }
 }

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)
         {
         }