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

Information
Class: AsyncResponse.Transports.RabbitMQ.RabbitMqChannelAdapter
Assembly: AsyncResponse.Transports.RabbitMQ
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.RabbitMQ/RabbitMqClientAdapters.cs
Line coverage
100%
Covered lines: 34
Uncovered lines: 0
Coverable lines: 34
Total lines: 171
Line coverage: 100%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
ExchangeDeclareAsync(...)100%11100%
QueueDeclareAsync()100%11100%
QueueBindAsync(...)100%11100%
BasicQosAsync(...)100%11100%
BasicPublishAsync(...)100%11100%
BasicConsumeAsync(...)100%11100%
BasicCancelAsync(...)100%11100%
BasicAckAsync(...)100%11100%
BasicNackAsync(...)100%11100%
CloseAsync(...)100%11100%
DisposeAsync()100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/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    Task<IRabbitMqChannel> CreateChannelAsync(bool publisherConfirmations = false, CancellationToken cancellationToken =
 52    Task CloseAsync(TimeSpan timeout, CancellationToken cancellationToken = default);
 53}
 54
 55internal sealed class RabbitMqConnectionAdapter(IConnection inner) : IRabbitMqConnection
 56{
 57    /// <summary>Creates the requested resource.</summary>
 58    public async Task<IRabbitMqChannel> CreateChannelAsync(bool publisherConfirmations = false, CancellationToken cancel
 59    {
 60        // Enabling publisher confirmations with tracking makes BasicPublishAsync await the broker
 61        // acknowledgement and throw on a nack or an unroutable (mandatory) return, so a worker job is
 62        // never silently lost. Consumer channels pass false and keep the lighter default behavior.
 63        var options = publisherConfirmations
 64            ? new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true)
 65            : null;
 66        var channel = await inner.CreateChannelAsync(options, cancellationToken).ConfigureAwait(false);
 67        return new RabbitMqChannelAdapter(channel);
 68    }
 69
 70    /// <summary>Runs the CloseAsync operation.</summary>
 71    public Task CloseAsync(TimeSpan timeout, CancellationToken cancellationToken = default)
 72        => inner.CloseAsync(200, "AsyncResponse shutdown", timeout, abort: false, cancellationToken);
 73
 74    /// <summary>Releases resources held by this instance.</summary>
 75    public ValueTask DisposeAsync() => inner.DisposeAsync();
 76}
 77
 78internal interface IRabbitMqChannel : IAsyncDisposable
 79{
 80    Task ExchangeDeclareAsync(string exchange, string type, bool durable, bool autoDelete, CancellationToken cancellatio
 81    Task QueueDeclareAsync(string queue, bool durable, bool exclusive, bool autoDelete, IDictionary<string, object?>? ar
 82    Task QueueBindAsync(string queue, string exchange, string routingKey, CancellationToken cancellationToken = default)
 83    Task BasicQosAsync(ushort prefetchCount, CancellationToken cancellationToken = default);
 84    ValueTask BasicPublishAsync(string exchange, string routingKey, BasicProperties properties, ReadOnlyMemory<byte> bod
 85    Task<string> BasicConsumeAsync(string queue, Func<RabbitMqDelivery, Task> handler, CancellationToken cancellationTok
 86    Task BasicCancelAsync(string consumerTag, CancellationToken cancellationToken = default);
 87    ValueTask BasicAckAsync(ulong deliveryTag, CancellationToken cancellationToken = default);
 88    ValueTask BasicNackAsync(ulong deliveryTag, bool requeue, CancellationToken cancellationToken = default);
 89    Task CloseAsync(CancellationToken cancellationToken = default);
 90}
 91
 392internal sealed class RabbitMqChannelAdapter(IChannel inner) : IRabbitMqChannel
 93{
 94    /// <summary>Runs the ExchangeDeclareAsync operation.</summary>
 95    public Task ExchangeDeclareAsync(string exchange, string type, bool durable, bool autoDelete, CancellationToken canc
 396        => inner.ExchangeDeclareAsync(exchange, type, durable, autoDelete, cancellationToken: cancellationToken);
 97
 98    /// <summary>Runs the QueueDeclareAsync operation.</summary>
 99    public async Task QueueDeclareAsync(string queue, bool durable, bool exclusive, bool autoDelete, IDictionary<string,
 3100        => await inner.QueueDeclareAsync(queue, durable, exclusive, autoDelete, arguments, cancellationToken: cancellati
 101
 102    /// <summary>Runs the QueueBindAsync operation.</summary>
 103    public Task QueueBindAsync(string queue, string exchange, string routingKey, CancellationToken cancellationToken = d
 3104        => inner.QueueBindAsync(queue, exchange, routingKey, cancellationToken: cancellationToken);
 105
 106    /// <summary>Runs the BasicQosAsync operation.</summary>
 107    public Task BasicQosAsync(ushort prefetchCount, CancellationToken cancellationToken = default)
 3108        => inner.BasicQosAsync(0, prefetchCount, global: false, cancellationToken);
 109
 110    /// <summary>Runs the BasicPublishAsync operation.</summary>
 111    public ValueTask BasicPublishAsync(string exchange, string routingKey, BasicProperties properties, ReadOnlyMemory<by
 3112        => inner.BasicPublishAsync(exchange, routingKey, mandatory: true, properties, body, cancellationToken);
 113
 114    /// <summary>Runs the BasicConsumeAsync operation.</summary>
 115    public Task<string> BasicConsumeAsync(string queue, Func<RabbitMqDelivery, Task> handler, CancellationToken cancella
 116    {
 3117        var consumer = new AsyncEventingBasicConsumer(inner);
 3118        consumer.ReceivedAsync += (_, args) =>
 3119        {
 3120            var delivery = new RabbitMqDelivery(
 3121                args.ConsumerTag,
 3122                args.DeliveryTag,
 3123                args.Redelivered,
 3124                args.Exchange,
 3125                args.RoutingKey,
 3126                args.BasicProperties,
 3127                args.Body,
 3128                args.CancellationToken);
 3129            return handler(delivery);
 3130        };
 131
 3132        return inner.BasicConsumeAsync(
 3133            queue,
 3134            autoAck: false,
 3135            consumerTag: string.Empty,
 3136            noLocal: false,
 3137            exclusive: false,
 3138            arguments: null,
 3139            consumer,
 3140            cancellationToken);
 141    }
 142
 143    /// <summary>Runs the BasicCancelAsync operation.</summary>
 144    public Task BasicCancelAsync(string consumerTag, CancellationToken cancellationToken = default)
 3145        => inner.BasicCancelAsync(consumerTag, cancellationToken: cancellationToken);
 146
 147    /// <summary>Runs the BasicAckAsync operation.</summary>
 148    public ValueTask BasicAckAsync(ulong deliveryTag, CancellationToken cancellationToken = default)
 3149        => inner.BasicAckAsync(deliveryTag, multiple: false, cancellationToken);
 150
 151    /// <summary>Runs the BasicNackAsync operation.</summary>
 152    public ValueTask BasicNackAsync(ulong deliveryTag, bool requeue, CancellationToken cancellationToken = default)
 3153        => inner.BasicNackAsync(deliveryTag, multiple: false, requeue, cancellationToken);
 154
 155    /// <summary>Runs the CloseAsync operation.</summary>
 156    public Task CloseAsync(CancellationToken cancellationToken = default)
 3157        => inner.CloseAsync(200, "AsyncResponse shutdown", abort: false, cancellationToken);
 158
 159    /// <summary>Releases resources held by this instance.</summary>
 3160    public ValueTask DisposeAsync() => inner.DisposeAsync();
 161}
 162
 163internal sealed record RabbitMqDelivery(
 164    string ConsumerTag,
 165    ulong DeliveryTag,
 166    bool Redelivered,
 167    string Exchange,
 168    string RoutingKey,
 169    IReadOnlyBasicProperties BasicProperties,
 170    ReadOnlyMemory<byte> Body,
 171    CancellationToken CancellationToken);
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(RabbitMQ.Client.IChannel)
ExchangeDeclareAsync(string, string, bool, bool, System.Threading.CancellationToken)
QueueDeclareAsync()
QueueBindAsync(string, string, string, System.Threading.CancellationToken)
BasicQosAsync(ushort, System.Threading.CancellationToken)
BasicPublishAsync(string, string, RabbitMQ.Client.BasicProperties, System.ReadOnlyMemory<byte>, System.Threading.CancellationToken)
BasicConsumeAsync(string, System.Func<AsyncResponse.Transports.RabbitMQ.RabbitMqDelivery, System.Threading.Tasks.Task>, System.Threading.CancellationToken)
BasicCancelAsync(string, System.Threading.CancellationToken)
BasicAckAsync(ulong, System.Threading.CancellationToken)
BasicNackAsync(ulong, bool, System.Threading.CancellationToken)
CloseAsync(System.Threading.CancellationToken)
DisposeAsync()