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

Information
Class: AsyncResponse.Transports.RabbitMQ.RabbitMqWorkerSubscriber
Assembly: AsyncResponse.Transports.RabbitMQ
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.RabbitMQ/RabbitMqSubscriberServices.cs
Line coverage
100%
Covered lines: 11
Uncovered lines: 0
Coverable lines: 11
Total lines: 210
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%
.ctor(...)100%11100%
get_QueueName()100%11100%
get_SubscriberOptions()100%11100%
get_SubscriberRole()100%11100%
EnsureTopologyAsync(...)100%11100%
HandleMessageAsync(...)100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.RabbitMQ/RabbitMqSubscriberServices.cs

#LineLine coverage
 1using Microsoft.Extensions.Hosting;
 2using Microsoft.Extensions.Logging;
 3using Microsoft.Extensions.Options;
 4using System.Text;
 5
 6namespace AsyncResponse.Transports.RabbitMQ;
 7
 8internal abstract class RabbitMqSubscriberService : BackgroundService
 9{
 10    private readonly IRabbitMqConnectionFactory _connectionFactory;
 11
 12    /// <summary>Runs the RabbitMqSubscriberService operation.</summary>
 13    protected RabbitMqSubscriberService(
 14        IOptions<RabbitMqAsyncResponseOptions> options,
 15        ILogger logger)
 16        : this(options, logger, new RabbitMqConnectionFactoryAdapter(options.Value))
 17    {
 18    }
 19
 20    /// <summary>Runs the RabbitMqSubscriberService operation.</summary>
 21    protected RabbitMqSubscriberService(
 22        IOptions<RabbitMqAsyncResponseOptions> options,
 23        ILogger logger,
 24        IRabbitMqConnectionFactory connectionFactory)
 25    {
 26        Options = options.Value;
 27        Logger = logger;
 28        _connectionFactory = connectionFactory;
 29    }
 30
 31    protected RabbitMqAsyncResponseOptions Options { get; }
 32    protected ILogger Logger { get; }
 33
 34    protected abstract string QueueName { get; }
 35    protected abstract RabbitMqSubscriberOptions SubscriberOptions { get; }
 36    protected abstract RabbitMqSubscriberRole SubscriberRole { get; }
 37    /// <summary>Ensures the required resource exists.</summary>
 38    protected abstract Task EnsureTopologyAsync(IRabbitMqChannel channel, CancellationToken cancellationToken);
 39    /// <summary>Handles the delivered message.</summary>
 40    protected abstract Task HandleMessageAsync(RabbitMqDelivery delivery, CancellationToken cancellationToken);
 41
 42    /// <summary>Runs this background operation until cancellation is requested.</summary>
 43    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 44    {
 45        var queue = QueueName;
 46        RabbitMqMessageDispatcher.ValidateOptions(Options, SubscriberOptions, SubscriberRole);
 47
 48        // basic.nack requeue does not increment the x-death header, so the resolved attempt for a plain
 49        // requeued delivery never exceeds 2. Warn once at startup instead of silently never enforcing the cap.
 50        if (SubscriberOptions.AckMode is RabbitMqAckMode.AckAfterHandlerCompletes
 51            && SubscriberOptions.MaxDeliveryAttempts > 2)
 52        {
 53            Logger.LogWarning(
 54                "RabbitMQ {OptionName} is {MaxDeliveryAttempts} for queue {Queue} ({Role}), but attempts beyond 2 cannot
 55                + "basic.nack requeue does not increment x-death, so the cap only takes effect once a TTL-retry dead-let
 56                + "re-delivers the message through a dead-letter exchange.",
 57                nameof(RabbitMqSubscriberOptions.MaxDeliveryAttempts),
 58                SubscriberOptions.MaxDeliveryAttempts,
 59                queue,
 60                SubscriberRole);
 61        }
 62
 63        while (!stoppingToken.IsCancellationRequested)
 64        {
 65            try
 66            {
 67                await RunSubscriberAsync(queue, stoppingToken).ConfigureAwait(false);
 68                return;
 69            }
 70            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 71            {
 72                return;
 73            }
 74            catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
 75            {
 76                var retryDelay = Options.NetworkRecoveryInterval > TimeSpan.Zero
 77                    ? Options.NetworkRecoveryInterval
 78                    : TimeSpan.FromSeconds(5);
 79                Logger.LogWarning(
 80                    ex,
 81                    "RabbitMQ subscriber could not start for queue {Queue} ({Role}); retrying in {RetryDelay}.",
 82                    queue,
 83                    SubscriberRole,
 84                    retryDelay);
 85                await Task.Delay(retryDelay, stoppingToken).ConfigureAwait(false);
 86            }
 87        }
 88    }
 89
 90    private async Task RunSubscriberAsync(string queue, CancellationToken stoppingToken)
 91    {
 92        await using var connection = await _connectionFactory.CreateConnectionAsync(stoppingToken).ConfigureAwait(false)
 93        await using var channel = await connection.CreateChannelAsync(cancellationToken: stoppingToken).ConfigureAwait(f
 94        await EnsureTopologyAsync(channel, stoppingToken).ConfigureAwait(false);
 95        await channel.BasicQosAsync(SubscriberOptions.PrefetchCount, stoppingToken).ConfigureAwait(false);
 96
 97        await using var dispatcher = RabbitMqMessageDispatcher.Create(
 98            HandleMessageAsync,
 99            Options,
 100            SubscriberOptions,
 101            Logger,
 102            queue,
 103            SubscriberRole);
 104
 105        Logger.LogInformation(
 106            "RabbitMQ subscriber started. Queue: {Queue}. Role: {Role}. AckMode: {AckMode}.",
 107            queue,
 108            SubscriberRole,
 109            SubscriberOptions.AckMode);
 110
 111        var consumerTag = await channel.BasicConsumeAsync(
 112            queue,
 113            delivery => dispatcher.HandleAsync(delivery, channel, stoppingToken),
 114            stoppingToken).ConfigureAwait(false);
 115
 116        try
 117        {
 118            await Task.Delay(Timeout.InfiniteTimeSpan, stoppingToken).ConfigureAwait(false);
 119        }
 120        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 121        {
 122            using var shutdown = new CancellationTokenSource(Options.ShutdownTimeout);
 123            await channel.BasicCancelAsync(consumerTag, shutdown.Token).ConfigureAwait(false);
 124            await channel.CloseAsync(shutdown.Token).ConfigureAwait(false);
 125            await connection.CloseAsync(Options.ShutdownTimeout, shutdown.Token).ConfigureAwait(false);
 126        }
 127    }
 128}
 129
 130internal sealed class RabbitMqWorkerSubscriber : RabbitMqSubscriberService
 131{
 132    private readonly IAsyncResponseIngress _ingress;
 133
 134    /// <summary>Runs the RabbitMqWorkerSubscriber operation.</summary>
 135    public RabbitMqWorkerSubscriber(
 136        IOptions<RabbitMqAsyncResponseOptions> options,
 137        IAsyncResponseIngress ingress,
 138        ILogger<RabbitMqWorkerSubscriber> logger)
 3139        : base(options, logger)
 140    {
 3141        _ingress = ingress;
 3142    }
 143
 144    internal RabbitMqWorkerSubscriber(
 145        IOptions<RabbitMqAsyncResponseOptions> options,
 146        IAsyncResponseIngress ingress,
 147        ILogger<RabbitMqWorkerSubscriber> logger,
 148        IRabbitMqConnectionFactory connectionFactory)
 2149        : base(options, logger, connectionFactory)
 150    {
 3151        _ingress = ingress;
 3152    }
 153
 154    protected override string QueueName
 3155        => RabbitMqOptionsValidator.Required(Options.WorkerQueue, nameof(Options.WorkerQueue));
 156
 3157    protected override RabbitMqSubscriberOptions SubscriberOptions => Options.WorkerSubscriber;
 3158    protected override RabbitMqSubscriberRole SubscriberRole => RabbitMqSubscriberRole.Worker;
 159
 160    /// <summary>Ensures the required resource exists.</summary>
 161    protected override Task EnsureTopologyAsync(IRabbitMqChannel channel, CancellationToken cancellationToken)
 3162        => RabbitMqTopology.EnsureWorkerAsync(channel, Options, cancellationToken);
 163
 164    /// <summary>Handles the delivered message.</summary>
 165    protected override Task HandleMessageAsync(RabbitMqDelivery delivery, CancellationToken cancellationToken)
 3166        => _ingress.HandleWorkerMessageAsync(Encoding.UTF8.GetString(delivery.Body.Span));
 167}
 168
 169internal sealed class RabbitMqResponseIngressSubscriber : RabbitMqSubscriberService
 170{
 171    private readonly IAsyncResponseIngress _ingress;
 172
 173    /// <summary>Runs the RabbitMqResponseIngressSubscriber operation.</summary>
 174    public RabbitMqResponseIngressSubscriber(
 175        IOptions<RabbitMqAsyncResponseOptions> options,
 176        IAsyncResponseIngress ingress,
 177        ILogger<RabbitMqResponseIngressSubscriber> logger)
 178        : base(options, logger)
 179    {
 180        _ingress = ingress;
 181    }
 182
 183    internal RabbitMqResponseIngressSubscriber(
 184        IOptions<RabbitMqAsyncResponseOptions> options,
 185        IAsyncResponseIngress ingress,
 186        ILogger<RabbitMqResponseIngressSubscriber> logger,
 187        IRabbitMqConnectionFactory connectionFactory)
 188        : base(options, logger, connectionFactory)
 189    {
 190        _ingress = ingress;
 191    }
 192
 193    protected override string QueueName
 194        => RabbitMqOptionsValidator.Required(Options.ResponseQueue, nameof(Options.ResponseQueue));
 195
 196    protected override RabbitMqSubscriberOptions SubscriberOptions => Options.ResponseSubscriber;
 197    protected override RabbitMqSubscriberRole SubscriberRole => RabbitMqSubscriberRole.ResponseIngress;
 198
 199    /// <summary>Ensures the required resource exists.</summary>
 200    protected override Task EnsureTopologyAsync(IRabbitMqChannel channel, CancellationToken cancellationToken)
 201        => RabbitMqTopology.EnsureResponseAsync(channel, Options, cancellationToken);
 202
 203    /// <summary>Handles the delivered message.</summary>
 204    protected override Task HandleMessageAsync(RabbitMqDelivery delivery, CancellationToken cancellationToken)
 205    {
 206        var messageJson = Encoding.UTF8.GetString(delivery.Body.Span);
 207        var correlationId = RabbitMqCorrelationIdExtractor.Extract(delivery, messageJson, Options);
 208        return _ingress.HandleResponseMessageAsync(messageJson, correlationId);
 209    }
 210}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.Transports.RabbitMQ.RabbitMqAsyncResponseOptions>, AsyncResponse.IAsyncResponseIngress, Microsoft.Extensions.Logging.ILogger<AsyncResponse.Transports.RabbitMQ.RabbitMqWorkerSubscriber>)
.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.Transports.RabbitMQ.RabbitMqAsyncResponseOptions>, AsyncResponse.IAsyncResponseIngress, Microsoft.Extensions.Logging.ILogger<AsyncResponse.Transports.RabbitMQ.RabbitMqWorkerSubscriber>, AsyncResponse.Transports.RabbitMQ.IRabbitMqConnectionFactory)
get_QueueName()
get_SubscriberOptions()
get_SubscriberRole()
EnsureTopologyAsync(AsyncResponse.Transports.RabbitMQ.IRabbitMqChannel, System.Threading.CancellationToken)
HandleMessageAsync(AsyncResponse.Transports.RabbitMQ.RabbitMqDelivery, System.Threading.CancellationToken)