SW-Bus 10.1: RabbitMQ.Client 7, per-queue concurrency, publisher confirms, drain

.github/workflows/nuget-publish.yml

@@ -23,7 +23,7 @@ jobs:
         SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj
         SW.Bus.RabbitMqViewer/SW.Bus.RabbitMqViewer.csproj
       major-version: '10'
-      minor-version: '0'
+      minor-version: '1'
       dotnet-version: '10.0.x'
       run-tests: false
       test-projects: '**/*UnitTests/*.csproj'

README.md

@@ -39,6 +39,8 @@ The library ships as three complementary NuGet packages:
 17. [Architecture](#architecture)
 18. [Testing](#testing)
 19. [Dependencies](#dependencies)
+20. [Concurrency, Publisher Confirms and Shutdown](#concurrency-publisher-confirms-and-shutdown)
+21. [Upgrading to 10.1](#upgrading-to-101)
 
 ## Features
 
@@ -47,6 +49,9 @@ The library ships as three complementary NuGet packages:
 - 🔄 **Automatic Retries** — configurable per-queue retry counts and delay with dead-letter routing
 - 🎯 **Typed Consumers** — `IConsume<T>` with strongly-typed message deserialization
 - ⚙️ **Extended Consumer Options** — per-consumer prefetch and priority via `IConsumeExtended`
+- 🧵 **Parallel Consumers**: per-queue concurrency, how many messages of a queue run at once on one node (default 1, in order)
+- ✅ **Publisher Confirms**: a publish completes when the broker has the message, and throws when it does not
+- 🛑 **Graceful Shutdown**: messages being processed finish and are acknowledged before the host stops
 - 🔐 **JWT Propagation** — user context forwarded through message headers across services
 - 📊 **Queue Monitoring** — live queue depth, rates, and consumer counts via `IConsumerReader`
 - 🔍 **Error Queue Inspection** — peek retry and dead-letter queues via `IErrorQueueReader`
@@ -262,6 +267,12 @@ services.AddBus(config =>
     config.AddQueueOption(
         "PaymentConsumer.PaymentProcessed",
         priority: 10);
+
+    // Process up to 8 messages of this queue at once on each node
+    config.AddQueueOption(
+        "ReportConsumer.ReportRequested",
+        prefetch: 16,
+        concurrency: 8);
 });
 ```
 
@@ -292,7 +303,7 @@ public class MultiConsumer : IConsumeExtended
             new Dictionary<string, ConsumerOptions>
             {
                 ["TypeA"] = new ConsumerOptions { Prefetch = 4 },
-                ["TypeB"] = new ConsumerOptions { Prefetch = 16, Priority = 3 }
+                ["TypeB"] = new ConsumerOptions { Prefetch = 16, Priority = 3, Concurrency = 8 }
             });
 
     public async Task Process(string typeName, string message) { ... }
@@ -301,6 +312,8 @@ public class MultiConsumer : IConsumeExtended
 
 `IConsumeExtended` consumers are **hot-reloadable** — call `IBroadcast.RefreshConsumers()` to apply updated options across all running instances without restart.
 
+A change of `Concurrency` moves the consumer to a new channel, because a channel's dispatch concurrency is fixed when it is created. The old channel finishes the messages it is processing and closes; messages it had prefetched but not started return to the queue and are taken by the new channel. No message is processed twice.
+
 ---
 
 ## Monitoring — `IConsumerReader`
@@ -878,14 +891,19 @@ services.AddBus(config =>
     config.VirtualHost        = "/";
 
     // ── Queue defaults ────────────────────────────────────────────────────
-    config.DefaultQueuePrefetch = 4;      // QoS prefetch per consumer
-    config.DefaultRetryCount    = 5;      // retries before dead-letter
-    config.DefaultRetryAfter    = 60;     // seconds between retries
-    config.DefaultMaxPriority   = 0;      // 0 = priority queues disabled
+    config.DefaultQueuePrefetch       = 4;   // QoS prefetch per consumer
+    config.DefaultConsumerConcurrency = 1;   // messages of one queue processed at once per node
+    config.DefaultRetryCount          = 5;   // retries before dead-letter
+    config.DefaultRetryAfter          = 60;  // seconds between retries
+    config.DefaultMaxPriority         = 0;   // 0 = priority queues disabled
 
     // ── Per-queue overrides ───────────────────────────────────────────────
     config.AddQueueOption("OrderConsumer.OrderCreated",
-        prefetch: 10, retryCount: 3, retryAfterSeconds: 30, priority: 5);
+        prefetch: 10, retryCount: 3, retryAfterSeconds: 30, priority: 5, concurrency: 4);
+
+    // ── Publishing and shutdown ───────────────────────────────────────────
+    config.PublisherConfirms = true;                     // a publish waits for the broker's confirm
+    config.DrainTimeout      = TimeSpan.FromSeconds(30); // wait for running messages on stop
 
     // ── Broadcast listeners ───────────────────────────────────────────────
     config.ListenRetryCount = 5;
@@ -1002,28 +1020,105 @@ dotnet test SW.Bus.IntegrationTests/SW.Bus.IntegrationTests.csproj
 
 **`ConsumerRefreshTests`** — calls `IBroadcast.RefreshConsumers` after a consumer's message type disappears, and asserts the consumer stops reading from that queue while the queue itself is left in place.
 
+**`ConcurrencyTests`**: a queue runs as many messages at once as its concurrency; at 1 it runs them one at a time in order; concurrency comes from the consumer options, then `AddQueueOption`, then `DefaultConsumerConcurrency`, and prefetch caps it; a refresh moves a consumer to a new concurrency without processing a message twice; a refresh applies a new prefetch; concurrent publishes from one scope and from many scopes all arrive.
+
+**`ConnectionTests`**: a publish against a stopped broker throws, and publishing resumes when it returns; a broker restart leaves exactly one consumer per queue (read through `IConsumerReader`) and on the node queue; a host whose broker is down at boot starts, reports not connected, and connects when the broker returns.
+
+**`DrainAndRetryTests`**: stopping lets running messages finish and acknowledge, and returns the ones not started; a message still running at the drain timeout returns to its queue; a failing message is retried its retry count and then moved to the bad queue with the same headers as before.
+
+**`PublishedValuesTests`**: values published beside the body reach the consumer as bus values.
+
 #### Structure
 
 | File | Role |
 |---|---|
 | `BusHarness` | Boots a real `IHost` against a container's connection string; waits for `ConsumersService` to declare topology |
+| `Broker` | Container helpers: stop and start the broker application, read a queue's counts and consumers |
 | `DynamicConsumer` | `IConsume` whose message types can change at runtime, like an application's database-driven consumers |
-| `ConsumerRefreshTests` | The refresh scenario above |
+| `WorkConsumer` | `IConsumeExtended` whose types, options and handlers a test sets, recording how many ran at once and in what order |
+| `ConsumerRefreshTests` and the classes above | The scenarios above |
 
 ---
 
 ## Dependencies
 
 | Package | Version | Used for |
 |---|---|---|
-| `RabbitMQ.Client` | 6.8.1 | AMQP connection and channel management |
+| `RabbitMQ.Client` | 7.2.2 | AMQP connection and channel management |
 | `EasyNetQ.Management.Client` | 3.0.1 | RabbitMQ Management API calls |
 | `Scrutor` | 4.2.2 | Assembly scanning for consumers |
 | `SimplyWorks.HttpExtensions` | 8.1.1 | JWT and request context propagation |
 | `SimplyWorks.PrimitiveTypes` | 10.0.0 | `RequestContext`, `IConsume<T>`, etc. |
 
 ---
 
+## Concurrency, Publisher Confirms and Shutdown
+
+### Consumer concurrency
+
+Every queue a node consumes has its own channel. The channel's consumer dispatch concurrency is how many messages of that queue the node processes at the same time:
+
+| Setting | Where | Default |
+|---|---|---|
+| `ConsumerOptions.Concurrency` | `IConsumeExtended` / `IConsumeExtended<T>` options | null: falls through |
+| `concurrency:` | `BusOptions.AddQueueOption` | null: falls through |
+| `BusOptions.DefaultConsumerConcurrency` | the bus | 1 |
+
+- **1** processes the queue's messages one at a time, in the order they arrive on the node. This is the behaviour of 10.x.
+- **Above 1** processes them in parallel and their order is not kept. The consumer must be safe to run concurrently (it gets a DI scope per message, as before).
+- **Prefetch still caps it.** The broker never sends a node more than prefetch unacknowledged messages, so the effective concurrency is the smaller of the two. Prefetch is the buffer, concurrency is the parallelism.
+- Broadcast listeners (`IListen<T>`) always run one at a time.
+
+Parallel handlers also run in parallel against whatever they call: a database pool, a partner's API, memory for payloads. Size the concurrency with that in mind.
+
+### Publisher confirms
+
+With `BusOptions.PublisherConfirms` (default true), `IPublish`, `IPublishWithValues` and `IBroadcast` complete only when the broker confirms it has the message, and throw when the broker refuses it or cannot be reached. A publish that returned means the broker has the message.
+
+- Each DI scope publishes on a channel of its own, opened on the scope's first publish and closed with the scope. The channel is safe to publish on from several threads of the scope.
+- A publish waits about one broker round trip (plus the broker's disk write for persistent messages) for its confirm. Publishes in parallel (for example `Task.WhenAll`) share that wait.
+- The client adds an `x-dotnet-pub-seq-no` header to each confirmed message, its own tracking number. It is not copied to the dead-letter copy of a failed message.
+- The publisher connection and the consumer connection are separate, as before, so the broker's flow control on a busy publisher never holds back consumer acknowledgements.
+
+### Connecting and recovery
+
+The bus no longer connects inside `AddBus` or `AddBusPublish`. The connections open, and the exchanges are declared, in the background after the host starts, retrying until the broker answers. A host whose broker is down at boot therefore starts, and connects when the broker returns. Use `IBusConnectionStatus.IsConnected` (and `LastError`) for a readiness check.
+
+After an outage the client's automatic recovery reconnects and restores the exchanges, queues, bindings and consumers on their existing channels. The bus does not re-attach consumers itself, so a recovered node has exactly one consumer per queue.
+
+### Shutdown
+
+When the host stops, the bus:
+
+1. stops starting new messages and cancels every consumer at the broker;
+2. waits up to `BusOptions.DrainTimeout` (default 30 s, also bounded by the host's shutdown token) for the messages being processed to finish and be acknowledged; they can still publish;
+3. closes the channels and connections.
+
+Messages prefetched but not started, and any still running when the drain times out, are unacknowledged and return to their queue when the channel closes, to be processed again by another node. The .NET host's own `HostOptions.ShutdownTimeout` (30 s by default) and, on Kubernetes, `terminationGracePeriodSeconds` (30 s by default) must leave room for the drain.
+
+---
+
+## Upgrading to 10.1
+
+10.1 moves the bus to RabbitMQ.Client 7. The public interfaces (`IPublish`, `IPublishWithValues`, `IBroadcast`, `IConsume`, `IConsume<T>`, `IConsumeExtended`, `IListen<T>`) and the option names are unchanged. Queue names, arguments and bindings, the retry and bad queue TTLs, and the exchanges are exactly as they were, so a 10.1 node redeclares an existing 10.0 topology without the broker refusing it, and 10.0 and 10.1 nodes can run side by side during a rollout.
+
+What changes in behaviour:
+
+- **Concurrency.** New, per queue; the default of 1 keeps the 10.x behaviour.
+- **Publisher confirms.** On by default: a publish now waits for the broker and throws when the broker does not have the message. Set `PublisherConfirms = false` for the 10.x fire-and-forget behaviour.
+- **Connecting.** `AddBus` and `AddBusPublish` no longer connect or declare the exchanges, and no longer throw when the broker is down; the host starts and the bus connects in the background. Check `IBusConnectionStatus` where readiness matters.
+- **Shutdown.** Stopping now drains running messages (`DrainTimeout`) instead of cutting them off.
+- **Message headers.** Published messages carry the client's `x-dotnet-pub-seq-no` header when confirms are on.
+
+What can break:
+
+- **`RabbitMQ.Client` 7 arrives transitively.** One process loads one version of the client, so code in the same application that uses `RabbitMQ.Client` directly must move to the 7 API at the same time: `IModel` is `IChannel`, `CreateModel()` is `CreateChannelAsync()`, every operation is async (`QueueDeclareAsync`, `BasicPublishAsync`, `BasicAckAsync`), consumers use `ReceivedAsync`, and `BasicDeliverEventArgs.BasicProperties` is `IReadOnlyBasicProperties`. See the [client's migration guide](https://github.com/rabbitmq/rabbitmq-dotnet-client/blob/main/v7-MIGRATION.md).
+- **`ConnectionFactory` is no longer registered in DI** by `AddBusConsume` / `AddBusListen`. Create your own if you resolved it.
+- **`BusOptions.AddQueueOption`** gained an optional `concurrency` parameter at the end: source compatible, but code compiled against 10.x must be rebuilt.
+- **`ConsumerDefinition.ConsumerObject`** is the client 7 `AsyncEventingBasicConsumer`.
+
+---
+
 ## Contributing
 
 1. Fork the repository

SW.Bus.IntegrationTests/Broker.cs

@@ -0,0 +1,102 @@
+using System;
+using System.Diagnostics;
+using System.Linq;
+using System.Net.Http;
+using System.Net.Http.Headers;
+using System.Text;
+using System.Text.Json;
+using System.Threading.Tasks;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Exceptions;
+using Testcontainers.RabbitMq;
+
+namespace SW.Bus.IntegrationTests;
+
+/// <summary>Helpers over a Testcontainers broker shared by the tests of a class.</summary>
+public static class Broker
+{
+    public static RabbitMqContainer Create() => new RabbitMqBuilder()
+        .WithImage("rabbitmq:3.13-management")
+        .WithPortBinding(15672, true)
+        .Build();
+
+    public static string ManagementUrl(RabbitMqContainer container) =>
+        $"http://{container.Hostname}:{container.GetMappedPublicPort(15672)}";
+
+    /// <summary>Stops the broker application inside the container: every client connection drops.</summary>
+    public static async Task StopApp(RabbitMqContainer container) =>
+        await Exec(container, "rabbitmqctl", "stop_app");
+
+    public static async Task StartApp(RabbitMqContainer container)
+    {
+        await Exec(container, "rabbitmqctl", "start_app");
+        await Exec(container, "rabbitmqctl", "await_startup");
+    }
+
+    private static async Task Exec(RabbitMqContainer container, params string[] command)
+    {
+        var result = await container.ExecAsync(command);
+        if (result.ExitCode != 0)
+            throw new InvalidOperationException($"{string.Join(' ', command)} failed: {result.Stderr}");
+    }
+
+    /// <summary>Ready plus unacknowledged messages, and consumers, of a queue, read on a channel of its own.</summary>
+    public static async Task<(uint Messages, uint Consumers)> Queue(RabbitMqContainer container, string queue)
+    {
+        await using var conn = await new ConnectionFactory { Uri = new Uri(container.GetConnectionString()) }
+            .CreateConnectionAsync();
+        await using var channel = await conn.CreateChannelAsync();
+        try
+        {
+            var ok = await channel.QueueDeclarePassiveAsync(queue);
+            return (ok.MessageCount, ok.ConsumerCount);
+        }
+        catch (OperationInterruptedException)
+        {
+            return (0, 0);
+        }
+    }
+
+    /// <summary>
+    /// Consumers of a queue read through the management API, which also sees exclusive queues that
+    /// a passive declare from another connection is refused.
+    /// </summary>
+    public static async Task<long> ManagementConsumers(RabbitMqContainer container, string queue)
+    {
+        using var json = await Management(container, $"/api/queues/%2F/{Uri.EscapeDataString(queue)}");
+        return json != null && json.RootElement.TryGetProperty("consumers", out var consumers) ? consumers.GetInt64() : 0;
+    }
+
+    /// <summary>The prefetch of each consumer of a queue, as the broker holds it.</summary>
+    public static async Task<long[]> ManagementPrefetch(RabbitMqContainer container, string queue)
+    {
+        using var all = await Management(container, "/api/consumers");
+        if (all == null) return [];
+        return all.RootElement.EnumerateArray()
+            .Where(c => c.GetProperty("queue").GetProperty("name").GetString() == queue)
+            .Select(c => c.GetProperty("prefetch_count").GetInt64())
+            .ToArray();
+    }
+
+    private static async Task<JsonDocument?> Management(RabbitMqContainer container, string path)
+    {
+        var user = new Uri(container.GetConnectionString()).UserInfo;
+        using var http = new HttpClient { BaseAddress = new Uri(ManagementUrl(container)) };
+        http.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Basic",
+            Convert.ToBase64String(Encoding.UTF8.GetBytes(Uri.UnescapeDataString(user))));
+        var response = await http.GetAsync(path);
+        if (!response.IsSuccessStatusCode) return null;
+        return JsonDocument.Parse(await response.Content.ReadAsStringAsync());
+    }
+
+    public static async Task<bool> Eventually(Func<Task<bool>> condition, int seconds = 30)
+    {
+        var sw = Stopwatch.StartNew();
+        while (!await condition() && sw.Elapsed < TimeSpan.FromSeconds(seconds))
+            await Task.Delay(250);
+        return await condition();
+    }
+
+    public static Task<bool> Eventually(Func<bool> condition, int seconds = 30) =>
+        Eventually(() => Task.FromResult(condition()), seconds);
+}

SW.Bus.IntegrationTests/BusHarness.cs

@@ -3,6 +3,7 @@
 using System.Threading.Tasks;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
 using Microsoft.Extensions.Hosting;
 using SW.PrimitiveTypes;
 
@@ -21,7 +22,9 @@ public sealed class BusHarness : IAsyncDisposable
     public IServiceProvider Services => host.Services;
     public BusOptions Options => host.Services.GetRequiredService<BusOptions>();
 
-    public static async Task<BusHarness> StartAsync(string amqpConnectionString)
+    public static async Task<BusHarness> StartAsync(string amqpConnectionString,
+        Action<BusOptions>? configureBus = null, Action<IServiceCollection>? configureServices = null,
+        bool waitForTopology = true)
     {
         var host = Host.CreateDefaultBuilder()
             .UseEnvironment("itest")
@@ -31,23 +34,32 @@ public static async Task<BusHarness> StartAsync(string amqpConnectionString)
             }))
             .ConfigureServices(services =>
             {
+                configureServices?.Invoke(services);
+                services.TryAddSingleton<WorkPlan>();
                 services.AddSingleton<DynamicMessageTypes>();
                 services.AddSingleton<ReceivedValues>();
                 services.AddScoped<RequestContext>();
-                services.AddBus(o => o.ApplicationName = "itest");
+                services.AddBus(o =>
+                {
+                    o.ApplicationName = "itest";
+                    configureBus?.Invoke(o);
+                });
                 services.AddBusConsume(typeof(BusHarness).Assembly);
                 services.AddBusPublish();
             })
             .Build();
 
         await host.StartAsync();
+        if (!waitForTopology) return new BusHarness(host);
 
         // ConsumersService attaches topology on a background task; give it time to declare and bind
         // the consumer queues before the test touches them.
         await Task.Delay(TimeSpan.FromSeconds(4));
         return new BusHarness(host);
     }
 
+    public Task StopAsync() => host.StopAsync();
+
     public async ValueTask DisposeAsync()
     {
         await host.StopAsync();