< Summary - AsyncResponse (Release / net8.0+net10.0 / unit+integration)

Information
Class: AsyncResponse.Transports.RabbitMQ.RabbitMqWorkerTransport
Assembly: AsyncResponse.Transports.RabbitMQ
File(s): /_/src/Transports/AsyncResponse.Transports.RabbitMQ/RabbitMqWorkerTransport.cs
Line coverage
97%
Covered lines: 100
Uncovered lines: 3
Coverable lines: 103
Total lines: 239
Line coverage: 97%
Branch coverage
84%
Covered branches: 37
Total branches: 44
Branch coverage: 84%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%22100%
.ctor(...)100%11100%
ValidatePublishOptions(...)100%11100%
get_MaxInFlightDuration()50%22100%
GetChannelAsync()94.44%181895.83%
DisposeDiscardedChannelLaterAsync()100%11100%
DisposeQuietlyAsync()100%1160%
PublishAsync()50%1010100%
DisposeAsync()100%1212100%

File(s)

/_/src/Transports/AsyncResponse.Transports.RabbitMQ/RabbitMqWorkerTransport.cs

#LineLine coverage
 1using Microsoft.Extensions.Options;
 2using System.Diagnostics;
 3using System.Text;
 4using System.Text.Json;
 5
 6namespace AsyncResponse.Transports.RabbitMQ;
 7
 8/// <summary>
 9/// Publishes <see cref="WorkerJobEnvelope"/> messages to a RabbitMQ exchange.
 10/// </summary>
 11/// <remarks>
 12/// The publish channel is created lazily and re-created on demand: a transient broker outage when the
 13/// first job is published no longer permanently breaks the transport (a faulted connect attempt is not
 14/// cached). The channel is opened with publisher confirmations so <see cref="PublishAsync"/> only completes
 15/// once the broker has accepted the message. A single channel is shared across concurrent publishers;
 16/// RabbitMQ.Client v7 tracks each in-flight confirmation independently, so concurrent publishing is safe.
 17/// </remarks>
 18public sealed class RabbitMqWorkerTransport : IWorkerTransport, IWorkerTransportInFlightLimit, IAsyncDisposable
 19{
 20    private readonly RabbitMqAsyncResponseOptions _options;
 21    private readonly IRabbitMqConnectionFactory _connectionFactory;
 22    private readonly TimeSpan _discardedChannelDisposeDelay;
 24023    private readonly SemaphoreSlim _connectionGate = new(1, 1);
 24    private IRabbitMqConnection? _connection;
 25    private IRabbitMqChannel? _channel;
 26    private int _disposeGate;
 27    private bool _disposed;
 28
 29    /// <summary>Runs the RabbitMqWorkerTransport operation.</summary>
 30    public RabbitMqWorkerTransport(IOptions<RabbitMqAsyncResponseOptions> options)
 19631        : this(options, new RabbitMqConnectionFactoryAdapter(options.Value))
 32    {
 19633    }
 34
 24035    internal RabbitMqWorkerTransport(
 24036        IOptions<RabbitMqAsyncResponseOptions> options,
 24037        IRabbitMqConnectionFactory connectionFactory,
 24038        TimeSpan? discardedChannelDisposeDelay = null)
 39    {
 24040        _options = options.Value;
 24041        ValidatePublishOptions(_options);
 22642        _connectionFactory = connectionFactory;
 43        // Long enough for a publisher that took the lock-free fast path with the discarded
 44        // channel to have failed its BasicPublish on it and moved on; tests shorten it.
 22645        _discardedChannelDisposeDelay = discardedChannelDisposeDelay ?? TimeSpan.FromSeconds(30);
 22646    }
 47
 48    private static void ValidatePublishOptions(RabbitMqAsyncResponseOptions options)
 49    {
 24050        _ = RabbitMqOptionsValidator.Required(options.WorkerExchange, nameof(options.WorkerExchange));
 23651        _ = RabbitMqOptionsValidator.Required(options.WorkerQueue, nameof(options.WorkerQueue));
 23252        _ = RabbitMqOptionsValidator.Required(options.WorkerRoutingKey, nameof(options.WorkerRoutingKey));
 22853        RabbitMqOptionsValidator.ValidateConnection(options);
 22854        RabbitMqOptionsValidator.ValidateConsumerTimeout(options);
 22855        AsyncResponseChannelOptions.EnsureTimerBacked(options.ShutdownTimeout, nameof(RabbitMqAsyncResponseOptions), nam
 22656    }
 57
 58    /// <summary>
 59    /// The broker's <c>consumer_timeout</c> as mirrored by
 60    /// <see cref="RabbitMqAsyncResponseOptions.BrokerConsumerTimeout"/>: past it RabbitMQ closes the
 61    /// consumer's channel and requeues the still-unacknowledged delivery while its handler runs.
 62    /// <c>null</c> when that option is <c>null</c>, and when the worker subscriber uses
 63    /// <see cref="RabbitMqAckMode.AckAfterEnqueue"/> — the delivery is acknowledged before its handler
 64    /// starts, so no handler run is ever in flight at the broker.
 65    /// </summary>
 66    public TimeSpan? MaxInFlightDuration
 24667        => _options.WorkerSubscriber.AckMode == RabbitMqAckMode.AckAfterEnqueue
 24668            ? null
 24669            : _options.BrokerConsumerTimeout;
 70
 71    private async Task<IRabbitMqChannel> GetChannelAsync(CancellationToken cancellationToken)
 72    {
 45973        var channel = Volatile.Read(ref _channel);
 45974        if (channel is not null && channel.IsOpen)
 23075            return channel;
 76
 22977        await _connectionGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 78        try
 79        {
 22980            ObjectDisposedException.ThrowIf(_disposed, this);
 22781            if (_channel is not null && _channel.IsOpen)
 082                return _channel;
 83
 84            // A cached-but-closed channel is treated as absent: a protocol error (404/406 after an
 85            // exchange delete/redeclare) closes the channel without failing the publish that comes
 86            // next, so keeping it would cache a dead channel forever. Deliberately dereferenced
 87            // WITHOUT disposing: a concurrent publisher that took the lock-free fast path above may
 88            // still hold this object, and disposing it under that publisher turns its retryable
 89            // AlreadyClosedException into a use-after-dispose. A closed channel holds no broker
 90            // resources — but the client only forgets it when it is disposed, and this connection
 91            // lives as long as the process, so every protocol-error replacement accumulated one
 92            // more recorded channel on it. Dispose the discarded one best-effort, later, off this
 93            // path: by then any fast-path publisher holding it has long failed on it.
 22794            if (_channel is { } discarded)
 695                _ = DisposeDiscardedChannelLaterAsync(discarded, _discardedChannelDisposeDelay);
 22796            _channel = null;
 97
 98            // The connection is replaced only when itself closed: with automatic recovery enabled the
 99            // client object stays open while it reconnects, so this only fires when it is truly dead
 100            // (e.g. AutomaticRecoveryEnabled = false).
 227101            if (_connection is not null && !_connection.IsOpen)
 102            {
 2103                await DisposeQuietlyAsync(_connection).ConfigureAwait(false);
 2104                _connection = null;
 105            }
 106
 107            // ??= only assigns when the await succeeds, so a failed connect leaves _connection null and
 108            // the next publish retries. A successful connection is reused even if channel/topology setup fails.
 227109            _connection ??= await _connectionFactory.CreateConnectionAsync(cancellationToken).ConfigureAwait(false);
 225110            var created = await _connection.CreateChannelAsync(publisherConfirmations: true, cancellationToken).Configur
 111            try
 112            {
 225113                await RabbitMqTopology.EnsureWorkerAsync(created, _options, cancellationToken).ConfigureAwait(false);
 223114            }
 2115            catch
 116            {
 117                // Not cached yet, so nothing else ever disposes it: release the channel here or leak it.
 2118                await DisposeQuietlyAsync(created).ConfigureAwait(false);
 2119                throw;
 120            }
 121
 122            // Publish the channel only once it is fully initialized; if anything above threw, _channel stays
 123            // null so a later publish recreates it instead of awaiting a permanently faulted task.
 223124            _channel = created;
 223125            return created;
 126        }
 127        finally
 128        {
 229129            _connectionGate.Release();
 130        }
 453131    }
 132
 133    private static async Task DisposeDiscardedChannelLaterAsync(IRabbitMqChannel discarded, TimeSpan delay)
 134    {
 6135        await Task.Delay(delay).ConfigureAwait(false);
 6136        await DisposeQuietlyAsync(discarded).ConfigureAwait(false);
 6137    }
 138
 139    private static async ValueTask DisposeQuietlyAsync(IAsyncDisposable resource)
 140    {
 141        try
 142        {
 10143            await resource.DisposeAsync().ConfigureAwait(false);
 10144        }
 0145        catch
 146        {
 147            // Best effort: the broker side of a dead channel/connection is already gone.
 0148        }
 10149    }
 150
 151    /// <summary>Publishes the supplied message.</summary>
 152    public async Task PublishAsync(WorkerJobEnvelope job, CancellationToken cancellationToken = default)
 153    {
 461154        ArgumentNullException.ThrowIfNull(job);
 155
 459156        using var activity = AsyncResponseDiagnostics.StartActivity(
 459157            "asyncresponse.worker.publish",
 459158            ActivityKind.Producer,
 459159            job.CorrelationId);
 459160        activity?.SetTag("asyncresponse.transport", "rabbitmq");
 459161        activity?.SetTag("messaging.system", "rabbitmq");
 459162        activity?.SetTag("messaging.destination.name", _options.WorkerExchange);
 459163        activity?.SetTag("messaging.rabbitmq.routing_key", _options.WorkerRoutingKey);
 459164        AsyncResponseDiagnostics.SetReplyTarget(activity, job.ReplyTarget);
 459165        AsyncResponseDiagnostics.SetWorker(activity, job.Call);
 166
 167        try
 168        {
 459169            var payload = Encoding.UTF8.GetBytes(AsyncResponseJson.Serialize(job));
 459170            var properties = RabbitMqTopology.CreatePersistentJsonProperties(job.CorrelationId, _options.CorrelationIdHe
 459171            var channel = await GetChannelAsync(cancellationToken).ConfigureAwait(false);
 453172            await channel.BasicPublishAsync(
 453173                _options.WorkerExchange,
 453174                _options.WorkerRoutingKey,
 453175                properties,
 453176                payload,
 453177                cancellationToken).ConfigureAwait(false);
 451178            activity?.SetTag("messaging.message.id", properties.MessageId);
 451179        }
 8180        catch (Exception ex)
 181        {
 8182            AsyncResponseDiagnostics.SetError(activity, ex);
 8183            throw;
 184        }
 451185    }
 186
 187    /// <summary>Releases resources held by this instance.</summary>
 188    public async ValueTask DisposeAsync()
 189    {
 400190        if (Interlocked.Exchange(ref _disposeGate, 1) != 0)
 196191            return;
 192
 204193        await _connectionGate.WaitAsync().ConfigureAwait(false);
 194        try
 195        {
 204196            _disposed = true;
 197
 204198            if (_channel is not null)
 199            {
 199200                using var cts = new CancellationTokenSource(_options.ShutdownTimeout);
 201                try
 202                {
 199203                    await _channel.CloseAsync(cts.Token).ConfigureAwait(false);
 197204                }
 2205                catch
 206                {
 207                    // Best effort: the channel may already be closed by broker-side shutdown.
 2208                }
 209
 199210                await _channel.DisposeAsync().ConfigureAwait(false);
 199211            }
 212
 204213            if (_connection is not null)
 214            {
 199215                using var cts = new CancellationTokenSource(_options.ShutdownTimeout);
 216                try
 217                {
 199218                    await _connection.CloseAsync(_options.ShutdownTimeout, cts.Token).ConfigureAwait(false);
 197219                }
 2220                catch
 221                {
 222                    // Best effort.
 2223                }
 224
 199225                await _connection.DisposeAsync().ConfigureAwait(false);
 199226            }
 204227        }
 228        finally
 229        {
 230            // Release, never Dispose: SemaphoreSlim.Dispose does not complete pending WaitAsync
 231            // waiters, so disposing here would strand publishers parked on the gate forever (and
 232            // the first woken waiter's finally would throw trying to Release a disposed
 233            // semaphore, never handing the permit on). Released, each parked waiter wakes in
 234            // turn and observes _disposed; the gate holds no unmanaged resources, so leaving it
 235            // undisposed leaks nothing.
 204236            _connectionGate.Release();
 237        }
 400238    }
 239}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Transports.RabbitMQ.RabbitMqAsyncResponseOptions>,AsyncResponse.Transports.RabbitMQ.IRabbitMqConnectionFactory,System.Nullable`1<System.TimeSpan>)
.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Transports.RabbitMQ.RabbitMqAsyncResponseOptions>)
ValidatePublishOptions(AsyncResponse.Transports.RabbitMQ.RabbitMqAsyncResponseOptions)
get_MaxInFlightDuration()
GetChannelAsync()
DisposeDiscardedChannelLaterAsync()
DisposeQuietlyAsync()
PublishAsync()
DisposeAsync()