Fix concurrent queue item handling (re #294)

Revert changes from #294 and instead pass ConsumerDispatchConcurrency in SubscribeOptions. Use this to control concurrency and not the prefetch count otherwise that will create way too many tasks in the batch handler

Author: MrSmoke

src/Hosting/Queue/src/BaseQueueHostedService.cs

@@ -192,8 +192,7 @@ private static RabbitMqClientOptions CreateOptions(BaseQueueHostedServiceOptions
             ConnectionTimeout = options.ConnectionTimeout,
             EnableSsl = options.EnableSsl,
             IgnoreSslErrors = options.IgnoreSslErrors,
-            LoggerFactory = loggerFactory,
-            ConsumerDispatchConcurrency = options.ConcurrentTaskCount
+            LoggerFactory = loggerFactory
         };
 
         if (options.SslVersion != null)

src/Hosting/Queue/src/BaseQueueHostedServiceOptions.cs

@@ -57,9 +57,4 @@ public abstract class BaseQueueHostedServiceOptions
     /// The serializer to use for deserializing messages from the Queue
     /// </summary>
     public IMessageSerializer? Serializer { get; set; }
-
-    /// <summary>
-    /// The number of tasks to run concurrently.
-    /// </summary>
-    public ushort ConcurrentTaskCount { get; set; } = 1;
 }

src/Hosting/Queue/src/BatchQueueHostedService.cs

@@ -49,6 +49,8 @@ protected override async Task<SubscriptionContext> SubscribeAsync(IQueueClient q
         var subscribeOptions = new SubscribeOptions
         {
             PrefetchCount = Options.BatchSize,
+            // We only want to run 1 thread when processing batches
+            ConsumerDispatchConcurrency = 1,
             AutoAcknowledge = false
         };
 
@@ -212,7 +214,7 @@ private async Task IntervalWorkerAsync(CancellationToken cancellationToken)
                     }
                     else
                     {
-                        snapshot = Array.Empty<TMessage>();
+                        snapshot = [];
                         latestDeliveryTag = 0;
                     }
                 }

src/Hosting/Queue/src/QueueHostedService.cs

@@ -28,7 +28,11 @@ protected override Task<SubscriptionContext> SubscribeAsync(IQueueClient queueCl
     {
         return queueClient.SubscribeAsync<TMessage>(queueName,
             OnMessageInternalAsync,
-            new SubscribeOptions {PrefetchCount = Options.ConcurrentTaskCount},
+            new SubscribeOptions
+            {
+                PrefetchCount = Options.ConcurrentTaskCount,
+                ConsumerDispatchConcurrency = Options.ConcurrentTaskCount
+            },
             cancellationToken);
     }
 

src/Hosting/Queue/src/QueueHostedServiceOptions.cs

@@ -5,4 +5,8 @@ namespace ClickView.GoodStuff.Hosting.Queue;
 /// </summary>
 public abstract class QueueHostedServiceOptions : BaseQueueHostedServiceOptions
 {
+    /// <summary>
+    /// The number of tasks to run concurrently.
+    /// </summary>
+    public ushort ConcurrentTaskCount { get; set; } = 1;
 }

src/Queues/RabbitMq/src/RabbitMqClient.cs

@@ -48,6 +48,7 @@ public async Task EnqueueAsync<TData>(string exchange, TData data, EnqueueOption
 
         await using var channel = await GetChannelAsync(
             options.EnablePublisherConfirms,
+            consumerDispatchConcurrency: null,
             cancellationToken : cancellationToken);
 
         var properties = new BasicProperties
@@ -81,7 +82,10 @@ public async Task<SubscriptionContext> SubscribeAsync<TData>(string queue,
 
         // We don't want to dispose the channel here (unless an exception is thrown, see below).
         // The returned SubscriptionContext is the object that should be disposed (which disposes the channel)
-        var channel = await GetChannelAsync(false, options.PrefetchCount, cancellationToken);
+        var channel = await GetChannelAsync(
+            enablePublisherConfirms: false,
+            consumerDispatchConcurrency: options.ConsumerDispatchConcurrency,
+            cancellationToken: cancellationToken);
 
         try
         {
@@ -193,15 +197,15 @@ private async ValueTask<IConnection> ConnectSlowAsync(CancellationToken cancella
     }
 
     private async Task<IChannel> GetChannelAsync(bool enablePublisherConfirms,
-        ushort consumerDispatchConcurrency = 1,
+        ushort? consumerDispatchConcurrency,
         CancellationToken cancellationToken = default)
     {
         var connection = await GetConnectionAsync(cancellationToken);
 
         var options = new CreateChannelOptions(
             publisherConfirmationsEnabled: enablePublisherConfirms,
             publisherConfirmationTrackingEnabled: enablePublisherConfirms,
-            consumerDispatchConcurrency : consumerDispatchConcurrency
+            consumerDispatchConcurrency: consumerDispatchConcurrency
         );
 
         return await connection.CreateChannelAsync(options, cancellationToken);

src/Queues/RabbitMq/src/SubscribeOptions.cs

@@ -5,5 +5,14 @@ public class SubscribeOptions
     public bool AutoAcknowledge { get; init; }
     public ushort PrefetchCount { get; init; } = 1;
 
+    /// <summary>
+    /// Set to a value greater than one to enable concurrent processing. For a concurrency greater than one,
+    /// tasks 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.
+    /// If set to null, the value from <see cref="RabbitMqClientOptions"/> will be used.
+    /// </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; }
+
     internal static readonly SubscribeOptions Default = new();
 }