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

Information
Class: AsyncResponse.Transports.RabbitMQ.RabbitMqConnectionAdapter
Assembly: AsyncResponse.Transports.RabbitMQ
File(s): /_/src/Transports/AsyncResponse.Transports.RabbitMQ/RabbitMqClientAdapters.cs
Line coverage
100%
Covered lines: 10
Uncovered lines: 0
Coverable lines: 10
Total lines: 204
Line coverage: 100%
Branch coverage
100%
Covered branches: 2
Total branches: 2
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
get_IsOpen()100%11100%
CreateChannelAsync()100%22100%
CloseAsync(...)100%11100%
DisposeAsync()100%11100%

File(s)

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

#LineLine coverage
 1using RabbitMQ.Client;
 2using RabbitMQ.Client.Events;
 3
 4namespace AsyncResponse.Transports.RabbitMQ;
 5
 6internal interface IRabbitMqConnectionFactory
 7{
 8    Task<IRabbitMqConnection> CreateConnectionAsync(CancellationToken cancellationToken = default);
 9}
 10
 11internal sealed class RabbitMqConnectionFactoryAdapter(
 12    RabbitMqAsyncResponseOptions options) : IRabbitMqConnectionFactory
 13{
 14    /// <summary>Creates the requested resource.</summary>
 15    public async Task<IRabbitMqConnection> CreateConnectionAsync(CancellationToken cancellationToken = default)
 16    {
 17        var factory = CreateFactory(options);
 18        var connection = await factory.CreateConnectionAsync(cancellationToken).ConfigureAwait(false);
 19        return new RabbitMqConnectionAdapter(connection);
 20    }
 21
 22    private static ConnectionFactory CreateFactory(RabbitMqAsyncResponseOptions options)
 23    {
 24        var factory = new ConnectionFactory
 25        {
 26            AutomaticRecoveryEnabled = options.AutomaticRecoveryEnabled,
 27            TopologyRecoveryEnabled = options.TopologyRecoveryEnabled,
 28            NetworkRecoveryInterval = options.NetworkRecoveryInterval,
 29            RequestedHeartbeat = options.RequestedHeartbeat,
 30            ClientProvidedName = options.ClientProvidedName,
 31            ConsumerDispatchConcurrency = 1
 32        };
 33
 34        if (!string.IsNullOrWhiteSpace(options.ConnectionString))
 35        {
 36            factory.Uri = new Uri(options.ConnectionString);
 37            return factory;
 38        }
 39
 40        factory.HostName = RabbitMqOptionsValidator.Required(options.HostName, nameof(options.HostName));
 41        factory.Port = options.Port;
 42        factory.VirtualHost = RabbitMqOptionsValidator.Required(options.VirtualHost, nameof(options.VirtualHost));
 43        factory.UserName = RabbitMqOptionsValidator.Required(options.UserName, nameof(options.UserName));
 44        factory.Password = options.Password;
 45        return factory;
 46    }
 47}
 48
 49internal interface IRabbitMqConnection : IAsyncDisposable
 50{
 51    /// <summary>False once the connection is closed for good; with automatic recovery enabled the client object stays o
 52    bool IsOpen { get; }
 53    Task<IRabbitMqChannel> CreateChannelAsync(bool publisherConfirmations = false, CancellationToken cancellationToken =
 54    Task CloseAsync(TimeSpan timeout, CancellationToken cancellationToken = default);
 55}
 56
 58957internal sealed class RabbitMqConnectionAdapter(IConnection inner) : IRabbitMqConnection
 58{
 459    public bool IsOpen => inner.IsOpen;
 60
 61    /// <summary>Creates the requested resource.</summary>
 62    public async Task<IRabbitMqChannel> CreateChannelAsync(bool publisherConfirmations = false, CancellationToken cancel
 63    {
 64        // Enabling publisher confirmations with tracking makes BasicPublishAsync await the broker
 65        // acknowledgement and throw on a nack or an unroutable (mandatory) return, so a worker job is
 66        // never silently lost. Consumer channels pass false and keep the lighter default behavior.
 58167        var options = publisherConfirmations
 58168            ? new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true)
 58169            : null;
 58170        var channel = await inner.CreateChannelAsync(options, cancellationToken).ConfigureAwait(false);
 58171        return new RabbitMqChannelAdapter(channel);
 58172    }
 73
 74    /// <summary>Runs the CloseAsync operation.</summary>
 75    public Task CloseAsync(TimeSpan timeout, CancellationToken cancellationToken = default)
 57776        => inner.CloseAsync(200, "AsyncResponse shutdown", timeout, abort: false, cancellationToken);
 77
 78    /// <summary>Releases resources held by this instance.</summary>
 57979    public ValueTask DisposeAsync() => inner.DisposeAsync();
 80}
 81
 82internal interface IRabbitMqChannel : IAsyncDisposable
 83{
 84    /// <summary>False once the channel is closed (a 404/406 protocol error closes it without any callback failing).</su
 85    bool IsOpen { get; }
 86    Task ExchangeDeclareAsync(string exchange, string type, bool durable, bool autoDelete, CancellationToken cancellatio
 87    Task QueueDeclareAsync(string queue, bool durable, bool exclusive, bool autoDelete, IDictionary<string, object?>? ar
 88    Task QueueBindAsync(string queue, string exchange, string routingKey, CancellationToken cancellationToken = default)
 89    Task BasicQosAsync(ushort prefetchCount, CancellationToken cancellationToken = default);
 90    ValueTask BasicPublishAsync(string exchange, string routingKey, BasicProperties properties, ReadOnlyMemory<byte> bod
 91    Task<RabbitMqConsumer> BasicConsumeAsync(string queue, Func<RabbitMqDelivery, Task> handler, CancellationToken cance
 92    Task BasicCancelAsync(string consumerTag, CancellationToken cancellationToken = default);
 93    ValueTask BasicAckAsync(ulong deliveryTag, CancellationToken cancellationToken = default);
 94    ValueTask BasicNackAsync(ulong deliveryTag, bool requeue, CancellationToken cancellationToken = default);
 95    Task CloseAsync(CancellationToken cancellationToken = default);
 96}
 97
 98internal sealed class RabbitMqChannelAdapter(IChannel inner) : IRabbitMqChannel
 99{
 100    public bool IsOpen => inner.IsOpen;
 101
 102    /// <summary>Runs the ExchangeDeclareAsync operation.</summary>
 103    public Task ExchangeDeclareAsync(string exchange, string type, bool durable, bool autoDelete, CancellationToken canc
 104        => inner.ExchangeDeclareAsync(exchange, type, durable, autoDelete, cancellationToken: cancellationToken);
 105
 106    /// <summary>Runs the QueueDeclareAsync operation.</summary>
 107    public async Task QueueDeclareAsync(string queue, bool durable, bool exclusive, bool autoDelete, IDictionary<string,
 108        => await inner.QueueDeclareAsync(queue, durable, exclusive, autoDelete, arguments, cancellationToken: cancellati
 109
 110    /// <summary>Runs the QueueBindAsync operation.</summary>
 111    public Task QueueBindAsync(string queue, string exchange, string routingKey, CancellationToken cancellationToken = d
 112        => inner.QueueBindAsync(queue, exchange, routingKey, cancellationToken: cancellationToken);
 113
 114    /// <summary>Runs the BasicQosAsync operation.</summary>
 115    public Task BasicQosAsync(ushort prefetchCount, CancellationToken cancellationToken = default)
 116        => inner.BasicQosAsync(0, prefetchCount, global: false, cancellationToken);
 117
 118    /// <summary>Runs the BasicPublishAsync operation.</summary>
 119    public ValueTask BasicPublishAsync(string exchange, string routingKey, BasicProperties properties, ReadOnlyMemory<by
 120        => inner.BasicPublishAsync(exchange, routingKey, mandatory: true, properties, body, cancellationToken);
 121
 122    /// <summary>Runs the BasicConsumeAsync operation.</summary>
 123    public async Task<RabbitMqConsumer> BasicConsumeAsync(string queue, Func<RabbitMqDelivery, Task> handler, Cancellati
 124    {
 125        var consumer = new AsyncEventingBasicConsumer(inner);
 126        consumer.ReceivedAsync += (_, args) =>
 127        {
 128            var delivery = new RabbitMqDelivery(
 129                args.ConsumerTag,
 130                args.DeliveryTag,
 131                args.Redelivered,
 132                args.Exchange,
 133                args.RoutingKey,
 134                args.BasicProperties,
 135                args.Body,
 136                args.CancellationToken);
 137            return handler(delivery);
 138        };
 139
 140        // Deliveries can stop without any exception reaching the subscriber: a broker-side
 141        // basic.cancel (queue deleted) only raises UnregisteredAsync and a channel-level protocol
 142        // close only raises ChannelShutdownAsync. Fold both into one termination task the subscriber
 143        // can await. A client-initiated BasicCancelAsync completes it too (cancel-ok also raises
 144        // UnregisteredAsync); the subscriber filters that out with its stopping token.
 145        var terminated = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
 146        consumer.UnregisteredAsync += (_, _) =>
 147        {
 148            terminated.TrySetResult("the broker canceled the consumer (basic.cancel, typically a deleted queue)");
 149            return Task.CompletedTask;
 150        };
 151        inner.ChannelShutdownAsync += (_, args) =>
 152        {
 153            terminated.TrySetResult($"the channel shut down ({args.ReplyCode} {args.ReplyText})");
 154            return Task.CompletedTask;
 155        };
 156
 157        var consumerTag = await inner.BasicConsumeAsync(
 158            queue,
 159            autoAck: false,
 160            consumerTag: string.Empty,
 161            noLocal: false,
 162            exclusive: false,
 163            arguments: null,
 164            consumer,
 165            cancellationToken).ConfigureAwait(false);
 166        return new RabbitMqConsumer(consumerTag, terminated.Task);
 167    }
 168
 169    /// <summary>Runs the BasicCancelAsync operation.</summary>
 170    public Task BasicCancelAsync(string consumerTag, CancellationToken cancellationToken = default)
 171        => inner.BasicCancelAsync(consumerTag, cancellationToken: cancellationToken);
 172
 173    /// <summary>Runs the BasicAckAsync operation.</summary>
 174    public ValueTask BasicAckAsync(ulong deliveryTag, CancellationToken cancellationToken = default)
 175        => inner.BasicAckAsync(deliveryTag, multiple: false, cancellationToken);
 176
 177    /// <summary>Runs the BasicNackAsync operation.</summary>
 178    public ValueTask BasicNackAsync(ulong deliveryTag, bool requeue, CancellationToken cancellationToken = default)
 179        => inner.BasicNackAsync(deliveryTag, multiple: false, requeue, cancellationToken);
 180
 181    /// <summary>Runs the CloseAsync operation.</summary>
 182    public Task CloseAsync(CancellationToken cancellationToken = default)
 183        => inner.CloseAsync(200, "AsyncResponse shutdown", abort: false, cancellationToken);
 184
 185    /// <summary>Releases resources held by this instance.</summary>
 186    public ValueTask DisposeAsync() => inner.DisposeAsync();
 187}
 188
 189/// <summary>
 190/// An active consumer subscription. <see cref="Terminated"/> completes (with a reason) when the
 191/// broker cancels the consumer or the channel shuts down — cases that stop deliveries forever
 192/// without failing any pending call.
 193/// </summary>
 194internal sealed record RabbitMqConsumer(string ConsumerTag, Task<string> Terminated);
 195
 196internal sealed record RabbitMqDelivery(
 197    string ConsumerTag,
 198    ulong DeliveryTag,
 199    bool Redelivered,
 200    string Exchange,
 201    string RoutingKey,
 202    IReadOnlyBasicProperties BasicProperties,
 203    ReadOnlyMemory<byte> Body,
 204    CancellationToken CancellationToken);
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(RabbitMQ.Client.IConnection)
get_IsOpen()
CreateChannelAsync()
CloseAsync(System.TimeSpan,System.Threading.CancellationToken)
DisposeAsync()