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

Information
Class: AsyncResponse.Channels.NATS.NatsResponseChannelClient
Assembly: AsyncResponse.Channels.NATS
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Channels/AsyncResponse.Channels.NATS/NatsChannelClientAdapters.cs
Line coverage
100%
Covered lines: 25
Uncovered lines: 0
Coverable lines: 25
Total lines: 354
Line coverage: 100%
Branch coverage
100%
Covered branches: 10
Total branches: 10
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%
RequestAsync()100%22100%
SubscribeAsync()100%11100%
FlushAsync(...)100%11100%
.ctor(...)100%11100%
ReadAsync()100%66100%
Reply()100%22100%
DisposeAsync()100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/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
 26internal 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            replyOpts: new NatsSubOpts { Timeout = timeout },
 92            cancellationToken: cancellationToken).ConfigureAwait(false);
 93
 94    /// <summary>Runs the SubscribeAsync operation.</summary>
 95    public async Task<INatsSub<string>> SubscribeAsync(string subject, CancellationToken cancellationToken)
 96        => await _connection.SubscribeCoreAsync<string>(subject, cancellationToken: cancellationToken).ConfigureAwait(fa
 97
 98    /// <summary>Publishes the supplied message.</summary>
 99    public ValueTask PublishReplyAsync(string replyTo, CancellationToken cancellationToken)
 100        => _connection.PublishAsync(replyTo, string.Empty, cancellationToken: cancellationToken);
 101
 102    /// <summary>Runs the FlushAsync operation.</summary>
 103    public async Task FlushAsync(CancellationToken cancellationToken)
 104        => await _connection.PingAsync(cancellationToken).ConfigureAwait(false);
 105}
 106
 107/// <summary>
 108/// Production <see cref="INatsResponseChannelClient"/>: maps NATS request/reply outcomes and wraps Core
 109/// subscriptions, delegating the raw (un-mockable) calls to an <see cref="INatsRawRequester"/>.
 110/// </summary>
 3111internal sealed class NatsResponseChannelClient(INatsRawRequester _raw) : INatsResponseChannelClient
 112{
 113    /// <summary>Runs the RequestAsync operation.</summary>
 114    public async Task<NatsDeliveryOutcome> RequestAsync(string subject, string? payload, bool probe, TimeSpan timeout, C
 115    {
 3116        NatsHeaders? headers = probe ? new NatsHeaders { [NatsChannelHeaders.Probe] = "1" } : null;
 117
 118        try
 119        {
 3120            await _raw.RequestAsync(subject, payload, headers, timeout, cancellationToken).ConfigureAwait(false);
 3121            return NatsDeliveryOutcome.Replied;
 122        }
 3123        catch (NatsNoRespondersException)
 124        {
 125            // The definitive "nobody is listening" signal — the server answered immediately because no
 126            // subscription has interest in the subject.
 3127            return NatsDeliveryOutcome.NoResponders;
 128        }
 3129        catch (NatsNoReplyException)
 130        {
 131            // Interest existed but no ack arrived within the timeout. For a publish the live subscriber
 132            // still received the message; for a probe it means no live subscriber answered promptly. The
 133            // channel interprets this per call site.
 3134            return NatsDeliveryOutcome.NoReply;
 135        }
 3136    }
 137
 138    /// <summary>Runs the SubscribeAsync operation.</summary>
 139    public async Task<INatsChannelSubscription> SubscribeAsync(string subject, CancellationToken cancellationToken)
 140    {
 3141        var subscription = await _raw.SubscribeAsync(subject, cancellationToken).ConfigureAwait(false);
 3142        return new NatsChannelSubscription(subscription, _raw);
 3143    }
 144
 145    /// <summary>Runs the FlushAsync operation.</summary>
 3146    public Task FlushAsync(CancellationToken cancellationToken) => _raw.FlushAsync(cancellationToken);
 147
 3148    private sealed class NatsChannelSubscription(INatsSub<string> _subscription, INatsRawRequester _raw) : INatsChannelS
 149    {
 150        /// <summary>Runs the ReadAsync operation.</summary>
 151        public async IAsyncEnumerable<NatsInboundResponse> ReadAsync([EnumeratorCancellation] CancellationToken cancella
 152        {
 3153            await foreach (var message in _subscription.Msgs.ReadAllAsync(cancellationToken).ConfigureAwait(false))
 154            {
 3155                var isProbe = message.Headers is { } headers
 3156                    && headers.TryGetValue(NatsChannelHeaders.Probe, out var marker)
 3157                    && marker == "1";
 158
 3159                var replyTo = message.ReplyTo;
 160                ValueTask Reply()
 3161                    => string.IsNullOrEmpty(replyTo)
 3162                        ? ValueTask.CompletedTask
 3163                        : _raw.PublishReplyAsync(replyTo, CancellationToken.None);
 164
 3165                yield return new NatsInboundResponse(message.Data, isProbe, Reply);
 166            }
 3167        }
 168
 169        /// <summary>Releases resources held by this instance.</summary>
 3170        public ValueTask DisposeAsync() => _subscription.DisposeAsync();
 171    }
 172}
 173
 174/// <summary>
 175/// Thin abstraction over the NATS JetStream Key-Value operations the recovery store needs, confining
 176/// the NATS.Net API surface to one place so the recovery store is unit-testable against a fake/mock.
 177/// The backing bucket is created lazily on first use.
 178/// </summary>
 179internal interface INatsKvStore
 180{
 181    /// <summary>Stores <paramref name="value"/> under <paramref name="key"/>, creating or replacing it.</summary>
 182    Task PutAsync(string key, string value, CancellationToken cancellationToken);
 183
 184    /// <summary>Creates <paramref name="key"/> only when absent; <c>false</c> when it already exists.</summary>
 185    Task<bool> TryCreateAsync(string key, string value, CancellationToken cancellationToken);
 186
 187    /// <summary>Replaces <paramref name="key"/> only while its revision still equals <paramref name="expectedRevision"/
 188    Task<bool> TryUpdateAsync(string key, string value, ulong expectedRevision, CancellationToken cancellationToken);
 189
 190    /// <summary>Returns the stored entry (value plus revision) for <paramref name="key"/>, or <c>null</c> when absent o
 191    Task<NatsKvEntry?> GetAsync(string key, CancellationToken cancellationToken);
 192
 193    /// <summary>Deletes <paramref name="key"/>; returns <c>true</c> when it existed, <c>false</c> when already gone.</s
 194    Task<bool> DeleteAsync(string key, CancellationToken cancellationToken);
 195
 196    /// <summary>Deletes <paramref name="key"/> only while its revision still equals <paramref name="expectedRevision"/>
 197    Task<bool> TryDeleteAsync(string key, ulong expectedRevision, CancellationToken cancellationToken);
 198
 199    /// <summary>Streams the live (non-deleted) keys in the bucket.</summary>
 200    IAsyncEnumerable<string> GetKeysAsync(CancellationToken cancellationToken);
 201}
 202
 203/// <summary>A stored value together with the KV revision it was read at, for optimistic conditional writes.</summary>
 204internal readonly record struct NatsKvEntry(string Value, ulong Revision);
 205
 206/// <summary>
 207/// Production <see cref="INatsKvStore"/> over a NATS JetStream Key-Value bucket. The bucket
 208/// (<c>{RecoveryBucket}</c>, backed by stream <c>KV_{RecoveryBucket}</c>) is created on first use
 209/// with a <c>MaxAge</c> ceiling equal to <see cref="AsyncResponseChannelOptions.RecoveryStateExpiry"/>.
 210/// </summary>
 211internal sealed class NatsKvStoreAdapter(INatsKVContext _kvContext, NatsAsyncResponseChannelOptions _options) : INatsKvS
 212{
 213    private readonly SemaphoreSlim _initGate = new(1, 1);
 214    private INatsKVStore? _store;
 215
 216    /// <summary>Runs the PutAsync operation.</summary>
 217    public async Task PutAsync(string key, string value, CancellationToken cancellationToken)
 218    {
 219        var store = await GetStoreAsync(cancellationToken).ConfigureAwait(false);
 220        await store.PutAsync(key, value, cancellationToken: cancellationToken).ConfigureAwait(false);
 221    }
 222
 223    /// <summary>Runs the TryCreateAsync operation.</summary>
 224    public async Task<bool> TryCreateAsync(string key, string value, CancellationToken cancellationToken)
 225    {
 226        var store = await GetStoreAsync(cancellationToken).ConfigureAwait(false);
 227        var result = await store.TryCreateAsync(key, value, cancellationToken: cancellationToken).ConfigureAwait(false);
 228        return result.Success;
 229    }
 230
 231    /// <summary>Runs the TryUpdateAsync operation.</summary>
 232    public async Task<bool> TryUpdateAsync(string key, string value, ulong expectedRevision, CancellationToken cancellat
 233    {
 234        var store = await GetStoreAsync(cancellationToken).ConfigureAwait(false);
 235        var result = await store.TryUpdateAsync(key, value, expectedRevision, cancellationToken: cancellationToken).Conf
 236        return result.Success;
 237    }
 238
 239    /// <summary>Runs the GetAsync operation.</summary>
 240    public async Task<NatsKvEntry?> GetAsync(string key, CancellationToken cancellationToken)
 241    {
 242        var store = await GetStoreAsync(cancellationToken).ConfigureAwait(false);
 243        try
 244        {
 245            var entry = await store.GetEntryAsync<string>(key, cancellationToken: cancellationToken).ConfigureAwait(fals
 246            return entry.Value is null ? null : new NatsKvEntry(entry.Value, entry.Revision);
 247        }
 248        catch (NatsKVKeyNotFoundException)
 249        {
 250            return null;
 251        }
 252        catch (NatsKVKeyDeletedException)
 253        {
 254            return null;
 255        }
 256    }
 257
 258    /// <summary>Runs the DeleteAsync operation.</summary>
 259    public async Task<bool> DeleteAsync(string key, CancellationToken cancellationToken)
 260    {
 261        var store = await GetStoreAsync(cancellationToken).ConfigureAwait(false);
 262
 263        bool existed;
 264        try
 265        {
 266            await store.GetEntryAsync<string>(key, cancellationToken: cancellationToken).ConfigureAwait(false);
 267            existed = true;
 268        }
 269        catch (NatsKVKeyNotFoundException)
 270        {
 271            existed = false;
 272        }
 273        catch (NatsKVKeyDeletedException)
 274        {
 275            existed = false;
 276        }
 277
 278        if (!existed)
 279            return false;
 280
 281        try
 282        {
 283            await store.DeleteAsync(key, cancellationToken: cancellationToken).ConfigureAwait(false);
 284            return true;
 285        }
 286        catch (NatsKVKeyNotFoundException)
 287        {
 288            return false;
 289        }
 290        catch (NatsKVKeyDeletedException)
 291        {
 292            return false;
 293        }
 294    }
 295
 296    /// <summary>Runs the TryDeleteAsync operation.</summary>
 297    public async Task<bool> TryDeleteAsync(string key, ulong expectedRevision, CancellationToken cancellationToken)
 298    {
 299        var store = await GetStoreAsync(cancellationToken).ConfigureAwait(false);
 300        try
 301        {
 302            await store.DeleteAsync(
 303                key,
 304                new NatsKVDeleteOpts { Revision = expectedRevision },
 305                cancellationToken: cancellationToken).ConfigureAwait(false);
 306            return true;
 307        }
 308        catch (NatsKVWrongLastRevisionException)
 309        {
 310            return false;
 311        }
 312        catch (NatsKVKeyNotFoundException)
 313        {
 314            return false;
 315        }
 316        catch (NatsKVKeyDeletedException)
 317        {
 318            return false;
 319        }
 320    }
 321
 322    /// <summary>Runs the GetKeysAsync operation.</summary>
 323    public async IAsyncEnumerable<string> GetKeysAsync([EnumeratorCancellation] CancellationToken cancellationToken)
 324    {
 325        var store = await GetStoreAsync(cancellationToken).ConfigureAwait(false);
 326        await foreach (var key in store.GetKeysAsync(cancellationToken: cancellationToken).ConfigureAwait(false))
 327            yield return key;
 328    }
 329
 330    private async ValueTask<INatsKVStore> GetStoreAsync(CancellationToken cancellationToken)
 331    {
 332        if (_store is not null)
 333            return _store;
 334
 335        await _initGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 336        try
 337        {
 338            _store ??= await _kvContext.CreateStoreAsync(
 339                new NatsKVConfig(_options.RecoveryBucket)
 340                {
 341                    MaxAge = _options.RecoveryStateExpiry,
 342                    History = 1,
 343                    NumberOfReplicas = _options.RecoveryBucketReplicas
 344                },
 345                cancellationToken).ConfigureAwait(false);
 346        }
 347        finally
 348        {
 349            _initGate.Release();
 350        }
 351
 352        return _store;
 353    }
 354}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(AsyncResponse.Channels.NATS.INatsRawRequester)
RequestAsync()
SubscribeAsync()
FlushAsync(System.Threading.CancellationToken)
.ctor(NATS.Client.Core.INatsSub<string>, AsyncResponse.Channels.NATS.INatsRawRequester)
ReadAsync()
Reply()
DisposeAsync()