feat(rabbitmq): pool RabbitMQ channels in RabbitMqMessageTransport - #675
Conversation
Replace the single lazily-created, publish-serialized channel in RabbitMqMessageTransport with a pooled IRabbitMqChannelPool / RabbitMqChannelPool backed by a ConcurrentQueue of idle channels and a SemaphoreSlim capped at the new RabbitMqTransportOptions.MaxChannelPoolSize (default 10). SendAsync rents a channel per call; SendBatchAsync rents a single channel for the whole batch and publishes sequentially on it, since only one thread ever touches that channel. Both always return the channel in a finally block. IsHealthyAsync now delegates to the pool. The channel pool is registered as a singleton via TryAddSingleton in UseRabbitMqTransport so repeated calls do not duplicate it.
|
Important Review skippedAuto reviews are limited based on label configuration. 🏷️ Required labels (at least one) (1)
Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
…sk conversion RentAsync_ConcurrentCalls_AreCappedAtMaxChannelPoolSize converts a fresh ValueTask returned by RentAsync to a Task exactly once per loop iteration, but SonarAnalyzer's cross-iteration analysis cannot tell the instances apart and flags a false double-consumption.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #675 +/- ##
==========================================
- Coverage 96.07% 96.05% -0.02%
==========================================
Files 172 188 +16
Lines 7179 7373 +194
Branches 661 679 +18
==========================================
+ Hits 6897 7082 +185
- Misses 136 145 +9
Partials 146 146 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Adds coverage for RentAsync releasing its rental slot when channel creation fails, and for resolving IRabbitMqChannelPool from a built service provider, raising patch coverage to the required threshold.
…lasses (#676) * feat: add IValidateOptions and ValidateOnStart for existing options classes Add IValidateOptions<T> validators for TimeoutRequestInterceptorOptions, QueryCachingOptions, OutboxOptions, OutboxProcessorOptions, AzureServiceBusTransportOptions, RabbitMqTransportOptions, and DaprMessageTransportOptions, each registered with AddOptions<T>().ValidateOnStart() so misconfiguration is caught at startup instead of at first use. The AzureServiceBus transport's imperative ValidateOptions() check is replaced by AzureServiceBusTransportOptionsValidator with equivalent behavior, now surfaced as an OptionsValidationException instead of an InvalidOperationException. LoggingInterceptorOptions is intentionally left untouched: it already has a validator but is out of scope for this change. SQLiteOutboxOptions does not exist as a distinct type; the SQLite provider reuses the shared OutboxOptions, which is already covered by OutboxOptionsValidator (TableName not empty). ConnectionString remains unvalidated there since it is legitimately null for EF Core-based outbox usage. * fix(outbox): reword test comment to avoid false-positive S125 match SonarAnalyzer flagged the explanatory comment as commented-out code because it contained a code-like fragment; reworded in prose only. * feat: IConfigureOptions with IConfiguration binding for existing options classes (#677) * feat: bind options from configuration via IConfigureOptions Add IConfigureOptions<TOptions> implementations that bind LoggingInterceptorOptions, TimeoutRequestInterceptorOptions, QueryCachingOptions, OutboxOptions, OutboxProcessorOptions, AzureServiceBusTransportOptions, RabbitMqTransportOptions, and DaprMessageTransportOptions from documented Pulse:* configuration sections, registered inside the respective existing Add*/Use* extension methods so IConfiguration-backed values are validated at startup by the #238 validators. AddRequestTimeout only applies its explicit globalTimeout parameter when a value is provided, so a configuration-bound GlobalTimeout is no longer unconditionally overwritten by the method's default null argument. * fix(rabbitmq): register IConfiguration in channel pool resolution test Resolving IRabbitMqChannelPool now requires IConfiguration to be resolvable, since RabbitMqTransportOptionsConfiguration (added by the IConfigureOptions binding work) depends on it.
Resolving IOptions<RabbitMqTransportOptions> now requires IConfiguration to be resolvable, since RabbitMqTransportOptionsConfiguration depends on it; the raw ServiceCollection built by this integration test did not register one.
Summary
IRabbitMqChannelPool/RabbitMqChannelPoolinNetEvolve.Pulse.Internals, backed by aConcurrentQueue<IRabbitMqChannelAdapter>of idle channels and aSemaphoreSlimcapped atMaxChannelPoolSizeconcurrent rentals.RentAsyncdequeues an open idle channel, disposing and skipping any closed one, or creates a new channel viaIRabbitMqConnectionAdapterwhen none are available;Returnre-queues an open channel or disposes a closed one, always releasing the rental slot exactly once (including when the pool itself has been disposed while the channel was rented out).RabbitMqTransportOptions.MaxChannelPoolSize(default10).RabbitMqMessageTransportto rent a channel from the pool perSendAsynccall and to rent a single channel for the wholeSendBatchAsyncbatch (published sequentially on it), always returning the channel in afinallyblock. This removes the old single shared_channelfield,_initializationLock, and_publishLock— per-channel publish safety is now inherent because each rental is exclusive.IsHealthyAsyncnow delegates to the pool's own health check.IRabbitMqChannelPool→RabbitMqChannelPoolas a singleton viaTryAddSingletoninUseRabbitMqTransport, so calling it more than once does not duplicate the registration.IRabbitMqChannelPoolinstead ofIRabbitMqConnectionAdapter) and addRabbitMqChannelPoolTestscovering: concurrent rent capped atMaxChannelPoolSize, returning an open channel makes it available for reuse, returning a closed channel disposes it and the next rent creates a fresh one, pool exhaustion causing a rent to wait until a return happens, and dispose (idempotent, disposes all idle channels,Return/RentAsyncbehave safely after dispose).IsHealthyAsyncbefore any channel exists now correctly reports healthy based on the connection/pool, not on the previous single-field-existence quirk).Closes #241
Deviations from the issue text
IRabbitMqChannelAdapter/IRabbitMqConnectionAdaptertypes were reused as-is, matching the codebase context already given).SendBatchAsyncrents one channel for the entire batch and publishes its messages sequentially on it, rather than renting a channel per message. This was called out as an explicit choice in the issue ("pick whichever keeps the implementation simplest and correct"); renting once per batch avoids rent/return churn for what is inherently a single-threaded, ordered sequence of publishes.RabbitMqMessageTransport's constructor now takesIRabbitMqChannelPoolinstead ofIRabbitMqConnectionAdapter, since the channel pool fully owns channel creation/health against the connection. This is a necessary consequence of removing the transport's private channel-management, and is reflected in the DI registration and the updated unit tests.IsHealthyAsyncno longer requires that a channel has already been created before reporting healthy (the old code returnedfalseuntil the firstSendAsync/SendBatchAsynccall because a_channelfield had not been set yet). With pooling there is no single persistent "the channel" to check, so health now reflects the pool/connection state directly. One integration test (IsHealthyAsync_Before_first_send_returns_false) was updated toIsHealthyAsync_Before_first_send_returns_true_when_connection_opento reflect this intentional, documented behavior change.Test plan
dotnet build Pulse.slnx— builds clean (net8.0/net9.0/net10.0)dotnet test Tests/NetEvolve.Pulse.Tests.Unit --treenode-filter "/*/*/*RabbitMq*/*"— 156 passed, 0 failed, 0 skipped (net8.0/net9.0/net10.0)csharpier format .run before committingNetEvolve.Pulse.Tests.Integration) require Docker/Testcontainers and were not run in this environment, but the project builds successfully with the updated pool-basedCreateTransporthelper.