Fix consumer dispatch concurrency

src/Hosting/Queue/src/BaseQueueHostedService.cs

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

src/Hosting/Queue/src/BaseQueueHostedServiceOptions.cs

@@ -57,4 +57,9 @@ 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/QueueHostedServiceOptions.cs

@@ -5,8 +5,4 @@ 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/IdGenerators/IdGen/src/ClickView.GoodStuff.IdGenerators.IdGen.csproj

@@ -11,7 +11,7 @@
 
     <ItemGroup>
       <PackageReference Include="IdGen" Version="3.0.7" />
-      <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="9.0.4" />
+      <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="9.0.6" />
     </ItemGroup>
 
 </Project>

src/IdGenerators/Serialization/SystemTextJson/src/ClickView.GoodStuff.IdGenerators.Serialization.SystemTextJson.csproj

@@ -6,7 +6,7 @@
     </PropertyGroup>
 
     <ItemGroup>
-      <PackageReference Include="System.Text.Json" Version="9.0.4" />
+      <PackageReference Include="System.Text.Json" Version="9.0.6" />
     </ItemGroup>
 
     <ItemGroup>

src/Queues/RabbitMq/src/RabbitMqClient.cs

@@ -46,7 +46,9 @@ public async Task EnqueueAsync<TData>(string exchange, TData data, EnqueueOption
         var message = MessageWrapper<TData>.New(data);
         var bytes = _options.Serializer.Serialize(message);
 
-        await using var channel = await GetChannelAsync(options.EnablePublisherConfirms, cancellationToken);
+        await using var channel = await GetChannelAsync(
+            options.EnablePublisherConfirms,
+            cancellationToken : cancellationToken);
 
         var properties = new BasicProperties
         {
@@ -79,7 +81,7 @@ 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, cancellationToken);
+        var channel = await GetChannelAsync(false, options.PrefetchCount, cancellationToken);
 
         try
         {
@@ -190,13 +192,16 @@ private async ValueTask<IConnection> ConnectSlowAsync(CancellationToken cancella
         }
     }
 
-    private async Task<IChannel> GetChannelAsync(bool enablePublisherConfirms, CancellationToken cancellationToken)
+    private async Task<IChannel> GetChannelAsync(bool enablePublisherConfirms,
+        ushort consumerDispatchConcurrency = 1,
+        CancellationToken cancellationToken = default)
     {
         var connection = await GetConnectionAsync(cancellationToken);
 
         var options = new CreateChannelOptions(
             publisherConfirmationsEnabled: enablePublisherConfirms,
-            publisherConfirmationTrackingEnabled: enablePublisherConfirms
+            publisherConfirmationTrackingEnabled: enablePublisherConfirms,
+            consumerDispatchConcurrency : consumerDispatchConcurrency
         );
 
         return await connection.CreateChannelAsync(options, cancellationToken);

src/Repositories/MySql/src/ClickView.GoodStuff.Repositories.MySql.csproj

@@ -7,7 +7,7 @@
 
   <ItemGroup>
     <PackageReference Include="Dapper" Version="2.1.66" />
-    <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="9.0.4" />
+    <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="9.0.6" />
     <PackageReference Include="MySqlConnector" Version="2.4.0" />
     <PackageReference Include="Polly" Version="8.6.0" />
   </ItemGroup>