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

Information
Class: AsyncResponse.Transports.NATS.NatsSubscriberService
Assembly: AsyncResponse.Transports.NATS
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.NATS/NatsSubscriberServices.cs
Line coverage
100%
Covered lines: 45
Uncovered lines: 0
Coverable lines: 45
Total lines: 179
Line coverage: 100%
Branch coverage
83%
Covered branches: 5
Total branches: 6
Branch coverage: 83.3%
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%
ExecuteAsync()50%2292.31%
RunSubscriberAsync()100%44100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.NATS/NatsSubscriberServices.cs

#LineLine coverage
 1using Microsoft.Extensions.Hosting;
 2using Microsoft.Extensions.Logging;
 3using Microsoft.Extensions.Options;
 4using NATS.Client.Core;
 5using NATS.Net;
 6
 7namespace AsyncResponse.Transports.NATS;
 8
 9/// <summary>
 10/// Base hosted service that consumes a JetStream subject through a durable consumer and routes each
 11/// message to the AsyncResponse ingress with the configured acknowledgement/redelivery/dead-letter
 12/// policy. A failed consume loop is retried with bounded backoff so a transient NATS outage does not
 13/// kill the subscriber.
 14/// </summary>
 15internal abstract class NatsSubscriberService : BackgroundService
 16{
 17    private readonly INatsJetStreamTransport _jetStream;
 18
 19    /// <summary>Runs the NatsSubscriberService operation.</summary>
 20    protected NatsSubscriberService(
 21        IOptions<NatsAsyncResponseTransportOptions> options,
 22        INatsConnection connection,
 23        ILogger logger)
 324        : this(options, new NatsJetStreamTransportAdapter(connection.CreateJetStreamContext()), logger)
 25    {
 326    }
 27
 28    /// <summary>Runs the NatsSubscriberService operation.</summary>
 329    protected NatsSubscriberService(
 330        IOptions<NatsAsyncResponseTransportOptions> options,
 331        INatsJetStreamTransport jetStream,
 332        ILogger logger)
 33    {
 334        Options = options.Value;
 335        NatsTransportOptionsValidator.ValidateCommon(Options);
 336        _jetStream = jetStream;
 337        Logger = logger;
 338        Schema = new NatsTransportSubjectSchema(Options);
 339    }
 40
 41    protected NatsAsyncResponseTransportOptions Options { get; }
 42    protected ILogger Logger { get; }
 43    protected NatsTransportSubjectSchema Schema { get; }
 44
 45    protected abstract string Subject { get; }
 46    protected abstract string Stream { get; }
 47    protected abstract string Consumer { get; }
 48    protected abstract NatsSubscriberOptions SubscriberOptions { get; }
 49    protected abstract NatsSubscriberRole Role { get; }
 50    /// <summary>Handles the delivered message.</summary>
 51    protected abstract Task HandleMessageAsync(NatsJobDelivery delivery, CancellationToken cancellationToken);
 52
 53    /// <summary>Runs this background operation until cancellation is requested.</summary>
 54    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 55    {
 356        NatsTransportOptionsValidator.ValidateSubscriber(Options, SubscriberOptions, Role.ToString());
 57
 358        var failures = 0;
 359        while (!stoppingToken.IsCancellationRequested)
 60        {
 61            try
 62            {
 363                await RunSubscriberAsync(stoppingToken).ConfigureAwait(false);
 164                return;
 65            }
 266            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 67            {
 268                return;
 69            }
 270            catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
 71            {
 272                failures++;
 273                var retryDelay = AsyncResponseRetry.Backoff(failures, Options.SubscriberRetryBaseDelay, Options.Subscrib
 274                Logger.LogWarning(ex, "NATS subscriber failed for subject {Subject} ({Role}); retrying in {RetryDelay}."
 275                await Task.Delay(retryDelay, stoppingToken).ConfigureAwait(false);
 76            }
 77        }
 378    }
 79
 80    private async Task RunSubscriberAsync(CancellationToken stoppingToken)
 81    {
 382        if (Options.CreateStreams)
 83        {
 384            await _jetStream.EnsureStreamAsync(Stream, Subject, Options.StreamMaxMessages, stoppingToken).ConfigureAwait
 385            if (Options.DeadLetterEnabled)
 386                await _jetStream.EnsureStreamAsync(Schema.DeadLetterStream, Schema.DeadLetterSubject, Options.DeadLetter
 87        }
 88
 389        await _jetStream.EnsureConsumerAsync(Stream, Consumer, Options.AckWait, stoppingToken).ConfigureAwait(false);
 90
 391        await using var dispatcher = new NatsMessageDispatcher(
 392            HandleMessageAsync,
 393            _jetStream,
 394            Options,
 395            SubscriberOptions,
 396            Schema,
 397            Logger,
 398            Role,
 399            Consumer);
 100
 3101        Logger.LogInformation(
 3102            "NATS subscriber started. Subject: {Subject}. Stream: {Stream}. Consumer: {Consumer}. Role: {Role}. AckMode:
 3103            Subject, Stream, Consumer, Role, SubscriberOptions.AckMode);
 104
 3105        await foreach (var delivery in _jetStream.ConsumeAsync(Stream, Consumer, SubscriberOptions.BatchSize, stoppingTo
 106        {
 3107            await dispatcher.HandleAsync(delivery, stoppingToken).ConfigureAwait(false);
 108        }
 1109    }
 110}
 111
 112/// <summary>Consumes worker-job messages and executes them through the AsyncResponse ingress.</summary>
 113internal sealed class NatsWorkerSubscriber : NatsSubscriberService
 114{
 115    private readonly IAsyncResponseIngress _ingress;
 116
 117    /// <summary>Runs the NatsWorkerSubscriber operation.</summary>
 118    public NatsWorkerSubscriber(
 119        IOptions<NatsAsyncResponseTransportOptions> options,
 120        INatsConnection connection,
 121        IAsyncResponseIngress ingress,
 122        ILogger<NatsWorkerSubscriber> logger)
 123        : base(options, connection, logger)
 124        => _ingress = ingress;
 125
 126    internal NatsWorkerSubscriber(
 127        IOptions<NatsAsyncResponseTransportOptions> options,
 128        INatsJetStreamTransport jetStream,
 129        IAsyncResponseIngress ingress,
 130        ILogger<NatsWorkerSubscriber> logger)
 131        : base(options, jetStream, logger)
 132        => _ingress = ingress;
 133
 134    protected override string Subject => Schema.WorkerSubject;
 135    protected override string Stream => Schema.WorkerStream;
 136    protected override string Consumer => Options.WorkerConsumer;
 137    protected override NatsSubscriberOptions SubscriberOptions => Options.WorkerSubscriber;
 138    protected override NatsSubscriberRole Role => NatsSubscriberRole.Worker;
 139
 140    /// <summary>Handles the delivered message.</summary>
 141    protected override Task HandleMessageAsync(NatsJobDelivery delivery, CancellationToken cancellationToken)
 142        => _ingress.HandleWorkerMessageAsync(delivery.Payload);
 143}
 144
 145/// <summary>Consumes response messages and feeds them into the AsyncResponse ingress, correlated by header or JSON body
 146internal sealed class NatsResponseIngressSubscriber : NatsSubscriberService
 147{
 148    private readonly IAsyncResponseIngress _ingress;
 149
 150    /// <summary>Runs the NatsResponseIngressSubscriber operation.</summary>
 151    public NatsResponseIngressSubscriber(
 152        IOptions<NatsAsyncResponseTransportOptions> options,
 153        INatsConnection connection,
 154        IAsyncResponseIngress ingress,
 155        ILogger<NatsResponseIngressSubscriber> logger)
 156        : base(options, connection, logger)
 157        => _ingress = ingress;
 158
 159    internal NatsResponseIngressSubscriber(
 160        IOptions<NatsAsyncResponseTransportOptions> options,
 161        INatsJetStreamTransport jetStream,
 162        IAsyncResponseIngress ingress,
 163        ILogger<NatsResponseIngressSubscriber> logger)
 164        : base(options, jetStream, logger)
 165        => _ingress = ingress;
 166
 167    protected override string Subject => Schema.ResponseSubject;
 168    protected override string Stream => Schema.ResponseStream;
 169    protected override string Consumer => Options.ResponseConsumer;
 170    protected override NatsSubscriberOptions SubscriberOptions => Options.ResponseSubscriber;
 171    protected override NatsSubscriberRole Role => NatsSubscriberRole.ResponseIngress;
 172
 173    /// <summary>Handles the delivered message.</summary>
 174    protected override Task HandleMessageAsync(NatsJobDelivery delivery, CancellationToken cancellationToken)
 175    {
 176        var correlationId = NatsCorrelationIdExtractor.Extract(delivery.Headers, delivery.Payload, Options);
 177        return _ingress.HandleResponseMessageAsync(delivery.Payload, correlationId);
 178    }
 179}
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.NATS.NatsAsyncResponseTransportOptions>, NATS.Client.Core.INatsConnection, Microsoft.Extensions.Logging.ILogger)
.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.Transports.NATS.NatsAsyncResponseTransportOptions>, AsyncResponse.Transports.NATS.INatsJetStreamTransport, Microsoft.Extensions.Logging.ILogger)
ExecuteAsync()
RunSubscriberAsync()