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

Information
Class: AsyncResponse.Transports.RabbitMQ.RabbitMqConnectionFactoryAdapter
Assembly: AsyncResponse.Transports.RabbitMQ
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.RabbitMQ/RabbitMqClientAdapters.cs
Line coverage
100%
Covered lines: 24
Uncovered lines: 0
Coverable lines: 24
Total lines: 171
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%
CreateConnectionAsync()100%11100%
CreateFactory(...)100%22100%

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
 311internal sealed class RabbitMqConnectionFactoryAdapter(
 312    RabbitMqAsyncResponseOptions options) : IRabbitMqConnectionFactory
 13{
 14    /// <summary>Creates the requested resource.</summary>
 15    public async Task<IRabbitMqConnection> CreateConnectionAsync(CancellationToken cancellationToken = default)
 16    {
 317        var factory = CreateFactory(options);
 118        var connection = await factory.CreateConnectionAsync(cancellationToken).ConfigureAwait(false);
 119        return new RabbitMqConnectionAdapter(connection);
 120    }
 21
 22    private static ConnectionFactory CreateFactory(RabbitMqAsyncResponseOptions options)
 23    {
 324        var factory = new ConnectionFactory
 325        {
 326            AutomaticRecoveryEnabled = options.AutomaticRecoveryEnabled,
 327            TopologyRecoveryEnabled = options.TopologyRecoveryEnabled,
 328            NetworkRecoveryInterval = options.NetworkRecoveryInterval,
 329            RequestedHeartbeat = options.RequestedHeartbeat,
 330            ClientProvidedName = options.ClientProvidedName,
 331            ConsumerDispatchConcurrency = 1
 332        };
 33
 334        if (!string.IsNullOrWhiteSpace(options.ConnectionString))
 35        {
 336            factory.Uri = new Uri(options.ConnectionString);
 337            return factory;
 38        }
 39
 240        factory.HostName = RabbitMqOptionsValidator.Required(options.HostName, nameof(options.HostName));
 241        factory.Port = options.Port;
 242        factory.VirtualHost = RabbitMqOptionsValidator.Required(options.VirtualHost, nameof(options.VirtualHost));
 243        factory.UserName = RabbitMqOptionsValidator.Required(options.UserName, nameof(options.UserName));
 244        factory.Password = options.Password;
 345        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
 92internal 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
 96        => 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,
 100        => 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
 104        => inner.QueueBindAsync(queue, exchange, routingKey, cancellationToken: cancellationToken);
 105
 106    /// <summary>Runs the BasicQosAsync operation.</summary>
 107    public Task BasicQosAsync(ushort prefetchCount, CancellationToken cancellationToken = default)
 108        => 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
 112        => 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    {
 117        var consumer = new AsyncEventingBasicConsumer(inner);
 118        consumer.ReceivedAsync += (_, args) =>
 119        {
 120            var delivery = new RabbitMqDelivery(
 121                args.ConsumerTag,
 122                args.DeliveryTag,
 123                args.Redelivered,
 124                args.Exchange,
 125                args.RoutingKey,
 126                args.BasicProperties,
 127                args.Body,
 128                args.CancellationToken);
 129            return handler(delivery);
 130        };
 131
 132        return inner.BasicConsumeAsync(
 133            queue,
 134            autoAck: false,
 135            consumerTag: string.Empty,
 136            noLocal: false,
 137            exclusive: false,
 138            arguments: null,
 139            consumer,
 140            cancellationToken);
 141    }
 142
 143    /// <summary>Runs the BasicCancelAsync operation.</summary>
 144    public Task BasicCancelAsync(string consumerTag, CancellationToken cancellationToken = default)
 145        => inner.BasicCancelAsync(consumerTag, cancellationToken: cancellationToken);
 146
 147    /// <summary>Runs the BasicAckAsync operation.</summary>
 148    public ValueTask BasicAckAsync(ulong deliveryTag, CancellationToken cancellationToken = default)
 149        => 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)
 153        => inner.BasicNackAsync(deliveryTag, multiple: false, requeue, cancellationToken);
 154
 155    /// <summary>Runs the CloseAsync operation.</summary>
 156    public Task CloseAsync(CancellationToken cancellationToken = default)
 157        => inner.CloseAsync(200, "AsyncResponse shutdown", abort: false, cancellationToken);
 158
 159    /// <summary>Releases resources held by this instance.</summary>
 160    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(AsyncResponse.Transports.RabbitMQ.RabbitMqAsyncResponseOptions)
CreateConnectionAsync()
CreateFactory(AsyncResponse.Transports.RabbitMQ.RabbitMqAsyncResponseOptions)