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

Information
Class: AsyncResponse.Transports.NATS.NatsJobDelivery
Assembly: AsyncResponse.Transports.NATS
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.NATS/NatsTransportClientAdapters.cs
Line coverage
100%
Covered lines: 8
Uncovered lines: 0
Coverable lines: 8
Total lines: 153
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%

File(s)

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

#LineLine coverage
 1using NATS.Client.Core;
 2using NATS.Client.JetStream;
 3using NATS.Client.JetStream.Models;
 4using System.Runtime.CompilerServices;
 5
 6namespace AsyncResponse.Transports.NATS;
 7
 8/// <summary>A worker/response message pulled from a JetStream consumer, decoupled from NATS client types for testabilit
 9/// <param name="Subject">The subject the message was published to.</param>
 10/// <param name="Payload">The raw JSON body.</param>
 11/// <param name="Headers">The message headers (correlation id, etc.).</param>
 12/// <param name="NumDelivered">How many times JetStream has delivered this message (1 on first delivery).</param>
 13/// <param name="AckAsync">Acknowledges the message so it is not redelivered.</param>
 14/// <param name="NakAsync">Negatively acknowledges the message, requesting redelivery after the given delay.</param>
 15/// <param name="TermAsync">Terminates the message so JetStream stops redelivering it (used after dead-lettering).</para
 316internal sealed record NatsJobDelivery(
 317    string Subject,
 318    string Payload,
 319    IReadOnlyDictionary<string, string> Headers,
 320    long NumDelivered,
 321    Func<ValueTask> AckAsync,
 322    Func<TimeSpan, ValueTask> NakAsync,
 323    Func<ValueTask> TermAsync);
 24
 25/// <summary>
 26/// Thin abstraction over the NATS JetStream operations the transport needs, confining the NATS.Net
 27/// API surface to one place so the worker transport, dispatcher, and subscribers are unit-testable
 28/// against a fake/mock.
 29/// </summary>
 30internal interface INatsJetStreamTransport
 31{
 32    /// <summary>Idempotently creates or updates the stream capturing <paramref name="subject"/>.</summary>
 33    Task EnsureStreamAsync(string stream, string subject, long? maxMessages, CancellationToken cancellationToken);
 34
 35    /// <summary>Idempotently creates or updates a durable explicit-ack consumer on <paramref name="stream"/>.</summary>
 36    Task EnsureConsumerAsync(string stream, string durable, TimeSpan ackWait, CancellationToken cancellationToken);
 37
 38    /// <summary>Publishes <paramref name="payload"/> to <paramref name="subject"/> via JetStream and returns the assign
 39    Task<string> PublishAsync(string subject, string payload, IReadOnlyDictionary<string, string>? headers, Cancellation
 40
 41    /// <summary>Continuously consumes messages from the durable consumer on <paramref name="stream"/>.</summary>
 42    IAsyncEnumerable<NatsJobDelivery> ConsumeAsync(string stream, string durable, int batchSize, CancellationToken cance
 43}
 44
 45/// <summary>Production <see cref="INatsJetStreamTransport"/> over a NATS <see cref="INatsJSContext"/>.</summary>
 46internal sealed class NatsJetStreamTransportAdapter(INatsJSContext _jetStream) : INatsJetStreamTransport
 47{
 48    /// <summary>Ensures the required resource exists.</summary>
 49    public async Task EnsureStreamAsync(string stream, string subject, long? maxMessages, CancellationToken cancellation
 50    {
 51        var config = new StreamConfig(stream, [subject])
 52        {
 53            MaxMsgs = maxMessages ?? -1,
 54            Retention = StreamConfigRetention.Limits
 55        };
 56        await _jetStream.CreateOrUpdateStreamAsync(config, cancellationToken).ConfigureAwait(false);
 57    }
 58
 59    /// <summary>Ensures the required resource exists.</summary>
 60    public async Task EnsureConsumerAsync(string stream, string durable, TimeSpan ackWait, CancellationToken cancellatio
 61    {
 62        var config = new ConsumerConfig(durable)
 63        {
 64            DurableName = durable,
 65            AckPolicy = ConsumerConfigAckPolicy.Explicit,
 66            AckWait = ackWait,
 67            // Redelivery attempts are bounded by the dispatcher (via NumDelivered + Terminate), so the
 68            // consumer itself is left unlimited rather than silently swallowing the last attempt.
 69            MaxDeliver = -1
 70        };
 71        await _jetStream.CreateOrUpdateConsumerAsync(stream, config, cancellationToken).ConfigureAwait(false);
 72    }
 73
 74    /// <summary>Publishes the supplied message.</summary>
 75    public async Task<string> PublishAsync(string subject, string payload, IReadOnlyDictionary<string, string>? headers,
 76    {
 77        var ack = await _jetStream.PublishAsync(
 78            subject,
 79            payload,
 80            headers: ToHeaders(headers),
 81            cancellationToken: cancellationToken).ConfigureAwait(false);
 82        ack.EnsureSuccess();
 83        return ack.Seq.ToString();
 84    }
 85
 86    /// <summary>Runs the ConsumeAsync operation.</summary>
 87    public async IAsyncEnumerable<NatsJobDelivery> ConsumeAsync(
 88        string stream,
 89        string durable,
 90        int batchSize,
 91        [EnumeratorCancellation] CancellationToken cancellationToken)
 92    {
 93        var consumer = await _jetStream.GetConsumerAsync(stream, durable, cancellationToken).ConfigureAwait(false);
 94        var consumeOpts = new NatsJSConsumeOpts { MaxMsgs = batchSize };
 95
 96        await foreach (var message in consumer.ConsumeAsync<string>(opts: consumeOpts, cancellationToken: cancellationTo
 97        {
 98            var numDelivered = (long)(message.Metadata?.NumDelivered ?? 1);
 99            var captured = message;
 100
 101            yield return new NatsJobDelivery(
 102                captured.Subject,
 103                captured.Data ?? string.Empty,
 104                FromHeaders(captured.Headers),
 105                numDelivered,
 106                () => captured.AckAsync(cancellationToken: CancellationToken.None),
 107                delay => captured.NakAsync(delay: delay, cancellationToken: CancellationToken.None),
 108                () => captured.AckTerminateAsync(cancellationToken: CancellationToken.None));
 109        }
 110    }
 111
 112    private static NatsHeaders? ToHeaders(IReadOnlyDictionary<string, string>? headers)
 113    {
 114        if (headers is null || headers.Count == 0)
 115            return null;
 116
 117        var natsHeaders = new NatsHeaders();
 118        foreach (var (key, value) in headers)
 119            natsHeaders[key] = value;
 120        return natsHeaders;
 121    }
 122
 123    private static IReadOnlyDictionary<string, string> FromHeaders(NatsHeaders? headers)
 124    {
 125        if (headers is null || headers.Count == 0)
 126            return EmptyHeaders;
 127
 128        var result = new Dictionary<string, string>(headers.Count, StringComparer.OrdinalIgnoreCase);
 129        foreach (var key in headers.Keys)
 130            result[key] = headers[key].ToString();
 131        return result;
 132    }
 133
 134    private static readonly IReadOnlyDictionary<string, string> EmptyHeaders =
 135        new Dictionary<string, string>(0, StringComparer.OrdinalIgnoreCase);
 136}
 137
 138/// <summary>Bounded exponential-backoff retry for transient NATS failures, mirroring the other transports.</summary>
 139internal static class NatsTransportRetry
 140{
 141    /// <summary>Runs this background operation until cancellation is requested.</summary>
 142    public static Task<T> ExecuteAsync<T>(
 143        Func<CancellationToken, Task<T>> action,
 144        int maxAttempts,
 145        TimeSpan baseDelay,
 146        TimeSpan maxDelay,
 147        CancellationToken cancellationToken)
 148        => AsyncResponseRetry.ExecuteAsync(action, IsTransient, maxAttempts, baseDelay, maxDelay, cancellationToken);
 149
 150    /// <summary>Runs the IsTransient operation.</summary>
 151    public static bool IsTransient(Exception exception)
 152        => exception is NatsException or TimeoutException && exception is not OperationCanceledException;
 153}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(string, string, System.Collections.Generic.IReadOnlyDictionary<string, string>, long, System.Func<System.Threading.Tasks.ValueTask>, System.Func<System.TimeSpan, System.Threading.Tasks.ValueTask>, System.Func<System.Threading.Tasks.ValueTask>)