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

Information
Class: AsyncResponse.Channels.NATS.NatsInboundResponse
Assembly: AsyncResponse.Channels.NATS
File(s): /_/src/Channels/AsyncResponse.Channels.NATS/NatsChannelClientAdapters.cs
Line coverage
100%
Covered lines: 1
Uncovered lines: 0
Coverable lines: 1
Total lines: 358
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
get_Payload()100%11100%

File(s)

/_/src/Channels/AsyncResponse.Channels.NATS/NatsChannelClientAdapters.cs

#LineLine coverage
 1using NATS.Client.Core;
 2using NATS.Client.KeyValueStore;
 3using NATS.Net;
 4using System.Runtime.CompilerServices;
 5using System.Threading.Channels;
 6
 7namespace AsyncResponse.Channels.NATS;
 8
 9/// <summary>Outcome of a response publish/probe over NATS request/reply.</summary>
 10internal enum NatsDeliveryOutcome
 11{
 12    /// <summary>A waiter acknowledged the message — confirmed delivery / confirmed live subscriber.</summary>
 13    Replied,
 14
 15    /// <summary>Interest existed but no ack arrived within the timeout (the subscriber received the message; only the a
 16    NoReply,
 17
 18    /// <summary>NATS reported no responders: nobody is subscribed, so the lost-subscriber fallback must run.</summary>
 19    NoResponders
 20}
 21
 22/// <summary>A response message received by a waiter, decoupled from the NATS client types for testability.</summary>
 23/// <param name="Payload">The raw JSON body, or <c>null</c> for an empty (e.g. probe) message.</param>
 24/// <param name="IsProbe">Whether the message is a liveness probe rather than a real response.</param>
 25/// <param name="ReplyAsync">Acknowledges receipt to the publisher (a no-op when the message has no reply subject).</par
 166426internal readonly record struct NatsInboundResponse(string? Payload, bool IsProbe, Func<ValueTask> ReplyAsync);
 27
 28/// <summary>A live subscription to a correlation id's response subject.</summary>
 29internal interface INatsChannelSubscription : IAsyncDisposable
 30{
 31    /// <summary>Streams inbound response messages until the subscription is disposed or the token is cancelled.</summar
 32    IAsyncEnumerable<NatsInboundResponse> ReadAsync(CancellationToken cancellationToken);
 33}
 34
 35/// <summary>
 36/// Thin abstraction over the NATS Core operations the response channel needs. Confines the NATS.Net
 37/// API surface to one place so the channel logic is unit-testable against a fake/mock.
 38/// </summary>
 39internal interface INatsResponseChannelClient
 40{
 41    /// <summary>
 42    /// Publishes <paramref name="payload"/> to <paramref name="subject"/> as a request, returning
 43    /// whether any waiter was listening. A <paramref name="probe"/> request carries no payload and is
 44    /// answered by waiters without being treated as a response.
 45    /// </summary>
 46    Task<NatsDeliveryOutcome> RequestAsync(string subject, string? payload, bool probe, TimeSpan timeout, CancellationTo
 47
 48    /// <summary>Establishes a subscription to <paramref name="subject"/>; awaiting the result guarantees the subscripti
 49    Task<INatsChannelSubscription> SubscribeAsync(string subject, CancellationToken cancellationToken);
 50
 51    /// <summary>Round-trips to the server so previously issued subscriptions are guaranteed processed before the caller
 52    Task FlushAsync(CancellationToken cancellationToken);
 53}
 54
 55/// <summary>Header marking a request as a liveness probe rather than a response payload.</summary>
 56internal static class NatsChannelHeaders
 57{
 58    public const string Probe = "AR-Probe";
 59}
 60
 61/// <summary>
 62/// The minimal raw NATS operations the response channel cannot express through mockable interface
 63/// members: <c>RequestAsync</c> and <c>PingAsync</c> are NATS.Net extension methods, and a reply is a
 64/// Core publish. Isolating them here keeps <see cref="NatsResponseChannelClient"/> fully unit-testable;
 65/// only this tiny shim wraps the un-mockable network calls.
 66/// </summary>
 67internal interface INatsRawRequester
 68{
 69    /// <summary>Sends a request and awaits a reply, throwing <c>NatsNoRespondersException</c>/<c>NatsNoReplyException</
 70    Task RequestAsync(string subject, string? payload, NatsHeaders? headers, TimeSpan timeout, CancellationToken cancell
 71
 72    /// <summary>Establishes (and awaits registration of) a Core subscription to <paramref name="subject"/>.</summary>
 73    Task<INatsSub<string>> SubscribeAsync(string subject, CancellationToken cancellationToken);
 74
 75    /// <summary>Publishes an empty acknowledgement to <paramref name="replyTo"/>.</summary>
 76    ValueTask PublishReplyAsync(string replyTo, CancellationToken cancellationToken);
 77
 78    /// <summary>Round-trips to the server (a ping) so prior subscriptions are guaranteed processed.</summary>
 79    Task FlushAsync(CancellationToken cancellationToken);
 80}
 81
 82/// <summary>Production <see cref="INatsRawRequester"/> over a NATS <see cref="INatsConnection"/>.</summary>
 83internal sealed class NatsRawRequester(INatsConnection _connection) : INatsRawRequester
 84{
 85    /// <summary>Runs the RequestAsync operation.</summary>
 86    public async Task RequestAsync(string subject, string? payload, NatsHeaders? headers, TimeSpan timeout, Cancellation
 87        => _ = await _connection.RequestAsync<string?, string>(
 88            subject,
 89            payload,
 90            headers: headers,
 91            // Pinned per call: the channel's lost-subscriber routing and liveness probe both rest on
 92            // a no-responders 503 THROWING. Left unset, that follows the app-supplied connection's
 93            // RequestReplyMode, and under Direct the sentinel arrives as an ordinary (discarded)
 94            // reply — a dead subject then looks answered and the response is dropped.
 95            replyOpts: new NatsSubOpts { Timeout = timeout, ThrowIfNoResponders = true },
 96            cancellationToken: cancellationToken).ConfigureAwait(false);
 97
 98    /// <summary>Runs the SubscribeAsync operation.</summary>
 99    public async Task<INatsSub<string>> SubscribeAsync(string subject, CancellationToken cancellationToken)
 100        => await _connection.SubscribeCoreAsync<string>(subject, cancellationToken: cancellationToken).ConfigureAwait(fa
 101
 102    /// <summary>Publishes the supplied message.</summary>
 103    public ValueTask PublishReplyAsync(string replyTo, CancellationToken cancellationToken)
 104        => _connection.PublishAsync(replyTo, string.Empty, cancellationToken: cancellationToken);
 105
 106    /// <summary>Runs the FlushAsync operation.</summary>
 107    public async Task FlushAsync(CancellationToken cancellationToken)
 108        => await _connection.PingAsync(cancellationToken).ConfigureAwait(false);
 109}
 110
 111/// <summary>
 112/// Production <see cref="INatsResponseChannelClient"/>: maps NATS request/reply outcomes and wraps Core
 113/// subscriptions, delegating the raw (un-mockable) calls to an <see cref="INatsRawRequester"/>.
 114/// </summary>
 115internal sealed class NatsResponseChannelClient(INatsRawRequester _raw) : INatsResponseChannelClient
 116{
 117    /// <summary>Runs the RequestAsync operation.</summary>
 118    public async Task<NatsDeliveryOutcome> RequestAsync(string subject, string? payload, bool probe, TimeSpan timeout, C
 119    {
 120        NatsHeaders? headers = probe ? new NatsHeaders { [NatsChannelHeaders.Probe] = "1" } : null;
 121
 122        try
 123        {
 124            await _raw.RequestAsync(subject, payload, headers, timeout, cancellationToken).ConfigureAwait(false);
 125            return NatsDeliveryOutcome.Replied;
 126        }
 127        catch (NatsNoRespondersException)
 128        {
 129            // The definitive "nobody is listening" signal — the server answered immediately because no
 130            // subscription has interest in the subject.
 131            return NatsDeliveryOutcome.NoResponders;
 132        }
 133        catch (NatsNoReplyException)
 134        {
 135            // Interest existed but no ack arrived within the timeout. For a publish the live subscriber
 136            // still received the message; for a probe it means no live subscriber answered promptly. The
 137            // channel interprets this per call site.
 138            return NatsDeliveryOutcome.NoReply;
 139        }
 140    }
 141
 142    /// <summary>Runs the SubscribeAsync operation.</summary>
 143    public async Task<INatsChannelSubscription> SubscribeAsync(string subject, CancellationToken cancellationToken)
 144    {
 145        var subscription = await _raw.SubscribeAsync(subject, cancellationToken).ConfigureAwait(false);
 146        return new NatsChannelSubscription(subscription, _raw);
 147    }
 148
 149    /// <summary>Runs the FlushAsync operation.</summary>
 150    public Task FlushAsync(CancellationToken cancellationToken) => _raw.FlushAsync(cancellationToken);
 151
 152    private sealed class NatsChannelSubscription(INatsSub<string> _subscription, INatsRawRequester _raw) : INatsChannelS
 153    {
 154        /// <summary>Runs the ReadAsync operation.</summary>
 155        public async IAsyncEnumerable<NatsInboundResponse> ReadAsync([EnumeratorCancellation] CancellationToken cancella
 156        {
 157            await foreach (var message in _subscription.Msgs.ReadAllAsync(cancellationToken).ConfigureAwait(false))
 158            {
 159                var isProbe = message.Headers is { } headers
 160                    && headers.TryGetValue(NatsChannelHeaders.Probe, out var marker)
 161                    && marker == "1";
 162
 163                var replyTo = message.ReplyTo;
 164                ValueTask Reply()
 165                    => string.IsNullOrEmpty(replyTo)
 166                        ? ValueTask.CompletedTask
 167                        : _raw.PublishReplyAsync(replyTo, CancellationToken.None);
 168
 169                yield return new NatsInboundResponse(message.Data, isProbe, Reply);
 170            }
 171        }
 172
 173        /// <summary>Releases resources held by this instance.</summary>
 174        public ValueTask DisposeAsync() => _subscription.DisposeAsync();
 175    }
 176}
 177
 178/// <summary>
 179/// Thin abstraction over the NATS JetStream Key-Value operations the recovery store needs, confining
 180/// the NATS.Net API surface to one place so the recovery store is unit-testable against a fake/mock.
 181/// The backing bucket is created lazily on first use.
 182/// </summary>
 183internal interface INatsKvStore
 184{
 185    /// <summary>Stores <paramref name="value"/> under <paramref name="key"/>, creating or replacing it.</summary>
 186    Task PutAsync(string key, string value, CancellationToken cancellationToken);
 187
 188    /// <summary>Creates <paramref name="key"/> only when absent; <c>false</c> when it already exists.</summary>
 189    Task<bool> TryCreateAsync(string key, string value, CancellationToken cancellationToken);
 190
 191    /// <summary>Replaces <paramref name="key"/> only while its revision still equals <paramref name="expectedRevision"/
 192    Task<bool> TryUpdateAsync(string key, string value, ulong expectedRevision, CancellationToken cancellationToken);
 193
 194    /// <summary>Returns the stored entry (value plus revision) for <paramref name="key"/>, or <c>null</c> when absent o
 195    Task<NatsKvEntry?> GetAsync(string key, CancellationToken cancellationToken);
 196
 197    /// <summary>Deletes <paramref name="key"/>; returns <c>true</c> when it existed, <c>false</c> when already gone.</s
 198    Task<bool> DeleteAsync(string key, CancellationToken cancellationToken);
 199
 200    /// <summary>Deletes <paramref name="key"/> only while its revision still equals <paramref name="expectedRevision"/>
 201    Task<bool> TryDeleteAsync(string key, ulong expectedRevision, CancellationToken cancellationToken);
 202
 203    /// <summary>Streams the live (non-deleted) keys in the bucket.</summary>
 204    IAsyncEnumerable<string> GetKeysAsync(CancellationToken cancellationToken);
 205}
 206
 207/// <summary>A stored value together with the KV revision it was read at, for optimistic conditional writes.</summary>
 208internal readonly record struct NatsKvEntry(string Value, ulong Revision);
 209
 210/// <summary>
 211/// Production <see cref="INatsKvStore"/> over a NATS JetStream Key-Value bucket. The bucket
 212/// (<c>{RecoveryBucket}</c>, backed by stream <c>KV_{RecoveryBucket}</c>) is created on first use
 213/// with a <c>MaxAge</c> ceiling equal to <see cref="AsyncResponseChannelOptions.RecoveryStateExpiry"/>.
 214/// </summary>
 215internal sealed class NatsKvStoreAdapter(INatsKVContext _kvContext, NatsAsyncResponseChannelOptions _options) : INatsKvS
 216{
 217    private readonly SemaphoreSlim _initGate = new(1, 1);
 218    private INatsKVStore? _store;
 219
 220    /// <summary>Runs the PutAsync operation.</summary>
 221    public async Task PutAsync(string key, string value, CancellationToken cancellationToken)
 222    {
 223        var store = await GetStoreAsync(cancellationToken).ConfigureAwait(false);
 224        await store.PutAsync(key, value, cancellationToken: cancellationToken).ConfigureAwait(false);
 225    }
 226
 227    /// <summary>Runs the TryCreateAsync operation.</summary>
 228    public async Task<bool> TryCreateAsync(string key, string value, CancellationToken cancellationToken)
 229    {
 230        var store = await GetStoreAsync(cancellationToken).ConfigureAwait(false);
 231        var result = await store.TryCreateAsync(key, value, cancellationToken: cancellationToken).ConfigureAwait(false);
 232        return result.Success;
 233    }
 234
 235    /// <summary>Runs the TryUpdateAsync operation.</summary>
 236    public async Task<bool> TryUpdateAsync(string key, string value, ulong expectedRevision, CancellationToken cancellat
 237    {
 238        var store = await GetStoreAsync(cancellationToken).ConfigureAwait(false);
 239        var result = await store.TryUpdateAsync(key, value, expectedRevision, cancellationToken: cancellationToken).Conf
 240        return result.Success;
 241    }
 242
 243    /// <summary>Runs the GetAsync operation.</summary>
 244    public async Task<NatsKvEntry?> GetAsync(string key, CancellationToken cancellationToken)
 245    {
 246        var store = await GetStoreAsync(cancellationToken).ConfigureAwait(false);
 247        try
 248        {
 249            var entry = await store.GetEntryAsync<string>(key, cancellationToken: cancellationToken).ConfigureAwait(fals
 250            return entry.Value is null ? null : new NatsKvEntry(entry.Value, entry.Revision);
 251        }
 252        catch (NatsKVKeyNotFoundException)
 253        {
 254            return null;
 255        }
 256        catch (NatsKVKeyDeletedException)
 257        {
 258            return null;
 259        }
 260    }
 261
 262    /// <summary>Runs the DeleteAsync operation.</summary>
 263    public async Task<bool> DeleteAsync(string key, CancellationToken cancellationToken)
 264    {
 265        var store = await GetStoreAsync(cancellationToken).ConfigureAwait(false);
 266
 267        bool existed;
 268        try
 269        {
 270            await store.GetEntryAsync<string>(key, cancellationToken: cancellationToken).ConfigureAwait(false);
 271            existed = true;
 272        }
 273        catch (NatsKVKeyNotFoundException)
 274        {
 275            existed = false;
 276        }
 277        catch (NatsKVKeyDeletedException)
 278        {
 279            existed = false;
 280        }
 281
 282        if (!existed)
 283            return false;
 284
 285        try
 286        {
 287            await store.DeleteAsync(key, cancellationToken: cancellationToken).ConfigureAwait(false);
 288            return true;
 289        }
 290        catch (NatsKVKeyNotFoundException)
 291        {
 292            return false;
 293        }
 294        catch (NatsKVKeyDeletedException)
 295        {
 296            return false;
 297        }
 298    }
 299
 300    /// <summary>Runs the TryDeleteAsync operation.</summary>
 301    public async Task<bool> TryDeleteAsync(string key, ulong expectedRevision, CancellationToken cancellationToken)
 302    {
 303        var store = await GetStoreAsync(cancellationToken).ConfigureAwait(false);
 304        try
 305        {
 306            await store.DeleteAsync(
 307                key,
 308                new NatsKVDeleteOpts { Revision = expectedRevision },
 309                cancellationToken: cancellationToken).ConfigureAwait(false);
 310            return true;
 311        }
 312        catch (NatsKVWrongLastRevisionException)
 313        {
 314            return false;
 315        }
 316        catch (NatsKVKeyNotFoundException)
 317        {
 318            return false;
 319        }
 320        catch (NatsKVKeyDeletedException)
 321        {
 322            return false;
 323        }
 324    }
 325
 326    /// <summary>Runs the GetKeysAsync operation.</summary>
 327    public async IAsyncEnumerable<string> GetKeysAsync([EnumeratorCancellation] CancellationToken cancellationToken)
 328    {
 329        var store = await GetStoreAsync(cancellationToken).ConfigureAwait(false);
 330        await foreach (var key in store.GetKeysAsync(cancellationToken: cancellationToken).ConfigureAwait(false))
 331            yield return key;
 332    }
 333
 334    private async ValueTask<INatsKVStore> GetStoreAsync(CancellationToken cancellationToken)
 335    {
 336        if (_store is not null)
 337            return _store;
 338
 339        await _initGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 340        try
 341        {
 342            _store ??= await _kvContext.CreateStoreAsync(
 343                new NatsKVConfig(_options.RecoveryBucket)
 344                {
 345                    MaxAge = _options.RecoveryStateExpiry,
 346                    History = 1,
 347                    NumberOfReplicas = _options.RecoveryBucketReplicas
 348                },
 349                cancellationToken).ConfigureAwait(false);
 350        }
 351        finally
 352        {
 353            _initGate.Release();
 354        }
 355
 356        return _store;
 357    }
 358}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

get_Payload()