From 11b168315c6fa0bc316c1d403c8fdcdc85ef6955 Mon Sep 17 00:00:00 2001 From: Samerz Date: Wed, 14 Oct 2026 16:10:01 +0300 Subject: [PATCH 1/2] SW-Bus 11.0: RabbitMQ.Client 7, per-queue concurrency, publisher confirms, drain - RabbitMQ.Client 6.8.1 to 7.2.2; package major 11. - Per-queue consumer concurrency: ConsumerOptions.Concurrency, AddQueueOption(concurrency:), BusOptions.DefaultConsumerConcurrency (default 1, the 10.x behaviour). Each queue's channel is created with that dispatch concurrency; prefetch still caps it. A refresh that changes it moves the consumer to a new channel and retires the old one once its running messages are acknowledged, so nothing is processed twice. - Publisher confirms (BusOptions.PublisherConfirms, default true): a publish completes when the broker has the message and throws otherwise. One channel per DI scope, opened on first publish, instead of one shared channel for the process. - BusConnection: publisher and consumer connections opened asynchronously after the host starts, exchanges declared there with retries; AddBus and AddBusPublish no longer connect. The client's automatic recovery owns reconnects and topology. IBusConnectionStatus reports whether the bus is connected. - Drain on stop (BusOptions.DrainTimeout, default 30 s): no new message starts, consumers are cancelled, running messages finish and ack, the rest return to their queue. Channel and connection closes are bounded because client 7 waits for running handlers on close. - Queue names, arguments, bindings, retry and bad queue TTLs and exchanges unchanged; the bad-queue copy drops the client's x-dotnet-pub-seq-no header as it drops x-death. - Integration tests for concurrency, order at 1, confirms, broker restart, boot with the broker down, drain, retry and bad queue; README options, behaviour and upgrade notes. --- .github/workflows/nuget-publish.yml | 2 +- README.md | 111 +++- SW.Bus.IntegrationTests/Broker.cs | 102 ++++ SW.Bus.IntegrationTests/BusHarness.cs | 16 +- SW.Bus.IntegrationTests/ConcurrencyTests.cs | 175 +++++++ SW.Bus.IntegrationTests/ConnectionTests.cs | 133 +++++ .../ConsumerRefreshTests.cs | 28 +- SW.Bus.IntegrationTests/DrainAndRetryTests.cs | 101 ++++ SW.Bus.IntegrationTests/WorkConsumer.cs | 73 +++ SW.Bus.RabbitMqExtensions/ConsumerOptions.cs | 8 + .../IBusConnectionStatus.cs | 20 + .../SW.Bus.RabbitMqExtensions.csproj | 2 +- SW.Bus/BasicPublisher.cs | 71 ++- SW.Bus/BusConnection.cs | 275 ++++++++++ SW.Bus/BusOptions.cs | 39 +- SW.Bus/ConsumerDefinition.cs | 3 + SW.Bus/ConsumerRunner.cs | 33 +- SW.Bus/ConsumersService.cs | 479 +++++++++++++----- SW.Bus/IServiceCollectionExtensions.cs | 67 +-- SW.Bus/OperationalEventInfrastructure.cs | 12 +- SW.Bus/SW.Bus.csproj | 4 +- 21 files changed, 1523 insertions(+), 231 deletions(-) create mode 100644 SW.Bus.IntegrationTests/Broker.cs create mode 100644 SW.Bus.IntegrationTests/ConcurrencyTests.cs create mode 100644 SW.Bus.IntegrationTests/ConnectionTests.cs create mode 100644 SW.Bus.IntegrationTests/DrainAndRetryTests.cs create mode 100644 SW.Bus.IntegrationTests/WorkConsumer.cs create mode 100644 SW.Bus.RabbitMqExtensions/IBusConnectionStatus.cs create mode 100644 SW.Bus/BusConnection.cs diff --git a/.github/workflows/nuget-publish.yml b/.github/workflows/nuget-publish.yml index eee6251..0cf876d 100644 --- a/.github/workflows/nuget-publish.yml +++ b/.github/workflows/nuget-publish.yml @@ -22,7 +22,7 @@ jobs: SW.Bus/SW.Bus.csproj SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj SW.Bus.RabbitMqViewer/SW.Bus.RabbitMqViewer.csproj - major-version: '10' + major-version: '11' minor-version: '0' dotnet-version: '10.0.x' run-tests: false diff --git a/README.md b/README.md index 226f173..24be41f 100644 --- a/README.md +++ b/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 from 10.x to 11.0](#upgrading-from-10x-to-110) ## 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` 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 { ["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,13 +1020,23 @@ 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 | --- @@ -1016,7 +1044,7 @@ dotnet test SW.Bus.IntegrationTests/SW.Bus.IntegrationTests.csproj | 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 | @@ -1024,6 +1052,73 @@ dotnet test SW.Bus.IntegrationTests/SW.Bus.IntegrationTests.csproj --- +## 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` 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`) 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 from 10.x to 11.0 + +11.0 moves the bus to RabbitMQ.Client 7. The public interfaces (`IPublish`, `IPublishWithValues`, `IBroadcast`, `IConsume`, `IConsume`, `IConsumeExtended`, `IListen`) 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 an 11.0 node redeclares an existing 10.x topology without the broker refusing it, and 10.x and 11.0 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 diff --git a/SW.Bus.IntegrationTests/Broker.cs b/SW.Bus.IntegrationTests/Broker.cs new file mode 100644 index 0000000..1204cba --- /dev/null +++ b/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; + +/// Helpers over a Testcontainers broker shared by the tests of a class. +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)}"; + + /// Stops the broker application inside the container: every client connection drops. + 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}"); + } + + /// Ready plus unacknowledged messages, and consumers, of a queue, read on a channel of its own. + 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); + } + } + + /// + /// Consumers of a queue read through the management API, which also sees exclusive queues that + /// a passive declare from another connection is refused. + /// + public static async Task 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; + } + + /// The prefetch of each consumer of a queue, as the broker holds it. + public static async Task 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 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 Eventually(Func> 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 Eventually(Func condition, int seconds = 30) => + Eventually(() => Task.FromResult(condition()), seconds); +} diff --git a/SW.Bus.IntegrationTests/BusHarness.cs b/SW.Bus.IntegrationTests/BusHarness.cs index 9c51567..5a64f71 100644 --- a/SW.Bus.IntegrationTests/BusHarness.cs +++ b/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(); - public static async Task StartAsync(string amqpConnectionString) + public static async Task StartAsync(string amqpConnectionString, + Action? configureBus = null, Action? configureServices = null, + bool waitForTopology = true) { var host = Host.CreateDefaultBuilder() .UseEnvironment("itest") @@ -31,16 +34,23 @@ public static async Task StartAsync(string amqpConnectionString) })) .ConfigureServices(services => { + configureServices?.Invoke(services); + services.TryAddSingleton(); services.AddSingleton(); services.AddSingleton(); services.AddScoped(); - 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. @@ -48,6 +58,8 @@ public static async Task StartAsync(string amqpConnectionString) return new BusHarness(host); } + public Task StopAsync() => host.StopAsync(); + public async ValueTask DisposeAsync() { await host.StopAsync(); diff --git a/SW.Bus.IntegrationTests/ConcurrencyTests.cs b/SW.Bus.IntegrationTests/ConcurrencyTests.cs new file mode 100644 index 0000000..5768b6b --- /dev/null +++ b/SW.Bus.IntegrationTests/ConcurrencyTests.cs @@ -0,0 +1,175 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using SW.Bus.RabbitMqExtensions; +using SW.PrimitiveTypes; +using Testcontainers.RabbitMq; + +namespace SW.Bus.IntegrationTests; + +/// +/// Per-queue consumer concurrency against a real broker: a queue runs as many messages at once as +/// its concurrency allows (capped by prefetch), and at 1 it runs them one at a time, in order. +/// +[TestClass] +public class ConcurrencyTests +{ + private static RabbitMqContainer _container = null!; + + [ClassInitialize] + public static async Task ClassInitialize(TestContext _) + { + _container = Broker.Create(); + await _container.StartAsync(); + } + + [ClassCleanup] + public static async Task ClassCleanup() => await _container.DisposeAsync(); + + private static Task Slow(string _) => Task.Delay(1000); + + [TestMethod] + public async Task A_queue_runs_as_many_messages_at_once_as_its_concurrency() + { + var plan = new WorkPlan(); + plan.Types["parallel"] = new ConsumerOptions { Prefetch = 12, Concurrency = 4 }; + plan.Handlers["parallel"] = Slow; + await using var harness = await BusHarness.StartAsync(_container.GetConnectionString(), + configureServices: s => s.AddSingleton(plan)); + + await Publish(harness, "parallel", 8); + + Assert.IsTrue(await Broker.Eventually(() => plan.Log("parallel").Completed.Count == 8), "The messages never completed."); + Assert.AreEqual(4, plan.Log("parallel").MaxRunning); + } + + [TestMethod] + public async Task At_concurrency_one_a_queue_runs_one_message_at_a_time_in_order() + { + var plan = new WorkPlan(); + plan.Types["ordered"] = new ConsumerOptions { Prefetch = 12 }; + plan.Handlers["ordered"] = _ => Task.Delay(30); + await using var harness = await BusHarness.StartAsync(_container.GetConnectionString(), + configureServices: s => s.AddSingleton(plan)); + + await Publish(harness, "ordered", 20); + + Assert.IsTrue(await Broker.Eventually(() => plan.Log("ordered").Completed.Count == 20), "The messages never completed."); + Assert.AreEqual(1, plan.Log("ordered").MaxRunning); + CollectionAssert.AreEqual(Enumerable.Range(0, 20).Select(i => i.ToString()).ToList(), + plan.Log("ordered").Completed.ToList()); + } + + [TestMethod] + public async Task Concurrency_comes_from_the_queue_option_then_the_bus_default_and_is_capped_by_prefetch() + { + var plan = new WorkPlan(); + plan.Types["byoption"] = new ConsumerOptions { Prefetch = 10 }; + plan.Types["bydefault"] = new ConsumerOptions { Prefetch = 10 }; + plan.Types["capped"] = new ConsumerOptions { Prefetch = 2, Concurrency = 6 }; + foreach (var type in plan.Types.Keys) plan.Handlers[type] = Slow; + await using var harness = await BusHarness.StartAsync(_container.GetConnectionString(), + configureBus: o => + { + o.DefaultConsumerConcurrency = 3; + o.AddQueueOption("workconsumer.byoption", concurrency: 2); + }, + configureServices: s => s.AddSingleton(plan)); + + foreach (var type in plan.Types.Keys) await Publish(harness, type, 6); + + Assert.IsTrue(await Broker.Eventually(() => plan.Types.Keys.All(t => plan.Log(t).Completed.Count == 6)), + "The messages never completed."); + Assert.AreEqual(2, plan.Log("byoption").MaxRunning, "AddQueueOption's concurrency"); + Assert.AreEqual(3, plan.Log("bydefault").MaxRunning, "DefaultConsumerConcurrency"); + Assert.AreEqual(2, plan.Log("capped").MaxRunning, "prefetch caps concurrency"); + } + + [TestMethod] + public async Task Refresh_moves_a_consumer_to_its_new_concurrency_without_processing_a_message_twice() + { + var plan = new WorkPlan(); + plan.Types["moving"] = new ConsumerOptions { Prefetch = 12, Concurrency = 4 }; + plan.Handlers["moving"] = _ => Task.Delay(1500); + await using var harness = await BusHarness.StartAsync(_container.GetConnectionString(), + configureServices: s => s.AddSingleton(plan)); + var queue = $"{harness.Options.ProcessExchange}.itest.workconsumer.moving"; + + await Publish(harness, "moving", 6); + Assert.IsTrue(await Broker.Eventually(() => plan.Log("moving").Started.Count == 4), "The first messages never started."); + + // Changed while four are running and two wait in the old channel's prefetch. + plan.Types["moving"] = new ConsumerOptions { Prefetch = 12, Concurrency = 2 }; + using (var scope = harness.Services.CreateScope()) + await scope.ServiceProvider.GetRequiredService().RefreshConsumers(); + + Assert.IsTrue(await Broker.Eventually(() => plan.Log("moving").Completed.Count == 6), "The messages never completed."); + await Task.Delay(2000); + Assert.AreEqual(6, plan.Log("moving").Started.Count, "A message was processed twice."); + Assert.AreEqual(0u, (await Broker.Queue(_container, queue)).Messages); + Assert.AreEqual(1u, (await Broker.Queue(_container, queue)).Consumers); + + plan.Logs.TryRemove("moving", out _); + await Publish(harness, "moving", 6); + Assert.IsTrue(await Broker.Eventually(() => plan.Log("moving").Completed.Count == 6), "The later messages never completed."); + Assert.AreEqual(2, plan.Log("moving").MaxRunning); + } + + [TestMethod] + public async Task Refresh_applies_a_new_prefetch_to_the_running_consumer() + { + var plan = new WorkPlan(); + plan.Types["refetch"] = new ConsumerOptions { Prefetch = 2 }; + await using var harness = await BusHarness.StartAsync(_container.GetConnectionString(), + configureServices: s => s.AddSingleton(plan)); + var queue = $"{harness.Options.ProcessExchange}.itest.workconsumer.refetch"; + Assert.IsTrue(await Broker.Eventually(async () => + (await Broker.ManagementPrefetch(_container, queue)).SequenceEqual(new long[] { 2 })), "Prefetch 2 never showed."); + + plan.Types["refetch"] = new ConsumerOptions { Prefetch = 5 }; + using (var scope = harness.Services.CreateScope()) + await scope.ServiceProvider.GetRequiredService().RefreshConsumers(); + + Assert.IsTrue(await Broker.Eventually(async () => + (await Broker.ManagementPrefetch(_container, queue)).SequenceEqual(new long[] { 5 })), "Prefetch 5 never showed."); + await Publish(harness, "refetch", 3); + Assert.IsTrue(await Broker.Eventually(() => plan.Log("refetch").Completed.Count == 3), + "The re-attached consumer did not process."); + } + + [TestMethod] + public async Task Concurrent_publishes_from_one_scope_and_from_many_scopes_all_arrive() + { + var plan = new WorkPlan(); + plan.Types["many"] = new ConsumerOptions { Prefetch = 50, Concurrency = 8 }; + await using var harness = await BusHarness.StartAsync(_container.GetConnectionString(), + configureServices: s => s.AddSingleton(plan)); + + await using (var scope = harness.Services.CreateAsyncScope()) + { + var publish = scope.ServiceProvider.GetRequiredService(); + await Task.WhenAll(Enumerable.Range(0, 200).Select(i => publish.Publish("many", $"s{i}"))); + } + + await Task.WhenAll(Enumerable.Range(0, 50).Select(async i => + { + await using var scope = harness.Services.CreateAsyncScope(); + await scope.ServiceProvider.GetRequiredService().Publish("many", $"m{i}"); + })); + + Assert.IsTrue(await Broker.Eventually(() => plan.Log("many").Completed.Count == 250), + $"Only {plan.Log("many").Completed.Count} of 250 arrived."); + Assert.AreEqual(250, plan.Log("many").Completed.Distinct().Count()); + } + + internal static async Task Publish(BusHarness harness, string type, int count) + { + await using var scope = harness.Services.CreateAsyncScope(); + var publish = scope.ServiceProvider.GetRequiredService(); + for (var i = 0; i < count; i++) + await publish.Publish(type, i.ToString()); + } +} diff --git a/SW.Bus.IntegrationTests/ConnectionTests.cs b/SW.Bus.IntegrationTests/ConnectionTests.cs new file mode 100644 index 0000000..7e0c2ac --- /dev/null +++ b/SW.Bus.IntegrationTests/ConnectionTests.cs @@ -0,0 +1,133 @@ +using System; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using SW.Bus.RabbitMqExtensions; +using SW.PrimitiveTypes; +using Testcontainers.RabbitMq; + +namespace SW.Bus.IntegrationTests; + +/// +/// The connection's life against a real broker that is stopped and started under it: publisher +/// confirms, recovery without duplicate consumers, and a host that boots while the broker is down. +/// +[TestClass] +public class ConnectionTests +{ + private static RabbitMqContainer _container = null!; + + [ClassInitialize] + public static async Task ClassInitialize(TestContext _) + { + _container = Broker.Create(); + await _container.StartAsync(); + } + + [ClassCleanup] + public static async Task ClassCleanup() => await _container.DisposeAsync(); + + [TestCleanup] + public async Task TestCleanup() => await Broker.StartApp(_container); + + [TestMethod] + public async Task A_publish_against_a_stopped_broker_throws_and_publishing_resumes_when_it_returns() + { + var plan = new WorkPlan(); + plan.Types["confirmed"] = new ConsumerOptions(); + await using var harness = await BusHarness.StartAsync(_container.GetConnectionString(), + configureServices: s => s.AddSingleton(plan)); + + await ConcurrencyTests.Publish(harness, "confirmed", 1); + Assert.IsTrue(await Broker.Eventually(() => plan.Log("confirmed").Completed.Count == 1), "The first publish never arrived."); + + await Broker.StopApp(_container); + await using (var scope = harness.Services.CreateAsyncScope()) + { + var publish = scope.ServiceProvider.GetRequiredService().Publish("confirmed", "lost"); + var finished = await Task.WhenAny(publish, Task.Delay(TimeSpan.FromSeconds(30))); + Assert.AreSame(publish, finished, "The publish neither completed nor failed."); + Assert.IsTrue(publish.IsFaulted, "A publish to a stopped broker returned as if it was sent."); + } + + await Broker.StartApp(_container); + Assert.IsTrue(await Broker.Eventually(async () => + { + try + { + await ConcurrencyTests.Publish(harness, "confirmed", 1); + return true; + } + catch + { + return false; + } + }, 60), "Publishing never resumed."); + Assert.IsTrue(await Broker.Eventually(() => plan.Log("confirmed").Completed.Count >= 2), "The later publish never arrived."); + Assert.IsFalse(plan.Log("confirmed").Completed.Contains("lost")); + } + + [TestMethod] + public async Task A_broker_restart_leaves_exactly_one_consumer_per_queue() + { + var plan = new WorkPlan(); + plan.Types["first"] = new ConsumerOptions { Concurrency = 2 }; + plan.Types["second"] = new ConsumerOptions(); + await using var harness = await BusHarness.StartAsync(_container.GetConnectionString(), + configureBus: o => o.ManagementUrl = Broker.ManagementUrl(_container), + configureServices: s => s.AddSingleton(plan)); + var reader = harness.Services.GetRequiredService(); + var status = harness.Services.GetRequiredService(); + + async Task OneEach() + { + var counts = (await reader.GetAllConsumersCount()).Where(c => c.Name == nameof(WorkConsumer)).ToList(); + return counts.Count == 2 && counts.All(c => c.TotalNodes == 1); + } + + Assert.IsTrue(status.IsConnected); + Assert.IsTrue(await Broker.Eventually(OneEach), "Before the restart, not one consumer per queue."); + + await Broker.StopApp(_container); + Assert.IsTrue(await Broker.Eventually(() => !status.IsConnected), "The bus never noticed the broker went away."); + await Broker.StartApp(_container); + + Assert.IsTrue(await Broker.Eventually(() => status.IsConnected, 60), "The bus never reconnected."); + Assert.IsTrue(await Broker.Eventually(OneEach, 60), "After the restart, not one consumer per queue."); + // A late second attach would show here: the reader caches for up to MonitoringCacheSeconds. + await Task.Delay(TimeSpan.FromSeconds(12)); + Assert.IsTrue(await OneEach(), "A second consumer attached after the restart."); + Assert.AreEqual(1, await Broker.ManagementConsumers(_container, harness.Options.NodeQueueName), "Node queue consumers."); + + await ConcurrencyTests.Publish(harness, "first", 3); + await ConcurrencyTests.Publish(harness, "second", 3); + Assert.IsTrue(await Broker.Eventually(() => + plan.Log("first").Completed.Count == 3 && plan.Log("second").Completed.Count == 3), + "Messages published after the restart were not processed."); + } + + [TestMethod] + public async Task A_host_whose_broker_is_down_at_boot_starts_reports_not_connected_and_connects_when_it_returns() + { + var plan = new WorkPlan(); + plan.Types["late"] = new ConsumerOptions(); + await Broker.StopApp(_container); + + await using var harness = await BusHarness.StartAsync(_container.GetConnectionString(), + configureServices: s => s.AddSingleton(plan), waitForTopology: false); + var status = harness.Services.GetRequiredService(); + await Task.Delay(TimeSpan.FromSeconds(3)); + Assert.IsFalse(status.IsConnected); + Assert.IsNotNull(status.LastError); + + await Broker.StartApp(_container); + Assert.IsTrue(await Broker.Eventually(() => status.IsConnected, 60), "The bus never connected."); + var queue = $"{harness.Options.ProcessExchange}.itest.workconsumer.late"; + Assert.IsTrue(await Broker.Eventually(async () => (await Broker.Queue(_container, queue)).Consumers == 1, 60), + "The consumer never attached."); + + await ConcurrencyTests.Publish(harness, "late", 1); + Assert.IsTrue(await Broker.Eventually(() => plan.Log("late").Completed.Count == 1), "The message never arrived."); + } +} diff --git a/SW.Bus.IntegrationTests/ConsumerRefreshTests.cs b/SW.Bus.IntegrationTests/ConsumerRefreshTests.cs index e3d7cc6..9507585 100644 --- a/SW.Bus.IntegrationTests/ConsumerRefreshTests.cs +++ b/SW.Bus.IntegrationTests/ConsumerRefreshTests.cs @@ -37,15 +37,15 @@ public async Task Refresh_detaches_a_consumer_that_is_no_longer_defined() await using var harness = await BusHarness.StartAsync(_container.GetConnectionString()); var prefix = $"{harness.Options.ProcessExchange}.itest.dynamicconsumer"; - using var conn = new ConnectionFactory { Uri = new Uri(_container.GetConnectionString()) }.CreateConnection(); + await using var conn = await new ConnectionFactory { Uri = new Uri(_container.GetConnectionString()) }.CreateConnectionAsync(); // A channel per check: a passive declare of a queue that doesn't exist yet closes its channel. - uint Consumers(string name) + async Task Consumers(string name) { - using var model = conn.CreateModel(); + await using var model = await conn.CreateChannelAsync(); try { - return model.QueueDeclarePassive($"{prefix}.{name}").ConsumerCount; + return (await model.QueueDeclarePassiveAsync($"{prefix}.{name}")).ConsumerCount; } catch (OperationInterruptedException) { @@ -54,27 +54,27 @@ uint Consumers(string name) } // Consumers attach on a background task, so both are waited for rather than assumed. - Assert.IsTrue(await Eventually(() => Consumers("kept") == 1 && Consumers("removed") == 1), + Assert.IsTrue(await Eventually(async () => await Consumers("kept") == 1 && await Consumers("removed") == 1), "The consumers never attached."); harness.Services.GetRequiredService().Names.TryRemove("removed", out _); using (var scope = harness.Services.CreateScope()) await scope.ServiceProvider.GetRequiredService().RefreshConsumers(); - Assert.IsTrue(await Eventually(() => Consumers("removed") == 0), "The removed consumer is still attached."); - Assert.AreEqual(1u, Consumers("kept"), "Refresh detached a consumer that is still defined."); + Assert.IsTrue(await Eventually(async () => await Consumers("removed") == 0), "The removed consumer is still attached."); + Assert.AreEqual(1u, await Consumers("kept"), "Refresh detached a consumer that is still defined."); // Passive declare throws if the queue is gone: stopping the consumer must leave it in place. - using var check = conn.CreateModel(); - check.QueueDeclarePassive($"{prefix}.removed"); - check.QueueDeclarePassive($"{prefix}.removed.retry"); - check.QueueDeclarePassive($"{prefix}.removed.bad"); + await using var check = await conn.CreateChannelAsync(); + await check.QueueDeclarePassiveAsync($"{prefix}.removed"); + await check.QueueDeclarePassiveAsync($"{prefix}.removed.retry"); + await check.QueueDeclarePassiveAsync($"{prefix}.removed.bad"); } - private static async Task Eventually(Func condition) + private static async Task Eventually(Func> condition) { var sw = Stopwatch.StartNew(); - while (!condition() && sw.Elapsed < TimeSpan.FromSeconds(15)) + while (!await condition() && sw.Elapsed < TimeSpan.FromSeconds(15)) await Task.Delay(250); - return condition(); + return await condition(); } } diff --git a/SW.Bus.IntegrationTests/DrainAndRetryTests.cs b/SW.Bus.IntegrationTests/DrainAndRetryTests.cs new file mode 100644 index 0000000..8f83a65 --- /dev/null +++ b/SW.Bus.IntegrationTests/DrainAndRetryTests.cs @@ -0,0 +1,101 @@ +using System; +using System.Linq; +using System.Text; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using RabbitMQ.Client; +using SW.Bus.RabbitMqExtensions; +using Testcontainers.RabbitMq; + +namespace SW.Bus.IntegrationTests; + +/// +/// What happens to messages when the host stops, and when processing fails, against a real broker. +/// +[TestClass] +public class DrainAndRetryTests +{ + private static RabbitMqContainer _container = null!; + + [ClassInitialize] + public static async Task ClassInitialize(TestContext _) + { + _container = Broker.Create(); + await _container.StartAsync(); + } + + [ClassCleanup] + public static async Task ClassCleanup() => await _container.DisposeAsync(); + + [TestMethod] + public async Task Stopping_lets_running_messages_finish_and_ack_and_returns_the_ones_not_started() + { + var plan = new WorkPlan(); + plan.Types["draining"] = new ConsumerOptions { Prefetch = 4, Concurrency = 2 }; + plan.Handlers["draining"] = _ => Task.Delay(3000); + var harness = await BusHarness.StartAsync(_container.GetConnectionString(), + configureServices: s => s.AddSingleton(plan)); + var queue = $"{harness.Options.ProcessExchange}.itest.workconsumer.draining"; + + await ConcurrencyTests.Publish(harness, "draining", 3); + Assert.IsTrue(await Broker.Eventually(() => plan.Log("draining").Started.Count == 2), "The messages never started."); + + await harness.DisposeAsync(); + + Assert.AreEqual(2, plan.Log("draining").Completed.Count, "A running message did not finish before the stop."); + Assert.AreEqual(2, plan.Log("draining").Started.Count, "A message started after the stop began."); + Assert.AreEqual(1u, (await Broker.Queue(_container, queue)).Messages, + "Only the message that never started should be back in the queue."); + } + + [TestMethod] + public async Task A_message_still_running_at_the_drain_timeout_returns_to_its_queue() + { + var plan = new WorkPlan(); + plan.Types["hung"] = new ConsumerOptions(); + plan.Handlers["hung"] = _ => Task.Delay(8000); + var harness = await BusHarness.StartAsync(_container.GetConnectionString(), + configureBus: o => o.DrainTimeout = TimeSpan.FromSeconds(1), + configureServices: s => s.AddSingleton(plan)); + var queue = $"{harness.Options.ProcessExchange}.itest.workconsumer.hung"; + + await ConcurrencyTests.Publish(harness, "hung", 1); + Assert.IsTrue(await Broker.Eventually(() => plan.Log("hung").Started.Count == 1), "The message never started."); + + var stop = System.Diagnostics.Stopwatch.StartNew(); + await harness.DisposeAsync(); + Assert.AreEqual(1u, (await Broker.Queue(_container, queue)).Messages); + } + + [TestMethod] + public async Task A_failing_message_is_retried_its_retry_count_then_moved_to_the_bad_queue() + { + var plan = new WorkPlan(); + plan.Types["flaky"] = new ConsumerOptions(); + plan.Handlers["flaky"] = _ => throw new InvalidOperationException("always fails"); + await using var harness = await BusHarness.StartAsync(_container.GetConnectionString(), + configureBus: o => o.AddQueueOption("workconsumer.flaky", retryCount: 2, retryAfterSeconds: 0), + configureServices: s => s.AddSingleton(plan)); + var bad = $"{harness.Options.ProcessExchange}.itest.workconsumer.flaky.bad"; + + await ConcurrencyTests.Publish(harness, "flaky", 1); + + Assert.IsTrue(await Broker.Eventually(async () => (await Broker.Queue(_container, bad)).Messages == 1), + "The message never reached the bad queue."); + CollectionAssert.AreEqual(new[] { "2", "1", "0" }, plan.Log("flaky").RemainingRetries.ToArray()); + + await using var conn = await new ConnectionFactory { Uri = new Uri(_container.GetConnectionString()) } + .CreateConnectionAsync(); + await using var channel = await conn.CreateChannelAsync(); + var message = await channel.BasicGetAsync(bad, true); + Assert.IsNotNull(message); + Assert.AreEqual("0", Encoding.UTF8.GetString(message.Body.Span)); + var headers = message.BasicProperties.Headers!; + Assert.IsTrue(headers.ContainsKey("exception1")); + Assert.IsTrue(headers.ContainsKey("source-node-id")); + Assert.IsFalse(headers.ContainsKey("x-death")); + Assert.IsFalse(headers.ContainsKey(Constants.PublishSequenceNumberHeader)); + Assert.AreEqual(DeliveryModes.Persistent, message.BasicProperties.DeliveryMode); + } +} diff --git a/SW.Bus.IntegrationTests/WorkConsumer.cs b/SW.Bus.IntegrationTests/WorkConsumer.cs new file mode 100644 index 0000000..ce2928d --- /dev/null +++ b/SW.Bus.IntegrationTests/WorkConsumer.cs @@ -0,0 +1,73 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using SW.Bus.RabbitMqExtensions; +using SW.PrimitiveTypes; + +namespace SW.Bus.IntegrationTests; + +/// +/// Singleton a test fills before the host starts: the message types consumes, +/// with their options and what processing one does, and a record of how they were processed. +/// Naked queue names resolve to "workconsumer.{type}". +/// +public class WorkPlan +{ + public ConcurrentDictionary Types { get; } = new(); + public ConcurrentDictionary> Handlers { get; } = new(); + public ConcurrentDictionary Logs { get; } = new(); + + public WorkLog Log(string type) => Logs.GetOrAdd(type, _ => new WorkLog()); +} + +public class WorkLog +{ + private int running; + private int maxRunning; + + public int MaxRunning => Volatile.Read(ref maxRunning); + public ConcurrentQueue Started { get; } = new(); + public ConcurrentQueue Completed { get; } = new(); + public ConcurrentQueue RemainingRetries { get; } = new(); + + internal void Enter() + { + var now = Interlocked.Increment(ref running); + int seen; + while (now > (seen = Volatile.Read(ref maxRunning)) && + Interlocked.CompareExchange(ref maxRunning, now, seen) != seen) + { + } + } + + internal void Exit() => Interlocked.Decrement(ref running); +} + +public class WorkConsumer(WorkPlan plan, RequestContext requestContext) : IConsumeExtended +{ + public Task> GetMessageTypeNames() => Task.FromResult(plan.Types.Keys.AsEnumerable()); + + public Task> GetMessageTypeNamesWithOptions() => + Task.FromResult>(plan.Types.ToDictionary(t => t.Key, t => t.Value)); + + public async Task Process(string messageTypeName, string message) + { + var log = plan.Log(messageTypeName); + log.Started.Enqueue(message); + log.RemainingRetries.Enqueue(requestContext.Values.FirstOrDefault(v => v.Name == "RemainingRetries")?.Value ?? ""); + log.Enter(); + try + { + if (plan.Handlers.TryGetValue(messageTypeName, out var handler)) + await handler(message); + log.Completed.Enqueue(message); + } + finally + { + log.Exit(); + } + } +} diff --git a/SW.Bus.RabbitMqExtensions/ConsumerOptions.cs b/SW.Bus.RabbitMqExtensions/ConsumerOptions.cs index 1b6c1f6..73d4efd 100644 --- a/SW.Bus.RabbitMqExtensions/ConsumerOptions.cs +++ b/SW.Bus.RabbitMqExtensions/ConsumerOptions.cs @@ -14,4 +14,12 @@ public class ConsumerOptions /// Gets or sets the priority of the consumer. Higher values indicate higher priority. /// public int? Priority { get; set; } + + /// + /// Gets or sets how many messages of this queue are processed at the same time on one node. + /// 1 processes them one at a time, in the order they arrive. Above 1 they run in parallel and + /// their order is not kept. Prefetch still caps it: a node never holds more than prefetch + /// unacknowledged messages. When null, BusOptions.DefaultConsumerConcurrency applies. + /// + public ushort? Concurrency { get; set; } } \ No newline at end of file diff --git a/SW.Bus.RabbitMqExtensions/IBusConnectionStatus.cs b/SW.Bus.RabbitMqExtensions/IBusConnectionStatus.cs new file mode 100644 index 0000000..1c33713 --- /dev/null +++ b/SW.Bus.RabbitMqExtensions/IBusConnectionStatus.cs @@ -0,0 +1,20 @@ +namespace SW.Bus.RabbitMqExtensions; + +/// +/// Whether the bus is connected to the broker. The bus connects in the background after the host +/// starts, and reconnects on its own after an outage, so an application can start while the broker +/// is down and use this to report itself not ready until the bus is connected. +/// +public interface IBusConnectionStatus +{ + /// + /// True when the exchanges are declared and every connection the bus has opened is open. + /// False before the first connection and while the client is recovering from a lost one. + /// + bool IsConnected { get; } + + /// + /// The error of the last failed attempt to connect or declare the exchanges, or null. + /// + string? LastError { get; } +} diff --git a/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj b/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj index 31a37d0..46a6b1d 100644 --- a/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj +++ b/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj @@ -2,7 +2,7 @@ net10.0 - 10.0.0.0 + 11.0.0.0 true $(NoWarn);1591 enable diff --git a/SW.Bus/BasicPublisher.cs b/SW.Bus/BasicPublisher.cs index b9d954e..266483b 100644 --- a/SW.Bus/BasicPublisher.cs +++ b/SW.Bus/BasicPublisher.cs @@ -14,13 +14,20 @@ namespace SW.Bus; +/// +/// Publishes for one DI scope 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. +/// internal class BasicPublisher( - IModel model, + BusConnection busConnection, BusOptions busOptions, RequestContext requestContext, IOperationalEventPublisher operationalEventPublisher, - BusMetrics metrics) + BusMetrics metrics) : IDisposable, IAsyncDisposable { + private readonly SemaphoreSlim channelLock = new(1, 1); + private IChannel channel; + async public Task Publish(TMessage message, string exchange, byte? priority = null) { var serializerOptions = new JsonSerializerOptions() @@ -48,14 +55,14 @@ public async Task Publish(string messageTypeName, string message, string exchang } /// Written by , which also checks its size. - public Task Publish(string messageTypeName, byte[] message, string exchange, byte? priority = null, + public async Task Publish(string messageTypeName, byte[] message, string exchange, byte? priority = null, string valuesHeader = null) { var activity = BusDiagnostics.ActivitySource.StartActivity($"bus.publish {messageTypeName}", ActivityKind.Producer); var stopwatch = Stopwatch.StartNew(); - var props = model.CreateBasicProperties(); - props.DeliveryMode = 2; + var props = new BasicProperties(); + props.DeliveryMode = DeliveryModes.Persistent; props.Headers = new Dictionary(); props.MessageId = Guid.NewGuid().ToString("N"); @@ -118,7 +125,10 @@ public Task Publish(string messageTypeName, byte[] message, string exchange, byt try { - model.BasicPublish(exchange, messageTypeName.ToLower(), props, message); + var publishChannel = await GetChannelAsync(); + // With confirms on, this completes when the broker has the message, and throws when the + // broker refuses it or the connection is lost before it answers. + await publishChannel.BasicPublishAsync(exchange, messageTypeName.ToLower(), false, props, message); stopwatch.Stop(); activity?.SetTag("messaging.system", "rabbitmq"); activity?.SetTag("messaging.destination.name", exchange); @@ -178,10 +188,57 @@ public Task Publish(string messageTypeName, byte[] message, string exchange, byt { activity?.Dispose(); } + } + + private async Task GetChannelAsync() + { + var current = channel; + if (current is { IsClosed: false }) return current; + + await channelLock.WaitAsync(); + try + { + // Closed by the broker (a channel error) or never opened: open one. A channel that is + // recovering with its connection is not closed and is kept. + if (channel is { IsClosed: false }) return channel; + if (channel != null) await DisposeChannelAsync(); + + var connection = await busConnection.GetPublisherConnectionAsync(); + channel = await connection.CreateChannelAsync(new CreateChannelOptions( + publisherConfirmationsEnabled: busOptions.PublisherConfirms, + publisherConfirmationTrackingEnabled: busOptions.PublisherConfirms)); + return channel; + } + finally + { + channelLock.Release(); + } + } - return Task.CompletedTask; + private async Task DisposeChannelAsync() + { + var closing = channel; + channel = null; + if (closing == null) return; + try + { + if (closing.IsOpen) await closing.CloseAsync(); + } + catch (Exception ex) + { + Trace.TraceWarning("Failed to close a publisher channel: {0}", ex.Message); + } + finally + { + await closing.DisposeAsync(); + } } + public async ValueTask DisposeAsync() => await DisposeChannelAsync(); + + // Scopes disposed synchronously (a plain "using var scope = CreateScope()") land here. + public void Dispose() => DisposeChannelAsync().GetAwaiter().GetResult(); + private void FireAndForget(IOperationalEvent evt) { operationalEventPublisher.Publish(evt, CancellationToken.None) diff --git a/SW.Bus/BusConnection.cs b/SW.Bus/BusConnection.cs new file mode 100644 index 0000000..e88f42f --- /dev/null +++ b/SW.Bus/BusConnection.cs @@ -0,0 +1,275 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using RabbitMQ.Client; +using SW.Bus.RabbitMqExtensions; + +namespace SW.Bus; + +/// Client names the broker shows for the bus's connections. +internal sealed class BusConnectionNames +{ + public string Publisher { get; set; } + public string Consumer { get; set; } +} + +/// +/// The bus's connections to the broker: one for publishing and one for consuming, so the broker's +/// flow control on a busy publisher never holds back consumer acknowledgements. Both are opened +/// asynchronously, never inside service registration, and the client's automatic recovery owns +/// reconnects and topology recovery (exchanges, queues, bindings and consumers) after an outage. +/// +/// +/// As a hosted service it declares the exchanges at start, retrying in the background until the +/// broker answers, so a host whose broker is down at boot still starts and connects later. +/// +internal sealed class BusConnection : IHostedService, IBusConnectionStatus, IAsyncDisposable, IDisposable +{ + /// + /// How long a close waits. The client's channel and connection close wait for consumer handlers + /// still running; past the drain the broker has requeued their messages, so nothing is gained by + /// waiting longer, and a hung handler must not hold the host's shutdown. + /// + internal static readonly TimeSpan CloseTimeout = TimeSpan.FromSeconds(1); + + private readonly BusOptions busOptions; + private readonly BusConnectionNames names; + private readonly ILogger logger; + private readonly ConnectionFactory publisherFactory; + private readonly ConnectionFactory consumerFactory; + private readonly SemaphoreSlim connectLock = new(1, 1); + private readonly TaskCompletionSource exchangesDeclared = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly CancellationTokenSource stopping = new(); + + private IConnection publisherConnection; + private IConnection consumerConnection; + private Task startTask = Task.CompletedTask; + + public BusConnection(BusOptions busOptions, BusConnectionNames names, ILogger logger) + { + this.busOptions = busOptions; + this.names = names; + this.logger = logger; + var uri = new Uri(busOptions.ConnectionString); + + // The publisher keeps the client's default heartbeat, the consumer takes HeartBeatTimeOut, + // as the two connections did on client 6. + publisherFactory = new ConnectionFactory + { + Uri = uri, + AutomaticRecoveryEnabled = true, + TopologyRecoveryEnabled = true + }; + consumerFactory = new ConnectionFactory + { + Uri = uri, + AutomaticRecoveryEnabled = true, + TopologyRecoveryEnabled = true, + RequestedHeartbeat = busOptions.RequestedHeartbeat + }; + } + + public bool IsConnected => + exchangesDeclared.Task.IsCompletedSuccessfully && + publisherConnection is { IsOpen: true } && + consumerConnection is null or { IsOpen: true }; + + public string LastError { get; private set; } + + /// + /// Set by the consumer service when it stopped with handlers still running past the drain. Their + /// messages are already requeued by the broker; closing the consumer connection would only wait + /// for the handlers, so it closes in the background instead. + /// + internal bool ConsumerHandlersStillRunning { get; set; } + + /// Completes once the exchanges are declared; consumer queues bind to them. + public Task WaitForExchangesAsync(CancellationToken cancellationToken) => + exchangesDeclared.Task.WaitAsync(cancellationToken); + + public Task GetPublisherConnectionAsync(CancellationToken cancellationToken = default) => + GetConnectionAsync(publisher: true, cancellationToken); + + public Task GetConsumerConnectionAsync(CancellationToken cancellationToken = default) => + GetConnectionAsync(publisher: false, cancellationToken); + + /// + /// The connection, opened on first use. Once opened, the same connection is returned for the life + /// of the bus, also while the client is recovering it: channels created on it recover with it. + /// Throws when the broker cannot be reached and the connection was never opened. + /// + private async Task GetConnectionAsync(bool publisher, CancellationToken cancellationToken) + { + var existing = publisher ? publisherConnection : consumerConnection; + if (existing != null) return existing; + + await connectLock.WaitAsync(cancellationToken); + try + { + existing = publisher ? publisherConnection : consumerConnection; + if (existing != null) return existing; + + var factory = publisher ? publisherFactory : consumerFactory; + var connection = await factory.CreateConnectionAsync( + publisher ? names.Publisher : names.Consumer, cancellationToken); + connection.ConnectionShutdownAsync += (_, args) => + { + if (args.Initiator != ShutdownInitiator.Application) + logger.LogWarning("RabbitMQ {Connection} connection lost, the client is recovering it. {Cause}", + publisher ? "publisher" : "consumer", args.ReplyText); + return Task.CompletedTask; + }; + connection.RecoverySucceededAsync += (_, _) => + { + logger.LogInformation("RabbitMQ {Connection} connection recovered.", publisher ? "publisher" : "consumer"); + return Task.CompletedTask; + }; + connection.ConnectionRecoveryErrorAsync += (_, args) => + { + LastError = args.Exception.Message; + logger.LogWarning(args.Exception, "RabbitMQ {Connection} connection recovery attempt failed.", + publisher ? "publisher" : "consumer"); + return Task.CompletedTask; + }; + + if (publisher) publisherConnection = connection; + else consumerConnection = connection; + return connection; + } + finally + { + connectLock.Release(); + } + } + + public Task StartAsync(CancellationToken cancellationToken) + { + // Not awaited: the host starts whether or not the broker is up. + startTask = Task.Run(() => DeclareExchangesAsync(stopping.Token), CancellationToken.None); + return Task.CompletedTask; + } + + private async Task DeclareExchangesAsync(CancellationToken cancellationToken) + { + var retryCount = 0; + while (!cancellationToken.IsCancellationRequested) + { + try + { + var connection = await GetPublisherConnectionAsync(cancellationToken); + await using (var channel = await connection.CreateChannelAsync(cancellationToken: cancellationToken)) + { + await channel.ExchangeDeclareAsync(busOptions.ProcessExchange, ExchangeType.Direct, true, + cancellationToken: cancellationToken); + await channel.ExchangeDeclareAsync(busOptions.DeadLetterExchange, ExchangeType.Direct, true, + cancellationToken: cancellationToken); + await channel.ExchangeDeclareAsync(busOptions.NodeExchange, ExchangeType.Direct, true, + cancellationToken: cancellationToken); + await channel.ExchangeDeclareAsync(busOptions.NodeDeadLetterExchange, ExchangeType.Direct, true, + cancellationToken: cancellationToken); + await channel.CloseAsync(cancellationToken); + } + + LastError = null; + exchangesDeclared.TrySetResult(); + return; + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + return; + } + catch (Exception ex) + { + LastError = ex.Message; + retryCount++; + var delay = Math.Min(retryCount * 2, 30); + logger.LogError(ex, "Failed to connect to RabbitMQ and declare the bus exchanges. Retrying in {Delay}s.", delay); + try + { + await Task.Delay(TimeSpan.FromSeconds(delay), cancellationToken); + } + catch (OperationCanceledException) + { + return; + } + } + } + } + + /// + /// Runs after the consumers have drained (hosted services stop in reverse order of registration, + /// and this one is registered first), so a message finishing during the drain can still publish. + /// + public async Task StopAsync(CancellationToken cancellationToken) + { + await stopping.CancelAsync(); + try + { + await startTask; + } + catch (Exception ex) + { + logger.LogDebug(ex, "Exchange declaration ended with an error during stop."); + } + + if (ConsumerHandlersStillRunning) InBackground(consumerConnection, c => c.CloseAsync(), "close"); + else await CloseAsync(consumerConnection, cancellationToken); + await CloseAsync(publisherConnection, cancellationToken); + } + + private async Task CloseAsync(IConnection connection, CancellationToken cancellationToken) + { + if (connection == null) return; + await Bounded(async () => + { + if (connection.IsOpen) await connection.CloseAsync(cancellationToken); + }, "close", cancellationToken); + } + + private void InBackground(IConnection connection, Func action, string what) + { + if (connection == null) return; + _ = Task.Run(() => action(connection)).ContinueWith( + t => logger.LogDebug(t.Exception, "RabbitMQ connection {What} failed.", what), + TaskContinuationOptions.OnlyOnFaulted); + } + + private async Task Bounded(Func action, string what, CancellationToken cancellationToken = default) + { + var task = Task.Run(action, CancellationToken.None); + var finished = await Task.WhenAny(task, Task.Delay(CloseTimeout, cancellationToken)).ConfigureAwait(false); + if (finished != task) + { + logger.LogWarning("A RabbitMQ connection {What} did not finish in {Timeout}; a consumer handler is still running.", + what, CloseTimeout); + _ = task.ContinueWith(t => logger.LogDebug(t.Exception, "RabbitMQ connection {What} failed.", what), + TaskContinuationOptions.OnlyOnFaulted); + return; + } + + try + { + await task; + } + catch (Exception ex) + { + logger.LogWarning(ex, "Failed to {What} a RabbitMQ connection.", what); + } + } + + public async ValueTask DisposeAsync() + { + var consumer = consumerConnection; + var publisher = publisherConnection; + consumerConnection = null; + publisherConnection = null; + if (ConsumerHandlersStillRunning) InBackground(consumer, c => c.DisposeAsync().AsTask(), "dispose"); + else if (consumer != null) await Bounded(() => consumer.DisposeAsync().AsTask(), "dispose"); + if (publisher != null) await Bounded(() => publisher.DisposeAsync().AsTask(), "dispose"); + stopping.Dispose(); + } + + public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult(); +} diff --git a/SW.Bus/BusOptions.cs b/SW.Bus/BusOptions.cs index 660398f..b4b6624 100644 --- a/SW.Bus/BusOptions.cs +++ b/SW.Bus/BusOptions.cs @@ -170,6 +170,9 @@ public int MonitoringCacheSeconds /// public string VirtualHost { get; set; } + /// The 'RabbitMQ' connection string, read by AddBus. + internal string ConnectionString { get; set; } + internal TimeSpan RequestedHeartbeat => HeartBeatTimeOut > 0 ? TimeSpan.FromSeconds(HeartBeatTimeOut) : TimeSpan.Zero; @@ -204,6 +207,35 @@ public int MonitoringCacheSeconds /// Can be overridden per queue using . /// public ushort DefaultQueuePrefetch { get; set; } + + /// + /// Gets or sets how many messages of one queue a node processes at the same time, for queues + /// that do not set their own ( or ). + /// Default: 1, one at a time in the order they arrive. + /// + /// + /// Each queue has its own channel, and this is that channel's consumer dispatch concurrency. + /// The effective concurrency is the smaller of this and the queue's prefetch, because the + /// broker never sends a node more than prefetch unacknowledged messages. Above 1, messages + /// of the queue are not processed in order and the consumer must be safe to run in parallel. + /// + public ushort DefaultConsumerConcurrency { get; set; } = 1; + + /// + /// Gets or sets whether publishes wait for the broker to confirm it has the message. + /// When true, and complete only once the broker + /// has taken responsibility for the message, and throw when it refuses it or cannot be reached. + /// Default: true. + /// + public bool PublisherConfirms { get; set; } = true; + + /// + /// Gets or sets how long stopping the bus waits for messages already being processed to finish + /// and be acknowledged, after the consumers stop receiving new ones. Messages still running + /// when it expires are returned to their queue when the channel closes. + /// Default: 30 seconds. The host's own shutdown timeout also bounds it. + /// + public TimeSpan DrainTimeout { get; set; } = TimeSpan.FromSeconds(30); /// /// Gets or sets the default number of retry attempts for failed messages. @@ -302,6 +334,7 @@ public int MonitoringCacheSeconds /// The number of retry attempts for this queue. If null, uses . /// The delay in seconds before retrying. If null, uses . /// The maximum priority level for this queue. If null, uses . + /// How many messages of this queue a node processes at the same time. If null, uses . /// /// /// services.AddBus(options => @@ -314,14 +347,16 @@ public int MonitoringCacheSeconds /// }); /// /// - public void AddQueueOption(string queueName, ushort? prefetch = null, int? retryCount = null, uint? retryAfterSeconds = null,int? priority = null) + public void AddQueueOption(string queueName, ushort? prefetch = null, int? retryCount = null, uint? retryAfterSeconds = null,int? priority = null, + ushort? concurrency = null) { Options[queueName.ToLower()] = new QueueOptions { Prefetch = prefetch, RetryCount = retryCount, RetryAfterSeconds = retryAfterSeconds, - Priority = priority + Priority = priority, + Concurrency = concurrency }; } diff --git a/SW.Bus/ConsumerDefinition.cs b/SW.Bus/ConsumerDefinition.cs index a64c0c5..731f3dd 100644 --- a/SW.Bus/ConsumerDefinition.cs +++ b/SW.Bus/ConsumerDefinition.cs @@ -29,6 +29,8 @@ public ConsumerDefinition(string queueNamePrefix, BusOptions busOptions, Consume queueOptions.Prefetch = consumerOptions.Prefetch; if (consumerOptions.Priority.HasValue) queueOptions.Priority = consumerOptions.Priority; + if (consumerOptions.Concurrency.HasValue) + queueOptions.Concurrency = consumerOptions.Concurrency; } public Type ServiceType { get; set; } @@ -38,6 +40,7 @@ public ConsumerDefinition(string queueNamePrefix, BusOptions busOptions, Consume public int RetryCount => queueOptions?.RetryCount ?? busOptions.DefaultRetryCount; public uint RetryAfter => queueOptions?.RetryAfterSeconds ?? busOptions.DefaultRetryAfter; public ushort QueuePrefetch => queueOptions?.Prefetch ?? busOptions.DefaultQueuePrefetch; + public ushort Concurrency => Math.Max((ushort)1, queueOptions?.Concurrency ?? busOptions.DefaultConsumerConcurrency); public string NakedQueueName { get; private set; } public string QueueName => $"{queueNamePrefix}.{NakedQueueName}".ToLower(); public string RoutingKey => MessageTypeName.ToLower(); diff --git a/SW.Bus/ConsumerRunner.cs b/SW.Bus/ConsumerRunner.cs index 7a6274a..0ad6e69 100644 --- a/SW.Bus/ConsumerRunner.cs +++ b/SW.Bus/ConsumerRunner.cs @@ -39,7 +39,7 @@ public ConsumerRunner( this.logger = logger; } - internal async Task Run(BasicDeliverEventArgs ea, ConsumerDefinition consumerDefinition, IModel model) + internal async Task Run(BasicDeliverEventArgs ea, ConsumerDefinition consumerDefinition, IChannel model) { var remainingRetryCount = consumerDefinition.RetryCount; @@ -107,7 +107,7 @@ internal async Task Run(BasicDeliverEventArgs ea, ConsumerDefinition consumerDef await (Task)consumerDefinition.Method.Invoke(svc, new[] { messageObject }); } - model.BasicAck(ea.DeliveryTag, false); + await model.BasicAckAsync(ea.DeliveryTag, false); stopwatch.Stop(); metrics.ProcessingCompleted.Add(1); metrics.ProcessingLatencyMs.Record(stopwatch.Elapsed.TotalMilliseconds); @@ -161,7 +161,7 @@ internal async Task Run(BasicDeliverEventArgs ea, ConsumerDefinition consumerDef if (remainingRetryCount != 0 && ex is not InvalidUserTokenException) { // reject the message, will be sent to wait queue - model.BasicReject(ea.DeliveryTag, false); + await model.BasicRejectAsync(ea.DeliveryTag, false); metrics.RetryScheduled.Add(1); FireAndForget(new MessageRetryScheduled( DateTime.UtcNow, @@ -189,7 +189,7 @@ Total retries configured {consumerDefinition.RetryCount}. } else { - model.BasicAck(ea.DeliveryTag, false); + await model.BasicAckAsync(ea.DeliveryTag, false); logger.LogError(ex, @$"Failed to process message '{consumerDefinition.MessageTypeName}', in '{consumerDefinition.ServiceType.Name}'. Message {message}, Total retries {consumerDefinition.RetryCount}"); @@ -216,7 +216,7 @@ Total retries configured {consumerDefinition.RetryCount}. } } } - internal async Task RunNodeMessage(BasicDeliverEventArgs ea, IModel model, + internal async Task RunNodeMessage(BasicDeliverEventArgs ea, IChannel model, ICollection listeners, Func refreshConsumers) { var remainingRetryCount = busOptions.ListenRetryCount; @@ -277,7 +277,7 @@ internal async Task RunNodeMessage(BasicDeliverEventArgs ea, IModel model, } } - model.BasicAck(ea.DeliveryTag, false); + await model.BasicAckAsync(ea.DeliveryTag, false); stopwatch.Stop(); metrics.ProcessingCompleted.Add(1); metrics.ProcessingLatencyMs.Record(stopwatch.Elapsed.TotalMilliseconds); @@ -289,7 +289,7 @@ internal async Task RunNodeMessage(BasicDeliverEventArgs ea, IModel model, if (remainingRetryCount != 0 && ex is not InvalidUserTokenException) { // reject the message, will be sent to wait queue - model.BasicReject(ea.DeliveryTag, false); + await model.BasicRejectAsync(ea.DeliveryTag, false); metrics.RetryScheduled.Add(1); await RunOnFail(svc, failMethod, ex, message); logger.LogWarning(ex, @@ -301,7 +301,7 @@ Total retries configured {busOptions.ListenRetryCount}. } else { - model.BasicAck(ea.DeliveryTag, false); + await model.BasicAckAsync(ea.DeliveryTag, false); logger.LogError(ex, @$"Failed to process message '{listenerDefinition?.MessageTypeName} ?? ?? message', in '{listenerDefinition?.ServiceType.Name ?? "reloading"}'. @@ -351,7 +351,7 @@ private async Task RunOnFail(object svc, MethodInfo failMethod, Exception ex, st } } - void TryBuildBusRequestContext(IServiceProvider serviceProvider,string messageName, IBasicProperties basicProperties, + void TryBuildBusRequestContext(IServiceProvider serviceProvider,string messageName, IReadOnlyBasicProperties basicProperties, int remainingRetries) { var remainingRetriesValue = new RequestValue("RemainingRetries", remainingRetries.ToString(), @@ -430,7 +430,7 @@ void TryBuildBusRequestContext(IServiceProvider serviceProvider,string messageNa /// publisher's, null or not text โ€” gives none: the message itself is still good, and failing /// it would only send it to retry. /// - List ReadPublishedValues(IBasicProperties basicProperties, params RequestValue[] reserved) + List ReadPublishedValues(IReadOnlyBasicProperties basicProperties, params RequestValue[] reserved) { var values = new List(); if (basicProperties.Headers == null || @@ -459,16 +459,17 @@ List ReadPublishedValues(IBasicProperties basicProperties, params } - private Task PublishBad(IModel model, ReadOnlyMemory body, IBasicProperties messageProps, + private async Task PublishBad(IChannel model, ReadOnlyMemory body, IReadOnlyBasicProperties messageProps, string exchange, string routingKey, Exception ex) { const string exception = "exception"; - var props = model.CreateBasicProperties(); + var props = new BasicProperties(); props.Headers = new Dictionary(); + // x-dotnet-pub-seq-no is the client's own confirm tracking header, not the publisher's. foreach (var (key, value) in messageProps.Headers?.Where( - h => h.Key != "x-death") ?? new Dictionary()) + h => h.Key != "x-death" && h.Key != Constants.PublishSequenceNumberHeader) ?? new Dictionary()) props.Headers.Add(key, value); // total bad is used in case the message was moved from bad to process (using shovel) and failed again. so we keep history of failures @@ -476,10 +477,8 @@ private Task PublishBad(IModel model, ReadOnlyMemory body, IBasicPropertie props.Headers.Add($"{exception}{totalBad}", SerializeException(ex)); - props.DeliveryMode = 2; - model.BasicPublish(exchange, routingKey, props, body); - - return Task.CompletedTask; + props.DeliveryMode = DeliveryModes.Persistent; + await model.BasicPublishAsync(exchange, routingKey, false, props, body); } // some exceptions (thrown ones carry a TargetSite) are not serializable; the message is already diff --git a/SW.Bus/ConsumersService.cs b/SW.Bus/ConsumersService.cs index 89c75e5..98836b9 100644 --- a/SW.Bus/ConsumersService.cs +++ b/SW.Bus/ConsumersService.cs @@ -16,42 +16,58 @@ internal class ConsumersService : IHostedService private readonly ILogger logger; private readonly BusOptions busOptions; private readonly ConsumerDiscovery consumerDiscovery; - private readonly ConnectionFactory connectionFactory; + private readonly BusConnection busConnection; // Thread-safe dictionary for runtime updates - private readonly ConcurrentDictionary openModels; + private readonly ConcurrentDictionary openModels; + // Consumers taken out of service (removed, or replaced to change their concurrency) whose channel + // closes once the messages they were processing are acknowledged. + private readonly ConcurrentDictionary retiring = new(); private readonly ConsumerRunner consumerRunner; private readonly IOperationalEventPublisher operationalEventPublisher; + private readonly CancellationTokenSource stopping = new(); private IConnection conn; - private IModel nodeModel; + private AttachedConsumer nodeConsumer; + private Task startTask = Task.CompletedTask; + private bool stopped; // Semaphore to prevent overlapping refresh operations private readonly SemaphoreSlim _topologyLock = new(1, 1); public ConsumersService(ILogger logger, BusOptions busOptions, - ConsumerDiscovery consumerDiscovery, ConnectionFactory connectionFactory, ConsumerRunner consumerRunner, + ConsumerDiscovery consumerDiscovery, BusConnection busConnection, ConsumerRunner consumerRunner, IOperationalEventPublisher operationalEventPublisher) { this.logger = logger; this.busOptions = busOptions; this.consumerDiscovery = consumerDiscovery; - this.connectionFactory = connectionFactory; + this.busConnection = busConnection; this.consumerRunner = consumerRunner; this.operationalEventPublisher = operationalEventPublisher; - openModels = new ConcurrentDictionary(); + openModels = new ConcurrentDictionary(); } public Task StartAsync(CancellationToken cancellationToken) { // Offload to background thread to not block Host startup - Task.Run(() => StartBusAsync(cancellationToken), cancellationToken); + startTask = Task.Run(() => StartBusAsync(stopping.Token), CancellationToken.None); return Task.CompletedTask; } private async Task StartBusAsync(CancellationToken cancellationToken) { + try + { + // BusConnection retries until the broker answers; queues bind to these exchanges. + await busConnection.WaitForExchangesAsync(cancellationToken); + } + catch (OperationCanceledException) + { + return; + } + var retryCount = 0; while (!cancellationToken.IsCancellationRequested) { @@ -62,27 +78,29 @@ private async Task StartBusAsync(CancellationToken cancellationToken) { var consumerDefinitions = await consumerDiscovery.Load(); - // 1. Establish Connection - if (conn is not { IsOpen: true }) + // 1. Establish Connection. Opened once; the client recovers it, with its channels, + // queues, bindings and consumers, after an outage, so nothing here re-attaches. + if (conn == null) { - conn = connectionFactory.CreateConnection(); - conn.ConnectionShutdown += ConnectionShutdown; + conn = await busConnection.GetConsumerConnectionAsync(cancellationToken); + conn.ConsumerTagChangeAfterRecoveryAsync += ConsumerTagChangedAfterRecovery; } // 2. Setup Node Listener (Control Channel) - DeclareAndBindListener(); + await DeclareAndBindListener(cancellationToken); // 3. Declare Queues (using a temporary channel to keep consumer channels clean) - using (var model = conn.CreateModel()) + await using (var model = await conn.CreateChannelAsync(cancellationToken: cancellationToken)) { foreach (var c in consumerDefinitions) - DeclareAndBind(model, c); + await DeclareAndBind(model, c, cancellationToken); + await model.CloseAsync(cancellationToken); } // 4. Attach Consumers foreach (var consumerDefinition in consumerDefinitions) { - AttachConsumer(consumerDefinition); + await AttachConsumer(consumerDefinition, cancellationToken); } logger.LogInformation("ConsumersService started successfully."); @@ -93,6 +111,10 @@ private async Task StartBusAsync(CancellationToken cancellationToken) _topologyLock.Release(); } } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + break; + } catch (Exception ex) { retryCount++; @@ -111,25 +133,27 @@ private async Task StartBusAsync(CancellationToken cancellationToken) } } - private void DeclareAndBind(IModel model, ConsumerDefinition c) + // Names, arguments and bindings must stay exactly as they are: the broker refuses to redeclare an + // existing queue with different arguments. + private async Task DeclareAndBind(IChannel model, ConsumerDefinition c, CancellationToken cancellationToken) { logger.LogInformation($"Declaring and binding: {c.QueueName}."); - model.QueueDeclare(c.QueueName, true, false, false, c.ProcessArgs); - model.QueueBind(c.QueueName, busOptions.ProcessExchange, c.RoutingKey, null); - model.QueueBind(c.QueueName, busOptions.ProcessExchange, c.RetryRoutingKey, null); + await model.QueueDeclareAsync(c.QueueName, true, false, false, c.ProcessArgs, cancellationToken: cancellationToken); + await model.QueueBindAsync(c.QueueName, busOptions.ProcessExchange, c.RoutingKey, null, cancellationToken: cancellationToken); + await model.QueueBindAsync(c.QueueName, busOptions.ProcessExchange, c.RetryRoutingKey, null, cancellationToken: cancellationToken); - model.QueueDeclare(c.RetryQueueName, true, false, false, c.RetryArgs); - model.QueueBind(c.RetryQueueName, busOptions.DeadLetterExchange, c.RetryRoutingKey, null); + await model.QueueDeclareAsync(c.RetryQueueName, true, false, false, c.RetryArgs, cancellationToken: cancellationToken); + await model.QueueBindAsync(c.RetryQueueName, busOptions.DeadLetterExchange, c.RetryRoutingKey, null, cancellationToken: cancellationToken); - model.QueueDeclare(c.BadQueueName, true, false, false, ConsumerDefinition.BadArgs); - model.QueueBind(c.BadQueueName, busOptions.DeadLetterExchange, c.BadRoutingKey, null); + await model.QueueDeclareAsync(c.BadQueueName, true, false, false, ConsumerDefinition.BadArgs, cancellationToken: cancellationToken); + await model.QueueBindAsync(c.BadQueueName, busOptions.DeadLetterExchange, c.BadRoutingKey, null, cancellationToken: cancellationToken); } - private void DeclareAndBindListener() + private async Task DeclareAndBindListener(CancellationToken cancellationToken) { - // Ensure we don't double-bind if this is a reconnect - if (nodeModel != null && nodeModel.IsOpen) return; + // Ensure we don't double-bind if this is a retried start; set only once fully attached. + if (nodeConsumer != null) return; logger.LogInformation($"Declaring and binding node queue: {busOptions.NodeQueueName}."); var listeners = consumerDiscovery.LoadListeners(); @@ -144,72 +168,111 @@ private void DeclareAndBindListener() throw new BusException( $"Duplicate node consumers defined for: {string.Join(',', repeated.Select(r => r.MessageType.FullName))}"); - nodeModel = conn.CreateModel(); - - nodeModel.QueueDeclare(busOptions.NodeQueueName, true, true, true, busOptions.NodeProcessArgs); - nodeModel.QueueBind(busOptions.NodeQueueName, busOptions.NodeExchange, busOptions.NodeRoutingKey, null); - nodeModel.QueueBind(busOptions.NodeQueueName, busOptions.NodeExchange, busOptions.NodeRetryRoutingKey, null); - - nodeModel.QueueDeclare(busOptions.NodeRetryQueueName, true, true, true, busOptions.NodeRetryArgs); - nodeModel.QueueBind(busOptions.NodeRetryQueueName, busOptions.NodeDeadLetterExchange, busOptions.NodeRetryRoutingKey, null); - - nodeModel.QueueDeclare(busOptions.NodeBadQueueName, true, false, false, ConsumerDefinition.BadArgs); - nodeModel.QueueBind(busOptions.NodeBadQueueName, busOptions.NodeDeadLetterExchange, busOptions.NodeBadRoutingKey, null); - - var consumer = new AsyncEventingBasicConsumer(nodeModel); - consumer.Received += async (ch, ea) => + var nodeModel = await conn.CreateChannelAsync(new CreateChannelOptions(false, false, + consumerDispatchConcurrency: 1), cancellationToken); + try { - // Pass the RefreshConsumers method as the callback - await consumerRunner.RunNodeMessage(ea, nodeModel, listeners, RefreshConsumers); - }; + await nodeModel.QueueDeclareAsync(busOptions.NodeQueueName, true, true, true, busOptions.NodeProcessArgs, cancellationToken: cancellationToken); + await nodeModel.QueueBindAsync(busOptions.NodeQueueName, busOptions.NodeExchange, busOptions.NodeRoutingKey, null, cancellationToken: cancellationToken); + await nodeModel.QueueBindAsync(busOptions.NodeQueueName, busOptions.NodeExchange, busOptions.NodeRetryRoutingKey, null, cancellationToken: cancellationToken); + + await nodeModel.QueueDeclareAsync(busOptions.NodeRetryQueueName, true, true, true, busOptions.NodeRetryArgs, cancellationToken: cancellationToken); + await nodeModel.QueueBindAsync(busOptions.NodeRetryQueueName, busOptions.NodeDeadLetterExchange, busOptions.NodeRetryRoutingKey, null, cancellationToken: cancellationToken); + + await nodeModel.QueueDeclareAsync(busOptions.NodeBadQueueName, true, false, false, ConsumerDefinition.BadArgs, cancellationToken: cancellationToken); + await nodeModel.QueueBindAsync(busOptions.NodeBadQueueName, busOptions.NodeDeadLetterExchange, busOptions.NodeBadRoutingKey, null, cancellationToken: cancellationToken); + + var attached = new AttachedConsumer(nodeModel, null); + var consumer = new AsyncEventingBasicConsumer(nodeModel); + consumer.ReceivedAsync += async (ch, ea) => + { + if (!attached.InFlight.TryEnter()) return; + try + { + // Pass the RefreshConsumers method as the callback + await consumerRunner.RunNodeMessage(ea, nodeModel, listeners, RefreshConsumers); + } + finally + { + attached.InFlight.Exit(); + } + }; - nodeModel.BasicQos(0, 1, false); - nodeModel.BasicConsume(busOptions.NodeQueueName, false, consumer); + await nodeModel.BasicQosAsync(0, 1, false, cancellationToken); + attached.ConsumerTag = await nodeModel.BasicConsumeAsync(busOptions.NodeQueueName, false, consumer, cancellationToken); + nodeConsumer = attached; + } + catch + { + await nodeModel.DisposeAsync(); + throw; + } } - private void AttachConsumer(ConsumerDefinition consumerDefinition) + private async Task AttachConsumer(ConsumerDefinition consumerDefinition, CancellationToken cancellationToken = default) { if (openModels.ContainsKey(consumerDefinition.QueueName)) return; - var model = conn.CreateModel(); - - var consumer = new AsyncEventingBasicConsumer(model); - consumerDefinition.ConsumerObject = consumer; - - consumer.Shutdown += (ch, args) => + // The channel's dispatch concurrency is how many messages of this queue run at once on this node. + var model = await conn.CreateChannelAsync(new CreateChannelOptions(false, false, + consumerDispatchConcurrency: consumerDefinition.Concurrency), cancellationToken); + var attached = new AttachedConsumer(model, consumerDefinition); + try { - logger.LogWarning($"Consumer {consumerDefinition.QueueName} shutdown. {args}"); - FireAndForget(new ConsumerDisconnected( - DateTime.UtcNow, - busOptions.OperationalEventsSchemaVersion, - Environment.MachineName, - busOptions.EnvironmentName, - busOptions.ApplicationName ?? string.Empty, - busOptions.ProcessExchange, - consumerDefinition.QueueName, - consumerDefinition.ServiceType?.Name ?? string.Empty, - consumerDefinition.MessageTypeName, - string.Empty, - string.Empty, - string.Empty, - string.Empty, - string.Empty, - null, - args?.ReplyText ?? "shutdown")); - return Task.CompletedTask; - }; - - consumer.Received += async (ch, ea) => - { - await consumerRunner.Run(ea, consumerDefinition, model); - }; + var consumer = new AsyncEventingBasicConsumer(model); + consumerDefinition.ConsumerObject = consumer; + + consumer.ShutdownAsync += (ch, args) => + { + logger.LogWarning($"Consumer {consumerDefinition.QueueName} shutdown. {args}"); + FireAndForget(new ConsumerDisconnected( + DateTime.UtcNow, + busOptions.OperationalEventsSchemaVersion, + Environment.MachineName, + busOptions.EnvironmentName, + busOptions.ApplicationName ?? string.Empty, + busOptions.ProcessExchange, + consumerDefinition.QueueName, + consumerDefinition.ServiceType?.Name ?? string.Empty, + consumerDefinition.MessageTypeName, + string.Empty, + string.Empty, + string.Empty, + string.Empty, + string.Empty, + null, + args?.ReplyText ?? "shutdown")); + return Task.CompletedTask; + }; + + consumer.ReceivedAsync += async (ch, ea) => + { + // Once the consumer is stopping, a message it has not started is left unacknowledged + // and returns to the queue when the channel closes. + if (!attached.InFlight.TryEnter()) return; + try + { + await consumerRunner.Run(ea, consumerDefinition, model); + } + finally + { + attached.InFlight.Exit(); + } + }; - model.BasicQos(0, consumerDefinition.QueuePrefetch, false); + await model.BasicQosAsync(0, consumerDefinition.QueuePrefetch, false, cancellationToken); - consumerDefinition.ConsumerTag = model.BasicConsume(consumerDefinition.QueueName, false, "", - consumerDefinition.ConsumerArgs, consumer); + consumerDefinition.ConsumerTag = await model.BasicConsumeAsync(consumerDefinition.QueueName, false, "", + false, false, consumerDefinition.ConsumerArgs, consumer, cancellationToken); + attached.ConsumerTag = consumerDefinition.ConsumerTag; + } + catch + { + await model.DisposeAsync(); + throw; + } - openModels.TryAdd(consumerDefinition.QueueName, (model, consumerDefinition)); + openModels.TryAdd(consumerDefinition.QueueName, attached); FireAndForget(new ConsumerConnected( DateTime.UtcNow, busOptions.OperationalEventsSchemaVersion, @@ -228,39 +291,84 @@ private void AttachConsumer(ConsumerDefinition consumerDefinition) null, consumerDefinition.ConsumerTag)); - TryEmitQueueBackpressure(model, consumerDefinition); + await TryEmitQueueBackpressure(model, consumerDefinition); } - private void DetachConsumer(string queueName) + private async Task DetachConsumer(string queueName) { if (!openModels.TryRemove(queueName, out var open)) return; logger.LogInformation($"Detaching removed consumer: {queueName}"); + await Retire(open); + } + + /// + /// Takes a consumer out of service: it starts no new message, its consumer is cancelled at the + /// broker, and its channel closes once the messages it is processing are acknowledged. Messages + /// prefetched but not started return to the queue when the channel closes. + /// + private async Task Retire(AttachedConsumer open) + { + open.InFlight.Close(); try { // Cancel before closing: it also drops the consumer from the client's recovery records, // so a reconnect doesn't re-attach it. - if (open.model.IsOpen) open.model.BasicCancel(open.consumerDefinition.ConsumerTag); + if (open.Channel.IsOpen) await open.Channel.BasicCancelAsync(open.ConsumerTag); } catch (Exception ex) { - logger.LogWarning(ex, $"Failed to cancel consumer {queueName}"); + logger.LogWarning(ex, $"Failed to cancel consumer {open.Definition?.QueueName}"); } // Closed whatever the cancel did: the entry is already gone, so no later refresh would - // retry, and closing the channel stops the consumer on its own. + // retry, and closing the channel stops the consumer on its own. Registered before it can + // finish, so a stop always sees it until its channel is closed. + var closed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + retiring[open] = closed.Task; + _ = CloseWhenIdle(open, closed); + } + + private async Task CloseWhenIdle(AttachedConsumer open, TaskCompletionSource closed) + { try { - if (open.model.IsOpen) open.model.Close(); + await open.InFlight.WhenIdle(); + await CloseChannel(open); } - catch (Exception ex) + finally { - logger.LogWarning(ex, $"Failed to close the channel of consumer {queueName}"); + retiring.TryRemove(open, out _); + closed.TrySetResult(); } - finally + } + + /// + /// Closes a consumer's channel. The client's close, abort and dispose all wait for a running + /// handler to return, so the close is bounded: the broker closes the channel and requeues its + /// unacknowledged messages at once, and a channel still processing is not disposed here. + /// + private async Task CloseChannel(AttachedConsumer open) + { + var name = open.Definition?.QueueName ?? busOptions.NodeQueueName; + using var bounded = new CancellationTokenSource(BusConnection.CloseTimeout); + try + { + if (open.Channel.IsOpen) await open.Channel.CloseAsync(bounded.Token); + } + catch (OperationCanceledException) + { + logger.LogWarning($"The channel of consumer {name} closed with a message still being processed; it returns to the queue."); + } + catch (Exception ex) { - open.model.Dispose(); + logger.LogWarning(ex, $"Failed to close the channel of consumer {name}"); } + + // Disposing a channel whose handler is still running blocks the closes of other channels + // on the connection until it returns, so such a channel is left to the connection's close. + if (open.InFlight.Count == 0) + await open.Channel.DisposeAsync(); } private async Task RefreshConsumers() @@ -269,17 +377,20 @@ private async Task RefreshConsumers() await _topologyLock.WaitAsync(); try { + if (stopped || conn == null) return; + logger.LogInformation("Refreshing consumers..."); var newDefinitions = await consumerDiscovery.Load(); // 1. Declare any NEW queues - using (var model = conn.CreateModel()) + await using (var model = await conn.CreateChannelAsync()) { foreach (var c in newDefinitions) { if (openModels.ContainsKey(c.QueueName)) continue; - DeclareAndBind(model, c); + await DeclareAndBind(model, c, CancellationToken.None); } + await model.CloseAsync(); } // 2. Update existing or Attach new @@ -288,18 +399,29 @@ private async Task RefreshConsumers() // CASE A: Existing Consumer if (openModels.TryGetValue(def.QueueName, out var existing)) { - var existingModel = existing.model; - var existingDef = existing.consumerDefinition; + var existingModel = existing.Channel; + var existingDef = existing.Definition; var prefetchChanged = existingDef.QueuePrefetch != def.QueuePrefetch; var priorityChanged = existingDef.ConsumerPriority != def.ConsumerPriority; + var concurrencyChanged = existingDef.Concurrency != def.Concurrency; - if (priorityChanged || prefetchChanged) + if (concurrencyChanged) + { + // A channel's dispatch concurrency is fixed when it is created, so the consumer + // moves to a new channel; the old one finishes what it is processing first. + logger.LogInformation($"Concurrency changed for {def.QueueName}. Moving the consumer to a new channel."); + openModels.TryRemove(def.QueueName, out _); + await Retire(existing); + existingDef.UpdateConsumerProps(def); + await AttachConsumer(existingDef); + } + else if (priorityChanged || prefetchChanged) { logger.LogInformation($"Configuration changed for {def.QueueName}. Restarting consumer."); try { - existingModel.BasicCancel(existingDef.ConsumerTag); + await existingModel.BasicCancelAsync(existingDef.ConsumerTag); } catch (Exception ex) { @@ -309,10 +431,11 @@ private async Task RefreshConsumers() // Update properties (including Args) BEFORE re-consuming existingDef.UpdateConsumerProps(def); - existingModel.BasicQos(0, def.QueuePrefetch, false); + await existingModel.BasicQosAsync(0, def.QueuePrefetch, false); - existingDef.ConsumerTag = existingModel.BasicConsume(def.QueueName, - false, "", existingDef.ConsumerArgs, existingDef.ConsumerObject); + existingDef.ConsumerTag = await existingModel.BasicConsumeAsync(def.QueueName, + false, "", false, false, existingDef.ConsumerArgs, existingDef.ConsumerObject); + existing.ConsumerTag = existingDef.ConsumerTag; } else @@ -324,7 +447,7 @@ private async Task RefreshConsumers() else { logger.LogInformation($"Attaching new consumer: {def.QueueName}"); - AttachConsumer(def); + await AttachConsumer(def); } } @@ -333,7 +456,7 @@ private async Task RefreshConsumers() // application to decide about. var defined = newDefinitions.Select(d => d.QueueName).ToHashSet(); foreach (var queueName in openModels.Keys.Where(q => !defined.Contains(q)).ToList()) - DetachConsumer(queueName); + await DetachConsumer(queueName); } catch (Exception ex) { @@ -359,53 +482,95 @@ private bool DictionariesEqual(IDictionary a, IDictionary o.ConsumerTag == args.TagBefore)) + { + open.ConsumerTag = args.TagAfter; + open.Definition.ConsumerTag = args.TagAfter; + } + if (nodeConsumer?.ConsumerTag == args.TagBefore) nodeConsumer.ConsumerTag = args.TagAfter; + return Task.CompletedTask; } + /// + /// Stops receiving, waits up to (and the host's shutdown + /// token) for the messages being processed to finish and be acknowledged, then closes the + /// channels. Messages prefetched but not started, and any still running at the deadline, return + /// to their queue when the channel closes. + /// public async Task StopAsync(CancellationToken cancellationToken) { + await stopping.CancelAsync(); + try + { + await startTask; + } + catch (Exception ex) + { + logger.LogDebug(ex, "Consumer start ended with an error during stop."); + } + + List consumers; await _topologyLock.WaitAsync(cancellationToken); try { - foreach (var (model, _) in openModels.Values) - { - try - { - if (model.IsOpen) model.Close(); - model.Dispose(); - } - catch (Exception ex) - { - logger.LogWarning(ex, "Failed to stop model."); - } - } + // The lock is not held while draining: a node message being drained may itself be a refresh. + stopped = true; + consumers = openModels.Values.ToList(); openModels.Clear(); + if (nodeConsumer != null) consumers.Add(nodeConsumer); + } + finally + { + _topologyLock.Release(); + } - nodeModel?.Dispose(); - if (conn != null) + using var drain = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + drain.CancelAfter(busOptions.DrainTimeout); + + foreach (var open in consumers) + { + open.InFlight.Close(); + try + { + if (open.Channel.IsOpen) await open.Channel.BasicCancelAsync(open.ConsumerTag, false, drain.Token); + } + catch (Exception ex) { - if (conn.IsOpen) conn.Close(); - conn.Dispose(); + logger.LogWarning(ex, $"Failed to cancel consumer {open.Definition?.QueueName ?? busOptions.NodeQueueName}."); } } - finally + + var running = consumers.Sum(c => c.InFlight.Count); + if (running > 0) + logger.LogInformation("Waiting for {Count} message(s) being processed to finish.", running); + try { - _topologyLock.Release(); + await Task.WhenAll(consumers.Select(c => c.InFlight.WhenIdle()).Concat(retiring.Values)) + .WaitAsync(drain.Token); } + catch (OperationCanceledException) + { + logger.LogWarning("Stopped waiting for {Count} message(s) still being processed; they return to their queue.", + consumers.Sum(c => c.InFlight.Count) + retiring.Keys.Sum(c => c.InFlight.Count)); + } + + var all = consumers.Concat(retiring.Keys).ToList(); + await Task.WhenAll(all.Select(CloseChannel)); + retiring.Clear(); + if (all.Any(c => c.InFlight.Count > 0)) busConnection.ConsumerHandlersStillRunning = true; } - private void TryEmitQueueBackpressure(IModel model, ConsumerDefinition consumerDefinition) + private async Task TryEmitQueueBackpressure(IChannel model, ConsumerDefinition consumerDefinition) { if (busOptions.QueueBackpressureThreshold <= 0) return; try { - var depth = model.MessageCount(consumerDefinition.QueueName); + var depth = await model.MessageCountAsync(consumerDefinition.QueueName); if (depth < busOptions.QueueBackpressureThreshold) return; @@ -443,4 +608,64 @@ private void FireAndForget(IOperationalEvent evt) evt.GetType().Name), TaskContinuationOptions.OnlyOnFaulted); } -} \ No newline at end of file + + private sealed class AttachedConsumer(IChannel channel, ConsumerDefinition definition) + { + public IChannel Channel { get; } = channel; + public ConsumerDefinition Definition { get; } = definition; + public string ConsumerTag { get; set; } + public InFlightMessages InFlight { get; } = new(); + } +} + +/// +/// Counts the messages a consumer is processing, and once closed lets no new one start, so a stop or +/// a channel replacement can wait for the running ones to be acknowledged. +/// +internal sealed class InFlightMessages +{ + private readonly object gate = new(); + private int count; + private bool closed; + private TaskCompletionSource idle = NewIdle(completed: true); + + public int Count + { + get { lock (gate) return count; } + } + + public bool TryEnter() + { + lock (gate) + { + if (closed) return false; + if (count++ == 0) idle = NewIdle(completed: false); + return true; + } + } + + public void Exit() + { + lock (gate) + { + if (--count == 0) idle.TrySetResult(); + } + } + + public void Close() + { + lock (gate) closed = true; + } + + public Task WhenIdle() + { + lock (gate) return idle.Task; + } + + private static TaskCompletionSource NewIdle(bool completed) + { + var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + if (completed) source.SetResult(); + return source; + } +} diff --git a/SW.Bus/IServiceCollectionExtensions.cs b/SW.Bus/IServiceCollectionExtensions.cs index 78d64ec..d8f350c 100644 --- a/SW.Bus/IServiceCollectionExtensions.cs +++ b/SW.Bus/IServiceCollectionExtensions.cs @@ -61,11 +61,10 @@ public static IServiceCollection AddBus(this IServiceCollection services, Action throw new BusException("Connection string named 'RabbitMQ' is required."); } - var factory = new ConnectionFactory - { - Uri = new Uri(rabbitUrl), - ClientProvidedName = $"{Assembly.GetCallingAssembly().GetName().Name} Exchange Declarer" - }; + busOptions.ConnectionString = rabbitUrl; + + // Parsed only, for the management API defaults below; the bus connects in BusConnection. + var factory = new ConnectionFactory { Uri = new Uri(rabbitUrl) }; if (string.IsNullOrEmpty(busOptions.ManagementUrl)) { @@ -104,16 +103,17 @@ public static IServiceCollection AddBus(this IServiceCollection services, Action services.TryAddSingleton(); services.TryAddSingleton(); - using var conn = factory.CreateConnection(); - using var model = conn.CreateModel(); - - model.ExchangeDeclare(busOptions.ProcessExchange, ExchangeType.Direct, true); - model.ExchangeDeclare(busOptions.DeadLetterExchange, ExchangeType.Direct, true); - model.ExchangeDeclare(busOptions.NodeExchange, ExchangeType.Direct, true); - model.ExchangeDeclare(busOptions.NodeDeadLetterExchange, ExchangeType.Direct, true); - - model.Close(); - conn.Close(); + // The connections open, and the exchanges are declared, when the host starts, not here, so a + // host whose broker is down still builds and starts. Registered before the consumer service so + // it stops after it: hosted services stop in reverse order, and draining consumers may publish. + services.AddSingleton(new BusConnectionNames + { + Publisher = $"{Assembly.GetCallingAssembly().GetName().Name} Publisher", + Consumer = $"{typeof(IServiceCollectionExtensions).Assembly.GetName().Name} Consumer" + }); + services.AddSingleton(); + services.AddSingleton(sp => sp.GetRequiredService()); + services.AddHostedService(sp => sp.GetRequiredService()); services.AddHttpClient((sp, client) => { @@ -141,23 +141,20 @@ public static IServiceCollection AddBus(this IServiceCollection services, Action /// /// The to add services to. /// The for chaining. + /// + /// Each DI scope publishes on its own channel, opened on the scope's first publish and closed with + /// the scope. With on, a publish completes when the + /// broker has the message and throws when it does not. + /// public static IServiceCollection AddBusPublish(this IServiceCollection services) { var sp = services.BuildServiceProvider(); - var rabbitUrl = sp.GetRequiredService().GetConnectionString("RabbitMQ"); var busOptions = sp.GetRequiredService(); - - var factory = new ConnectionFactory - { - Uri = new Uri(rabbitUrl), - ClientProvidedName = $"{Assembly.GetCallingAssembly().GetName().Name} Publisher" - }; - - var conn = factory.CreateConnection(); - var model = conn.CreateModel(); + var names = sp.GetRequiredService(); + names.Publisher = $"{Assembly.GetCallingAssembly().GetName().Name} Publisher"; return services.AddScoped(serviceProvider => new BasicPublisher( - model, + serviceProvider.GetRequiredService(), serviceProvider.GetRequiredService(), serviceProvider.GetRequiredService(), serviceProvider.GetRequiredService(), @@ -236,24 +233,6 @@ public static IServiceCollection AddBusListen(this IServiceCollection services, private static IServiceCollection AddConsumerService(this IServiceCollection services) { - var clientProvidedName = $"{Assembly.GetCallingAssembly().GetName().Name} Consumer"; - services.AddSingleton(sp => - { - var rabbitUrl = sp.GetRequiredService().GetConnectionString("RabbitMQ"); - - - var factory = new ConnectionFactory - { - //AutomaticRecoveryEnabled = false, - Uri = new Uri(rabbitUrl), - DispatchConsumersAsync = true, - RequestedHeartbeat = sp.GetRequiredService().RequestedHeartbeat, - ClientProvidedName = clientProvidedName - }; - - return factory; - }); - services.AddHostedService(); services.AddSingleton(); return services; diff --git a/SW.Bus/OperationalEventInfrastructure.cs b/SW.Bus/OperationalEventInfrastructure.cs index a099949..7e675ab 100644 --- a/SW.Bus/OperationalEventInfrastructure.cs +++ b/SW.Bus/OperationalEventInfrastructure.cs @@ -233,26 +233,26 @@ internal static string HeaderAsString(IDictionary headers, strin return value.ToString(); } - internal static string GetMessageId(RabbitMQ.Client.IBasicProperties props) + internal static string GetMessageId(RabbitMQ.Client.IReadOnlyBasicProperties props) => props?.MessageId ?? HeaderAsString(props?.Headers, "Id") ?? string.Empty; - internal static string GetCorrelationId(RabbitMQ.Client.IBasicProperties props) + internal static string GetCorrelationId(RabbitMQ.Client.IReadOnlyBasicProperties props) => props?.CorrelationId ?? HeaderAsString(props?.Headers, "X-Correlation-Id") ?? string.Empty; - internal static string GetCausationId(RabbitMQ.Client.IBasicProperties props) + internal static string GetCausationId(RabbitMQ.Client.IReadOnlyBasicProperties props) => HeaderAsString(props?.Headers, CausationIdHeader) ?? string.Empty; - internal static string GetTraceId(RabbitMQ.Client.IBasicProperties props) + internal static string GetTraceId(RabbitMQ.Client.IReadOnlyBasicProperties props) => HeaderAsString(props?.Headers, TraceIdHeader) ?? string.Empty; - internal static string GetSpanId(RabbitMQ.Client.IBasicProperties props) + internal static string GetSpanId(RabbitMQ.Client.IReadOnlyBasicProperties props) => HeaderAsString(props?.Headers, SpanIdHeader) ?? string.Empty; - internal static ActivityContext ExtractParentContext(RabbitMQ.Client.IBasicProperties props) + internal static ActivityContext ExtractParentContext(RabbitMQ.Client.IReadOnlyBasicProperties props) { var traceParent = HeaderAsString(props?.Headers, TraceParentHeader); if (string.IsNullOrWhiteSpace(traceParent)) diff --git a/SW.Bus/SW.Bus.csproj b/SW.Bus/SW.Bus.csproj index ce15278..94a44f2 100644 --- a/SW.Bus/SW.Bus.csproj +++ b/SW.Bus/SW.Bus.csproj @@ -3,7 +3,7 @@ net10.0 true - 10.0.0.0 + 11.0.0.0 $(NoWarn);1591 SimplyWorks.Bus SimplyWorks.Bus @@ -28,7 +28,7 @@ - + From 4a6034b9c7c4dce5bc0fd4c5524bb72ab5d029a3 Mon Sep 17 00:00:00 2001 From: Samerz Date: Wed, 14 Oct 2026 16:34:32 +0300 Subject: [PATCH 2/2] Version 10.1 rather than 11.0: publish as a minor release --- .github/workflows/nuget-publish.yml | 4 ++-- README.md | 6 +++--- SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj | 2 +- SW.Bus/SW.Bus.csproj | 2 +- 4 files changed, 7 insertions(+), 7 deletions(-) diff --git a/.github/workflows/nuget-publish.yml b/.github/workflows/nuget-publish.yml index 0cf876d..044e69a 100644 --- a/.github/workflows/nuget-publish.yml +++ b/.github/workflows/nuget-publish.yml @@ -22,8 +22,8 @@ jobs: SW.Bus/SW.Bus.csproj SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj SW.Bus.RabbitMqViewer/SW.Bus.RabbitMqViewer.csproj - major-version: '11' - minor-version: '0' + major-version: '10' + minor-version: '1' dotnet-version: '10.0.x' run-tests: false test-projects: '**/*UnitTests/*.csproj' diff --git a/README.md b/README.md index 24be41f..9b24149 100644 --- a/README.md +++ b/README.md @@ -40,7 +40,7 @@ The library ships as three complementary NuGet packages: 18. [Testing](#testing) 19. [Dependencies](#dependencies) 20. [Concurrency, Publisher Confirms and Shutdown](#concurrency-publisher-confirms-and-shutdown) -21. [Upgrading from 10.x to 11.0](#upgrading-from-10x-to-110) +21. [Upgrading to 10.1](#upgrading-to-101) ## Features @@ -1098,9 +1098,9 @@ Messages prefetched but not started, and any still running when the drain times --- -## Upgrading from 10.x to 11.0 +## Upgrading to 10.1 -11.0 moves the bus to RabbitMQ.Client 7. The public interfaces (`IPublish`, `IPublishWithValues`, `IBroadcast`, `IConsume`, `IConsume`, `IConsumeExtended`, `IListen`) 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 an 11.0 node redeclares an existing 10.x topology without the broker refusing it, and 10.x and 11.0 nodes can run side by side during a rollout. +10.1 moves the bus to RabbitMQ.Client 7. The public interfaces (`IPublish`, `IPublishWithValues`, `IBroadcast`, `IConsume`, `IConsume`, `IConsumeExtended`, `IListen`) 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: diff --git a/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj b/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj index 46a6b1d..c62ff68 100644 --- a/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj +++ b/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj @@ -2,7 +2,7 @@ net10.0 - 11.0.0.0 + 10.1.0.0 true $(NoWarn);1591 enable diff --git a/SW.Bus/SW.Bus.csproj b/SW.Bus/SW.Bus.csproj index 94a44f2..79b5651 100644 --- a/SW.Bus/SW.Bus.csproj +++ b/SW.Bus/SW.Bus.csproj @@ -3,7 +3,7 @@ net10.0 true - 11.0.0.0 + 10.1.0.0 $(NoWarn);1591 SimplyWorks.Bus SimplyWorks.Bus