Make message consumers pausable through a restart-safe RestartableBackgroundService and ITransport.StopAsync, and run the outbox relay on the same base - #429
Merged
Conversation
…kgroundService and ITransport.StopAsync, and run the outbox relay on the same base
Code reviewNo issues found. Checked for bugs and CLAUDE.md compliance. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Message consumers can now be paused and resumed, the same way the outbox relay already could. This is the first of two PRs for one owner of the test-host scope: the test reset pauses every
IRestartableHostedService, and before this change the RabbitMQ consumers never stopped (ITransporthad onlyStartAsync, and the consumer host's loop ended once the transport had started). On a factory that uses RabbitMQ, a shared host and a per-test host read from the same per-class virtual host, so they compete for the same messages.RestartableBackgroundService(new, public,Vulthil.Extensions.Hosting): a restart-safe base class used in place ofBackgroundService. It is the outbox relay's start/stop code, moved: every start runsExecuteAsyncas a new generation, a stop cancels the generation and waits for it, a start first waits for the previous generation, and a fault is logged and stops the application. One gap is closed on the way: of two overlapping starts, only one now begins a generation (before, both could, and one generation ran with no owner).OutboxBackgroundServicederives from it. The relay behavior does not change; its lifecycle tests move to the base class.ITransport.StopAsync(new, with a default that does nothing): stops consumption and waits for the deliveries being handled. A stopped transport can start again.ConsumerHostedServicederives from the base class and is now restartable. Each generation starts the transport (with the same retry loop as before), waits until it is stopped, then stops the transport. Because the next start waits for the previous generation, a restart never starts the transport while the previous stop still runs. A failed stop is logged and does not stop the application.RabbitMqBus.StopAsynccancels every consumer, waits for running handlers, and closes the consumer channels;StartAsyncthen consumes again with new consumers. Worker disposal now disposes every worker even when one fails, and always empties the list, so a later start never keeps an old worker.Vulthil.Messagingnow referencesVulthil.Extensions.Hosting.StopAsync), andtesting.md(consumers pause during the reset).No breaking change: every API change is an addition, so there are no compatibility suppressions.
Verification
dotnet build Vulthil.SharedKernel.slnx: 0 warnings, 0 errors.tests/Vulthil.Extensions.Hosting.Tests(13 tests, 8 of them moved from the relay tests), 4 more consumer-host tests, 3RabbitMqBusStopTests.Vulthil.IntegrationTests(Docker): 69/69 on net10.0 and net9.0, including the newRabbitMqConsumerPauseIntegrationTestson a real broker: while the consumer host is paused, a published message waits in its queue (ready count 1); after the resume it is delivered once.MessagingIntegrationTests: 13/13 on net10.0.dotnet packpackage validation against 1.2.0 passes forVulthil.Extensions.Hosting,Vulthil.Messaging,Vulthil.Messaging.RabbitMqandVulthil.SharedKernel.Outbox.RabbitMqBus.StopAsyncdoes nothing: 3 unit tests and the real-broker test fail;Backport to v1.0: no