diff --git a/sandbox/Example.Client/Program.cs b/sandbox/Example.Client/Program.cs index 0e78c0ba6..4bac3d56a 100644 --- a/sandbox/Example.Client/Program.cs +++ b/sandbox/Example.Client/Program.cs @@ -1,6 +1,7 @@ // See https://aka.ms/new-console-template for more information using System.Text; +using NATS.Client.JetStream; using NATS.Net; CancellationTokenSource cts = new(); @@ -88,7 +89,12 @@ } // Use JetStream by referencing NATS.Client.JetStream package -// var js = client.GetJetStream(); +var js = client.CreateJetStreamContext(); +await foreach (var stream in js.ListStreamsAsync()) +{ + Console.WriteLine($"JetStream Stream: {stream.Info.Config.Name}"); +} + await cts.CancelAsync(); await Task.WhenAll(tasks); diff --git a/src/NATS.Client.Core/INatsConnection.cs b/src/NATS.Client.Core/INatsConnection.cs index c54c9454a..2ecf74caa 100644 --- a/src/NATS.Client.Core/INatsConnection.cs +++ b/src/NATS.Client.Core/INatsConnection.cs @@ -1,23 +1,56 @@ using System.Diagnostics.CodeAnalysis; +using System.Threading.Channels; namespace NATS.Client.Core; public interface INatsConnection : INatsClient { + /// + /// Event that is raised when the connection to the NATS server is disconnected. + /// event AsyncEventHandler? ConnectionDisconnected; + /// + /// Event that is raised when the connection to the NATS server is opened. + /// event AsyncEventHandler? ConnectionOpened; + /// + /// Event that is raised when a reconnect attempt is failed. + /// event AsyncEventHandler? ReconnectFailed; + /// + /// Event that is raised when a message is dropped for a subscription. + /// event AsyncEventHandler? MessageDropped; + /// + /// Server information received from the NATS server. + /// INatsServerInfo? ServerInfo { get; } + /// + /// Options used to configure the NATS connection. + /// NatsOpts Opts { get; } + /// + /// Connection state of the NATS connection. + /// NatsConnectionState ConnectionState { get; } + /// + /// Subscription manager used to manage subscriptions for the NATS connection. + /// + INatsSubscriptionManager SubscriptionManager { get; } + + /// + /// Singleton instance of the NATS header parser used to parse message headers + /// used by the NATS connection. + /// + NatsHeaderParser HeaderParser { get; } + /// /// Publishes a serializable message payload to the given subject name, optionally supplying a reply subject. /// @@ -87,4 +120,61 @@ IAsyncEnumerable> RequestManyAsync( NatsPubOpts? requestOpts = default, NatsSubOpts? replyOpts = default, CancellationToken cancellationToken = default); + + /// + /// Adds a subscription to the NATS connection for a given object. + /// Subscriptions are managed by the connection and are automatically removed when the connection is closed. + /// + /// The object representing the subscription details. + /// A used to cancel the operation. + /// A that represents the asynchronous subscription operation. + ValueTask AddSubAsync(NatsSubBase sub, CancellationToken cancellationToken = default); + + /// + /// Creates a subscription with appropriate request and reply subjects publishing the request. + /// It's the caller's responsibility to retrieve the reply messages and complete the subscription. + /// + /// The type of the request data. + /// The type of the expected reply. + /// The subject to subscribe to. + /// The optional request data. + /// The optional headers to include with the request. + /// The optional serializer for the request data. + /// The optional deserializer for the reply data. + /// The optional publishing options for the request. + /// The optional subscription options for the reply. + /// The optional cancellation token. + /// A representing the asynchronous operation of creating the request subscription. + ValueTask> CreateRequestSubAsync( + string subject, + TRequest? data, + NatsHeaders? headers = default, + INatsSerialize? requestSerializer = default, + INatsDeserialize? replySerializer = default, + NatsPubOpts? requestOpts = default, + NatsSubOpts? replyOpts = default, + CancellationToken cancellationToken = default); + + /// + /// Retrieves the bounded channel options for creating a channel used by a subscription. + /// Options are built from the connection's configuration and the subscription channel options. + /// Used to aid in custom message handling when building a subscription channel. + /// + /// The options for configuring the subscription channel. + /// The bounded channel options used for creating the subscription channel. + BoundedChannelOptions GetBoundedChannelOpts(NatsSubChannelOpts? subChannelOpts); + + /// + /// Called when a message is dropped for a subscription. + /// Used to aid in custom message handling when a subscription's message channel is full. + /// + /// The representing the subscription. + /// The number of pending messages at the time the drop occurred. + /// The dropped message represented by . + /// Specifies the type of data in the dropped message. + /// + /// This method is expected to complete quickly to avoid further delays in processing; + /// if complex work is required, it is recommended to offload to a channel or other out-of-band processor. + /// + void OnMessageDropped(NatsSubBase natsSub, int pending, NatsMsg msg); } diff --git a/src/NATS.Client.Core/INatsSubscriptionManager.cs b/src/NATS.Client.Core/INatsSubscriptionManager.cs new file mode 100644 index 000000000..bd5dde65c --- /dev/null +++ b/src/NATS.Client.Core/INatsSubscriptionManager.cs @@ -0,0 +1,19 @@ +namespace NATS.Client.Core; + +/// +/// Subscription manager interface. +/// +/// +/// This interface is used to manage subscriptions. However, it is not intended to be used directly. +/// You can implement this interface if you are using low-level APIs and implement your own +/// subscription manager. +/// +public interface INatsSubscriptionManager +{ + /// + /// Remove a subscription. + /// + /// Subscription to remove. + /// A value task that represents the asynchronous remove operation. + public ValueTask RemoveAsync(NatsSubBase sub); +} diff --git a/src/NATS.Client.Core/Internal/InboxSub.cs b/src/NATS.Client.Core/Internal/InboxSub.cs index fc1eea5d3..39923cc78 100644 --- a/src/NATS.Client.Core/Internal/InboxSub.cs +++ b/src/NATS.Client.Core/Internal/InboxSub.cs @@ -15,7 +15,7 @@ public InboxSub( string subject, NatsSubOpts? opts, NatsConnection connection, - ISubscriptionManager manager) + INatsSubscriptionManager manager) : base(connection, manager, subject, queueGroup: default, opts) { _inbox = inbox; @@ -35,7 +35,7 @@ protected override void TryComplete() } } -internal class InboxSubBuilder : ISubscriptionManager +internal class InboxSubBuilder : INatsSubscriptionManager { private readonly ILogger _logger; #if NETSTANDARD2_0 @@ -46,7 +46,7 @@ internal class InboxSubBuilder : ISubscriptionManager public InboxSubBuilder(ILogger logger) => _logger = logger; - public InboxSub Build(string subject, NatsSubOpts? opts, NatsConnection connection, ISubscriptionManager manager) + public InboxSub Build(string subject, NatsSubOpts? opts, NatsConnection connection, INatsSubscriptionManager manager) { return new InboxSub(this, subject, opts, connection, manager); } diff --git a/src/NATS.Client.Core/Internal/SubscriptionManager.cs b/src/NATS.Client.Core/Internal/SubscriptionManager.cs index e422d7d03..d07758b96 100644 --- a/src/NATS.Client.Core/Internal/SubscriptionManager.cs +++ b/src/NATS.Client.Core/Internal/SubscriptionManager.cs @@ -6,16 +6,11 @@ namespace NATS.Client.Core.Internal; -internal interface ISubscriptionManager -{ - public ValueTask RemoveAsync(NatsSubBase sub); -} - internal record struct SidMetadata(string Subject, WeakReference WeakReference); internal sealed record SubscriptionMetadata(int Sid); -internal sealed class SubscriptionManager : ISubscriptionManager, IAsyncDisposable +internal sealed class SubscriptionManager : INatsSubscriptionManager, IAsyncDisposable { private readonly ILogger _logger; private readonly bool _trace; @@ -192,7 +187,7 @@ public ValueTask RemoveAsync(NatsSubBase sub) /// Commands returned form all the subscriptions will be run as a priority right after reconnection is established. /// /// Enumerable list of commands - public async ValueTask WriteReconnectCommandsAsync(CommandWriter commandWriter) + internal async ValueTask WriteReconnectCommandsAsync(CommandWriter commandWriter) { if (_debug) { @@ -226,7 +221,7 @@ public async ValueTask WriteReconnectCommandsAsync(CommandWriter commandWriter) } } - public ISubscriptionManager GetManagerFor(string subject) + internal INatsSubscriptionManager GetManagerFor(string subject) { if (IsInboxSubject(subject)) return InboxSubBuilder; diff --git a/src/NATS.Client.Core/NatsConnection.LowLevelApi.cs b/src/NATS.Client.Core/NatsConnection.LowLevelApi.cs index 9cbcf23e7..47ee2f383 100644 --- a/src/NATS.Client.Core/NatsConnection.LowLevelApi.cs +++ b/src/NATS.Client.Core/NatsConnection.LowLevelApi.cs @@ -2,14 +2,15 @@ namespace NATS.Client.Core; public partial class NatsConnection { - internal ValueTask SubAsync(NatsSubBase sub, CancellationToken cancellationToken = default) => + /// + public ValueTask AddSubAsync(NatsSubBase sub, CancellationToken cancellationToken = default) => ConnectionState != NatsConnectionState.Open ? ConnectAndSubAsync(sub, cancellationToken) - : SubscriptionManager.SubscribeAsync(sub, cancellationToken); + : _subscriptionManager.SubscribeAsync(sub, cancellationToken); private async ValueTask ConnectAndSubAsync(NatsSubBase sub, CancellationToken cancellationToken = default) { await ConnectAsync().AsTask().WaitAsync(cancellationToken).ConfigureAwait(false); - await SubscriptionManager.SubscribeAsync(sub, cancellationToken).ConfigureAwait(false); + await _subscriptionManager.SubscribeAsync(sub, cancellationToken).ConfigureAwait(false); } } diff --git a/src/NATS.Client.Core/NatsConnection.RequestReply.cs b/src/NATS.Client.Core/NatsConnection.RequestReply.cs index 5c838116f..1da7b7346 100644 --- a/src/NATS.Client.Core/NatsConnection.RequestReply.cs +++ b/src/NATS.Client.Core/NatsConnection.RequestReply.cs @@ -38,7 +38,7 @@ public async ValueTask> RequestAsync( try { replyOpts = SetReplyOptsDefaults(replyOpts); - await using var sub1 = await RequestSubAsync(subject, data, headers, requestSerializer, replySerializer, requestOpts, replyOpts, cancellationToken) + await using var sub1 = await CreateRequestSubAsync(subject, data, headers, requestSerializer, replySerializer, requestOpts, replyOpts, cancellationToken) .ConfigureAwait(false); await foreach (var msg in sub1.Msgs.ReadAllAsync(cancellationToken).ConfigureAwait(false)) @@ -56,7 +56,7 @@ public async ValueTask> RequestAsync( } replyOpts = SetReplyOptsDefaults(replyOpts); - await using var sub = await RequestSubAsync(subject, data, headers, requestSerializer, replySerializer, requestOpts, replyOpts, cancellationToken) + await using var sub = await CreateRequestSubAsync(subject, data, headers, requestSerializer, replySerializer, requestOpts, replyOpts, cancellationToken) .ConfigureAwait(false); await foreach (var msg in sub.Msgs.ReadAllAsync(cancellationToken).ConfigureAwait(false)) @@ -95,7 +95,7 @@ public async IAsyncEnumerable> RequestManyAsync(subject, data, headers, requestSerializer, replySerializer, requestOpts, replyOpts, cancellationToken) + await using var sub = await CreateRequestSubAsync(subject, data, headers, requestSerializer, replySerializer, requestOpts, replyOpts, cancellationToken) .ConfigureAwait(false); await foreach (var msg in sub.Msgs.ReadAllAsync(cancellationToken).ConfigureAwait(false)) diff --git a/src/NATS.Client.Core/NatsConnection.RequestSub.cs b/src/NATS.Client.Core/NatsConnection.RequestSub.cs index a3f94c24f..8763d8967 100644 --- a/src/NATS.Client.Core/NatsConnection.RequestSub.cs +++ b/src/NATS.Client.Core/NatsConnection.RequestSub.cs @@ -2,7 +2,8 @@ namespace NATS.Client.Core; public partial class NatsConnection { - internal async ValueTask> RequestSubAsync( + /// + public async ValueTask> CreateRequestSubAsync( string subject, TRequest? data, NatsHeaders? headers = default, @@ -15,8 +16,8 @@ internal async ValueTask> RequestSubAsync( var replyTo = NewInbox(); replySerializer ??= Opts.SerializerRegistry.GetDeserializer(); - var sub = new NatsSub(this, SubscriptionManager.InboxSubBuilder, replyTo, queueGroup: default, replyOpts, replySerializer); - await SubAsync(sub, cancellationToken).ConfigureAwait(false); + var sub = new NatsSub(this, _subscriptionManager.InboxSubBuilder, replyTo, queueGroup: default, replyOpts, replySerializer); + await AddSubAsync(sub, cancellationToken).ConfigureAwait(false); requestSerializer ??= Opts.SerializerRegistry.GetSerializer(); await PublishAsync(subject, data, headers, replyTo, requestSerializer, requestOpts, cancellationToken).ConfigureAwait(false); diff --git a/src/NATS.Client.Core/NatsConnection.Subscribe.cs b/src/NATS.Client.Core/NatsConnection.Subscribe.cs index d163db30e..770bd6421 100644 --- a/src/NATS.Client.Core/NatsConnection.Subscribe.cs +++ b/src/NATS.Client.Core/NatsConnection.Subscribe.cs @@ -10,8 +10,8 @@ public async IAsyncEnumerable> SubscribeAsync(string subject, stri { serializer ??= Opts.SerializerRegistry.GetDeserializer(); - await using var sub = new NatsSub(this, SubscriptionManager.GetManagerFor(subject), subject, queueGroup, opts, serializer, cancellationToken); - await SubAsync(sub, cancellationToken: cancellationToken).ConfigureAwait(false); + await using var sub = new NatsSub(this, _subscriptionManager.GetManagerFor(subject), subject, queueGroup, opts, serializer, cancellationToken); + await AddSubAsync(sub, cancellationToken: cancellationToken).ConfigureAwait(false); // We don't cancel the channel reader here because we want to keep reading until the subscription // channel writer completes so that messages left in the channel can be consumed before exit the loop. @@ -25,8 +25,8 @@ public async IAsyncEnumerable> SubscribeAsync(string subject, stri public async ValueTask> SubscribeCoreAsync(string subject, string? queueGroup = default, INatsDeserialize? serializer = default, NatsSubOpts? opts = default, CancellationToken cancellationToken = default) { serializer ??= Opts.SerializerRegistry.GetDeserializer(); - var sub = new NatsSub(this, SubscriptionManager.GetManagerFor(subject), subject, queueGroup, opts, serializer, cancellationToken); - await SubAsync(sub, cancellationToken).ConfigureAwait(false); + var sub = new NatsSub(this, _subscriptionManager.GetManagerFor(subject), subject, queueGroup, opts, serializer, cancellationToken); + await AddSubAsync(sub, cancellationToken).ConfigureAwait(false); return sub; } } diff --git a/src/NATS.Client.Core/NatsConnection.cs b/src/NATS.Client.Core/NatsConnection.cs index f08315b8f..b777a4bec 100644 --- a/src/NATS.Client.Core/NatsConnection.cs +++ b/src/NATS.Client.Core/NatsConnection.cs @@ -48,6 +48,7 @@ public partial class NatsConnection : INatsConnection private readonly BoundedChannelOptions _defaultSubscriptionChannelOpts; private readonly Channel<(NatsEvent, NatsEventArgs)> _eventChannel; private readonly ClientOpts _clientOpts; + private readonly SubscriptionManager _subscriptionManager; private int _pongCount; private int _connectionState; @@ -84,7 +85,7 @@ public NatsConnection(NatsOpts opts) Counter = new ConnectionStatsCounter(); CommandWriter = new CommandWriter("main", this, _pool, Opts, Counter, EnqueuePing); InboxPrefix = NewInbox(opts.InboxPrefix); - SubscriptionManager = new SubscriptionManager(this, InboxPrefix); + _subscriptionManager = new SubscriptionManager(this, InboxPrefix); _clientOpts = ClientOpts.Create(Opts); HeaderParser = new NatsHeaderParser(opts.HeaderEncoding); _defaultSubscriptionChannelOpts = new BoundedChannelOptions(opts.SubPendingChannelCapacity) @@ -130,16 +131,16 @@ private set public INatsServerInfo? ServerInfo => WritableServerInfo; // server info is set when received INFO + public INatsSubscriptionManager SubscriptionManager => _subscriptionManager; + + public NatsHeaderParser HeaderParser { get; } + internal bool IsDisposed { get => Interlocked.CompareExchange(ref _isDisposed, 0, 0) == 1; private set => Interlocked.Exchange(ref _isDisposed, value ? 1 : 0); } - internal NatsHeaderParser HeaderParser { get; } - - internal SubscriptionManager SubscriptionManager { get; } - internal CommandWriter CommandWriter { get; } internal string InboxPrefix { get; } @@ -182,6 +183,36 @@ public async ValueTask ConnectAsync() await InitialConnectAsync().ConfigureAwait(false); } + /// + public void OnMessageDropped(NatsSubBase natsSub, int pending, NatsMsg msg) + { + var subject = msg.Subject; + _logger.LogWarning("Dropped message from {Subject} with {Pending} pending messages", subject, pending); + _eventChannel.Writer.TryWrite((NatsEvent.MessageDropped, new NatsMessageDroppedEventArgs(natsSub, pending, subject, msg.ReplyTo, msg.Headers, msg.Data))); + } + + /// + public BoundedChannelOptions GetBoundedChannelOpts(NatsSubChannelOpts? subChannelOpts) + { + if (subChannelOpts is { } overrideOpts) + { + return new BoundedChannelOptions(overrideOpts.Capacity ?? + _defaultSubscriptionChannelOpts.Capacity) + { + AllowSynchronousContinuations = + _defaultSubscriptionChannelOpts.AllowSynchronousContinuations, + FullMode = + overrideOpts.FullMode ?? _defaultSubscriptionChannelOpts.FullMode, + SingleWriter = _defaultSubscriptionChannelOpts.SingleWriter, + SingleReader = _defaultSubscriptionChannelOpts.SingleReader, + }; + } + else + { + return _defaultSubscriptionChannelOpts; + } + } + public virtual async ValueTask DisposeAsync() { if (!IsDisposed) @@ -199,7 +230,7 @@ public virtual async ValueTask DisposeAsync() #endif } - await SubscriptionManager.DisposeAsync().ConfigureAwait(false); + await _subscriptionManager.DisposeAsync().ConfigureAwait(false); await CommandWriter.DisposeAsync().ConfigureAwait(false); _waitForOpenConnection.TrySetCanceled(); #if NET8_0_OR_GREATER @@ -224,7 +255,7 @@ internal string SpanDestinationName(string subject) internal ValueTask PublishToClientHandlersAsync(string subject, string? replyTo, int sid, in ReadOnlySequence? headersBuffer, in ReadOnlySequence payloadBuffer) { - return SubscriptionManager.PublishToClientHandlersAsync(subject, replyTo, sid, headersBuffer, payloadBuffer); + return _subscriptionManager.PublishToClientHandlersAsync(subject, replyTo, sid, headersBuffer, payloadBuffer); } internal void ResetPongCount() @@ -258,34 +289,6 @@ internal ValueTask UnsubscribeAsync(int sid) return default; } - internal void OnMessageDropped(NatsSubBase natsSub, int pending, NatsMsg msg) - { - var subject = msg.Subject; - _logger.LogWarning("Dropped message from {Subject} with {Pending} pending messages", subject, pending); - _eventChannel.Writer.TryWrite((NatsEvent.MessageDropped, new NatsMessageDroppedEventArgs(natsSub, pending, subject, msg.ReplyTo, msg.Headers, msg.Data))); - } - - internal BoundedChannelOptions GetChannelOpts(NatsOpts connectionOpts, NatsSubChannelOpts? subChannelOpts) - { - if (subChannelOpts is { } overrideOpts) - { - return new BoundedChannelOptions(overrideOpts.Capacity ?? - _defaultSubscriptionChannelOpts.Capacity) - { - AllowSynchronousContinuations = - _defaultSubscriptionChannelOpts.AllowSynchronousContinuations, - FullMode = - overrideOpts.FullMode ?? _defaultSubscriptionChannelOpts.FullMode, - SingleWriter = _defaultSubscriptionChannelOpts.SingleWriter, - SingleReader = _defaultSubscriptionChannelOpts.SingleReader, - }; - } - else - { - return _defaultSubscriptionChannelOpts; - } - } - private async ValueTask InitialConnectAsync() { Debug.Assert(ConnectionState == NatsConnectionState.Connecting, "Connection state"); @@ -465,7 +468,7 @@ private async ValueTask SetupReaderWriterAsync(bool reconnect) if (reconnect) { // Reestablish subscriptions and consumers - reconnectTask = SubscriptionManager.WriteReconnectCommandsAsync(priorityCommandWriter.CommandWriter).AsTask(); + reconnectTask = _subscriptionManager.WriteReconnectCommandsAsync(priorityCommandWriter.CommandWriter).AsTask(); } // receive COMMAND response (PONG or ERROR) diff --git a/src/NATS.Client.Core/NatsSub.cs b/src/NATS.Client.Core/NatsSub.cs index 0d5e51ec0..4c42ae971 100644 --- a/src/NATS.Client.Core/NatsSub.cs +++ b/src/NATS.Client.Core/NatsSub.cs @@ -9,9 +9,9 @@ public sealed class NatsSub : NatsSubBase, INatsSub { private readonly Channel> _msgs; - internal NatsSub( - NatsConnection connection, - ISubscriptionManager manager, + public NatsSub( + INatsConnection connection, + INatsSubscriptionManager manager, string subject, string? queueGroup, NatsSubOpts? opts, @@ -20,7 +20,7 @@ internal NatsSub( : base(connection, manager, subject, queueGroup, opts, cancellationToken) { _msgs = Channel.CreateBounded>( - connection.GetChannelOpts(connection.Opts, opts?.ChannelOpts), + connection.GetBoundedChannelOpts(opts?.ChannelOpts), msg => Connection.OnMessageDropped(this, _msgs?.Reader.Count ?? 0, msg)); Msgs = new ActivityEndingMsgReader(_msgs.Reader, this); diff --git a/src/NATS.Client.Core/NatsSubBase.cs b/src/NATS.Client.Core/NatsSubBase.cs index dbbcb7606..c6c2d699b 100644 --- a/src/NATS.Client.Core/NatsSubBase.cs +++ b/src/NATS.Client.Core/NatsSubBase.cs @@ -24,13 +24,16 @@ public enum NatsSubEndReason JetStreamError, } +/// +/// The base class for NATS subscriptions. +/// public abstract class NatsSubBase { private static readonly byte[] NoRespondersHeaderSequence = { (byte)' ', (byte)'5', (byte)'0', (byte)'3' }; private readonly ILogger _logger; private readonly object _gate = new(); private readonly bool _debug; - private readonly ISubscriptionManager _manager; + private readonly INatsSubscriptionManager _manager; private readonly Timer? _timeoutTimer; private readonly Timer? _idleTimeoutTimer; private readonly TimeSpan _idleTimeout; @@ -46,9 +49,18 @@ public abstract class NatsSubBase private int _pendingMsgs; private Exception? _exception; - internal NatsSubBase( - NatsConnection connection, - ISubscriptionManager manager, + /// + /// Creates a new instance of . + /// + /// NATS connection. + /// Subscription manager. + /// Subject to subscribe to. + /// Queue group name. + /// Subscription options. + /// Cancellation token. + protected NatsSubBase( + INatsConnection connection, + INatsSubscriptionManager manager, string subject, string? queueGroup, NatsSubOpts? opts, @@ -131,18 +143,32 @@ internal NatsSubBase( /// public string? QueueGroup { get; } + /// + /// Represents an exception that occurs during the execution of a NATS subscription. + /// public Exception? Exception => Volatile.Read(ref _exception); // Hide from public API using explicit interface implementations // since INatsSub is marked as internal. public int? PendingMsgs => _pendingMsgs == -1 ? null : Volatile.Read(ref _pendingMsgs); + /// + /// The reason for the subscription ending. + /// public NatsSubEndReason EndReason => (NatsSubEndReason)Volatile.Read(ref _endReasonRaw); internal NatsSubOpts? Opts { get; private set; } - protected NatsConnection Connection { get; } + /// + /// Represents a connection to the NATS server. + /// + protected INatsConnection Connection { get; } + /// + /// Signals that the subscription is ready to receive messages. + /// Override this method to perform any initialization logic. + /// + /// A that represents the asynchronous operation. public virtual ValueTask ReadyAsync() { // Let idle timer start with the first message, in case @@ -191,6 +217,11 @@ public ValueTask UnsubscribeAsync() return _manager.RemoveAsync(this); } + /// + /// Disposes the instance asynchronously. + /// + /// A representing the asynchronous disposal operation. + /// Thrown when an exception occurs during disposal. public virtual ValueTask DisposeAsync() { lock (_gate) @@ -223,6 +254,14 @@ public virtual ValueTask DisposeAsync() return unsubscribeAsync; } + /// + /// Called when a message is received for the subscription. + /// Calls to process the message handling any exceptions. + /// + /// Subject received for this subscription. + /// Reply subject received for this subscription. + /// Headers buffer received for this subscription. + /// Payload buffer received for this subscription. public virtual async ValueTask ReceiveAsync(string subject, string? replyTo, ReadOnlySequence? headersBuffer, ReadOnlySequence payloadBuffer) { ResetIdleTimeout(); @@ -300,12 +339,19 @@ public virtual async ValueTask ReceiveAsync(string subject, string? replyTo, Rea /// protected abstract ValueTask ReceiveInternalAsync(string subject, string? replyTo, ReadOnlySequence? headersBuffer, ReadOnlySequence payloadBuffer); + /// + /// Sets the exception that caused the subscription to end. + /// + /// Exception that caused the subscription to end. protected void SetException(Exception exception) { Interlocked.Exchange(ref _exception, exception); EndSubscription(NatsSubEndReason.Exception); } + /// + /// Resets the idle timeout timer. + /// protected void ResetIdleTimeout() { _idleTimeoutTimer?.Change(dueTime: _idleTimeout, period: Timeout.InfiniteTimeSpan); @@ -318,6 +364,9 @@ protected void ResetIdleTimeout() } } + /// + /// Decrements the maximum number of messages. + /// protected void DecrementMaxMsgs() { if (!_countPendingMsgs) @@ -337,6 +386,10 @@ protected void DecrementMaxMsgs() /// protected abstract void TryComplete(); + /// + /// Ends the subscription with the specified reason. + /// + /// Reason for ending the subscription. protected void EndSubscription(NatsSubEndReason reason) { if (_debug) diff --git a/src/NATS.Client.JetStream/INatsJSContext.cs b/src/NATS.Client.JetStream/INatsJSContext.cs index 0d48abd58..2c51a20cb 100644 --- a/src/NATS.Client.JetStream/INatsJSContext.cs +++ b/src/NATS.Client.JetStream/INatsJSContext.cs @@ -6,6 +6,11 @@ namespace NATS.Client.JetStream; public interface INatsJSContext { + /// + /// Connection to the NATS server. + /// + INatsConnection Connection { get; } + /// /// Creates new ordered consumer. /// diff --git a/src/NATS.Client.JetStream/Internal/NatsJSOrderedConsume.cs b/src/NATS.Client.JetStream/Internal/NatsJSOrderedConsume.cs index e7c9997b6..0969e4bd5 100644 --- a/src/NATS.Client.JetStream/Internal/NatsJSOrderedConsume.cs +++ b/src/NATS.Client.JetStream/Internal/NatsJSOrderedConsume.cs @@ -95,7 +95,7 @@ public NatsJSOrderedConsume( // This channel is used to pass messages to the user from the subscription. _userMsgs = Channel.CreateBounded>( - Connection.GetChannelOpts(Connection.Opts, opts?.ChannelOpts), + Connection.GetBoundedChannelOpts(opts?.ChannelOpts), msg => Connection.OnMessageDropped(this, _userMsgs?.Reader.Count ?? 0, msg.Msg)); Msgs = _userMsgs.Reader; diff --git a/src/NATS.Client.JetStream/Internal/NatsJSOrderedPushConsumer.cs b/src/NATS.Client.JetStream/Internal/NatsJSOrderedPushConsumer.cs index 14d1d9bff..03e8e51bf 100644 --- a/src/NATS.Client.JetStream/Internal/NatsJSOrderedPushConsumer.cs +++ b/src/NATS.Client.JetStream/Internal/NatsJSOrderedPushConsumer.cs @@ -52,7 +52,7 @@ internal class NatsJSOrderedPushConsumer private readonly NatsJSOrderedPushConsumerOpts _opts; private readonly NatsSubOpts? _subOpts; private readonly CancellationToken _cancellationToken; - private readonly NatsConnection _nats; + private readonly INatsConnection _nats; private readonly Channel> _commandChannel; private readonly Channel> _msgChannel; private readonly Channel _consumerCreateChannel; @@ -342,7 +342,7 @@ private async ValueTask CreatePushConsumer(string origin) } _sub = new NatsJSOrderedPushConsumerSub(_context, _commandChannel, _serializer, _subOpts, _cancellationToken); - await _context.Connection.SubAsync(_sub, _cancellationToken).ConfigureAwait(false); + await _context.Connection.AddSubAsync(_sub, _cancellationToken).ConfigureAwait(false); if (_debug) { @@ -419,7 +419,7 @@ internal class NatsJSOrderedPushConsumerSub : NatsSubBase { private readonly NatsJSContext _context; private readonly CancellationToken _cancellationToken; - private readonly NatsConnection _nats; + private readonly INatsConnection _nats; private readonly NatsHeaderParser _headerParser; private readonly INatsDeserialize _serializer; private readonly ChannelWriter> _commands; diff --git a/src/NATS.Client.JetStream/NatsClientExtensions.cs b/src/NATS.Client.JetStream/NatsClientExtensions.cs new file mode 100644 index 000000000..31e4e9809 --- /dev/null +++ b/src/NATS.Client.JetStream/NatsClientExtensions.cs @@ -0,0 +1,12 @@ +using NATS.Client.Core; + +namespace NATS.Client.JetStream; + +public static class NatsClientExtensions +{ + public static INatsJSContext CreateJetStreamContext(this INatsClient client) + => CreateJetStreamContext(client.Connection); + + public static INatsJSContext CreateJetStreamContext(this INatsConnection connection) + => new NatsJSContext(connection); +} diff --git a/src/NATS.Client.JetStream/NatsJSConsumer.cs b/src/NATS.Client.JetStream/NatsJSConsumer.cs index e1c7b400d..cc3e35e2e 100644 --- a/src/NATS.Client.JetStream/NatsJSConsumer.cs +++ b/src/NATS.Client.JetStream/NatsJSConsumer.cs @@ -307,7 +307,7 @@ internal async ValueTask> ConsumeInternalAsync(INatsDeserial notificationHandler: opts.NotificationHandler, cancellationToken: cancellationToken); - await _context.Connection.SubAsync(sub: sub, cancellationToken).ConfigureAwait(false); + await _context.Connection.AddSubAsync(sub: sub, cancellationToken).ConfigureAwait(false); // Start consuming with the first Pull Request await sub.CallMsgNextAsync( @@ -354,7 +354,7 @@ internal async ValueTask> OrderedConsumeInternalAsync idle: timeouts.IdleHeartbeat, cancellationToken: cancellationToken); - await _context.Connection.SubAsync(sub: sub, cancellationToken).ConfigureAwait(false); + await _context.Connection.AddSubAsync(sub: sub, cancellationToken).ConfigureAwait(false); // Start consuming with the first Pull Request await sub.CallMsgNextAsync( @@ -403,7 +403,7 @@ internal async ValueTask> FetchInternalAsync( idle: timeouts.IdleHeartbeat, cancellationToken: cancellationToken); - await _context.Connection.SubAsync(sub: sub, cancellationToken).ConfigureAwait(false); + await _context.Connection.AddSubAsync(sub: sub, cancellationToken).ConfigureAwait(false); await sub.CallMsgNextAsync( opts.NoWait diff --git a/src/NATS.Client.JetStream/NatsJSContext.cs b/src/NATS.Client.JetStream/NatsJSContext.cs index 78feb7fbf..66ed5cacc 100644 --- a/src/NATS.Client.JetStream/NatsJSContext.cs +++ b/src/NATS.Client.JetStream/NatsJSContext.cs @@ -12,8 +12,8 @@ public partial class NatsJSContext { private readonly ILogger _logger; - /// > - public NatsJSContext(NatsConnection connection) + /// > + public NatsJSContext(INatsConnection connection) : this(connection, new NatsJSOpts(connection.Opts)) { } @@ -23,14 +23,14 @@ public NatsJSContext(NatsConnection connection) /// /// A NATS server connection to access the JetStream APIs, publishers and consumers. /// Context wide JetStream options. - public NatsJSContext(NatsConnection connection, NatsJSOpts opts) + public NatsJSContext(INatsConnection connection, NatsJSOpts opts) { Connection = connection; Opts = opts; _logger = connection.Opts.LoggerFactory.CreateLogger(); } - internal NatsConnection Connection { get; } + public INatsConnection Connection { get; } internal NatsJSOpts Opts { get; } @@ -119,7 +119,7 @@ public async ValueTask PublishAsync( for (var i = 0; i < retryMax; i++) { - await using var sub = await Connection.RequestSubAsync( + await using var sub = await Connection.CreateRequestSubAsync( subject: subject, data: data, headers: headers, @@ -213,7 +213,7 @@ public async ValueTask PublishConcurrentAsync( opts ??= NatsJSPubOpts.Default; - var sub = await Connection.RequestSubAsync( + var sub = await Connection.CreateRequestSubAsync( subject: subject, data: data, headers: headers, @@ -289,7 +289,7 @@ internal async ValueTask> JSRequestAsync( + await using var sub = await Connection.CreateRequestSubAsync( subject: subject, data: request, headers: default, diff --git a/src/NATS.Client.KeyValueStore/Internal/NatsKVWatchSub.cs b/src/NATS.Client.KeyValueStore/Internal/NatsKVWatchSub.cs index fb0d2c7eb..e2f7f1d3b 100644 --- a/src/NATS.Client.KeyValueStore/Internal/NatsKVWatchSub.cs +++ b/src/NATS.Client.KeyValueStore/Internal/NatsKVWatchSub.cs @@ -9,7 +9,7 @@ internal class NatsKVWatchSub : NatsSubBase { private readonly NatsJSContext _context; private readonly CancellationToken _cancellationToken; - private readonly NatsConnection _nats; + private readonly INatsConnection _nats; private readonly NatsHeaderParser _headerParser; private readonly INatsDeserialize _serializer; private readonly ChannelWriter> _commands; diff --git a/src/NATS.Client.KeyValueStore/Internal/NatsKVWatcher.cs b/src/NATS.Client.KeyValueStore/Internal/NatsKVWatcher.cs index b5599c310..30f54dd00 100644 --- a/src/NATS.Client.KeyValueStore/Internal/NatsKVWatcher.cs +++ b/src/NATS.Client.KeyValueStore/Internal/NatsKVWatcher.cs @@ -36,7 +36,7 @@ internal sealed class NatsKVWatcher : IAsyncDisposable private readonly CancellationToken _cancellationToken; private readonly string _keyBase; private readonly string[] _filters; - private readonly NatsConnection _nats; + private readonly INatsConnection _nats; private readonly Channel> _commandChannel; private readonly Channel> _entryChannel; private readonly Channel _consumerCreateChannel; @@ -363,7 +363,7 @@ private async ValueTask CreatePushConsumer(string origin) } _sub = new NatsKVWatchSub(_context, _commandChannel, _serializer, _subOpts, _cancellationToken); - await _context.Connection.SubAsync(_sub, _cancellationToken).ConfigureAwait(false); + await _context.Connection.AddSubAsync(_sub, _cancellationToken).ConfigureAwait(false); if (_debug) { diff --git a/src/NATS.Client.Services/NatsSvcEndPoint.cs b/src/NATS.Client.Services/NatsSvcEndPoint.cs index cdecd7578..8e6947c2f 100644 --- a/src/NATS.Client.Services/NatsSvcEndPoint.cs +++ b/src/NATS.Client.Services/NatsSvcEndPoint.cs @@ -178,7 +178,7 @@ public override async ValueTask DisposeAsync() internal override void SetLastError(string error) => Interlocked.Exchange(ref _lastError, error); internal ValueTask StartAsync(CancellationToken cancellationToken) => - _nats.SubAsync(this, cancellationToken); + _nats.AddSubAsync(this, cancellationToken); protected override ValueTask ReceiveInternalAsync( string subject, diff --git a/tests/NATS.Client.Core.Tests/LowLevelApiTest.cs b/tests/NATS.Client.Core.Tests/LowLevelApiTest.cs index 3d7efe29b..252306f72 100644 --- a/tests/NATS.Client.Core.Tests/LowLevelApiTest.cs +++ b/tests/NATS.Client.Core.Tests/LowLevelApiTest.cs @@ -18,7 +18,7 @@ public async Task Sub_custom_builder_test() var subject = "foo.*"; var builder = new NatsSubCustomTestBuilder(_output); var sub = builder.Build(subject, default, nats, nats.SubscriptionManager); - await nats.SubAsync(sub); + await nats.AddSubAsync(sub); await Retry.Until( "subscription is ready", @@ -44,7 +44,7 @@ private class NatsSubTest : NatsSubBase private readonly NatsSubCustomTestBuilder _builder; private readonly ITestOutputHelper _output; - public NatsSubTest(string subject, NatsConnection connection, NatsSubCustomTestBuilder builder, ITestOutputHelper output, ISubscriptionManager manager) + public NatsSubTest(string subject, NatsConnection connection, NatsSubCustomTestBuilder builder, ITestOutputHelper output, INatsSubscriptionManager manager) : base(connection, manager, subject, default, default) { _builder = builder; @@ -110,7 +110,7 @@ public IEnumerable Messages } } - public NatsSubTest Build(string subject, NatsSubOpts? opts, NatsConnection connection, ISubscriptionManager manager) + public NatsSubTest Build(string subject, NatsSubOpts? opts, NatsConnection connection, INatsSubscriptionManager manager) { return new NatsSubTest(subject, connection, builder: this, _output, manager); } diff --git a/tests/NATS.Client.Core.Tests/ProtocolTest.cs b/tests/NATS.Client.Core.Tests/ProtocolTest.cs index 6490a9976..37866de35 100644 --- a/tests/NATS.Client.Core.Tests/ProtocolTest.cs +++ b/tests/NATS.Client.Core.Tests/ProtocolTest.cs @@ -334,7 +334,7 @@ public async Task Reconnect_with_sub_and_additional_commands() var sync = 0; await using var sub = new NatsSubReconnectTest(nats, subject, i => Interlocked.Exchange(ref sync, i)); - await nats.SubAsync(sub); + await nats.AddSubAsync(sub); await Retry.Until( "subscribed", diff --git a/tests/NATS.Client.Core.Tests/RequestReplyTest.cs b/tests/NATS.Client.Core.Tests/RequestReplyTest.cs index 83846cdf5..0f9eee201 100644 --- a/tests/NATS.Client.Core.Tests/RequestReplyTest.cs +++ b/tests/NATS.Client.Core.Tests/RequestReplyTest.cs @@ -150,7 +150,7 @@ public async Task Request_reply_many_test_overall_timeout() var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); var opts = new NatsSubOpts { Timeout = TimeSpan.FromSeconds(4) }; await using var rep = - await nats.RequestSubAsync("foo", 4, replyOpts: opts, cancellationToken: cts.Token); + await nats.CreateRequestSubAsync("foo", 4, replyOpts: opts, cancellationToken: cts.Token); await foreach (var msg in rep.Msgs.ReadAllAsync(cts.Token)) { Assert.Equal(results[count++], msg.Data); @@ -184,7 +184,7 @@ public async Task Request_reply_many_test_idle_timeout() var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); var opts = new NatsSubOpts { IdleTimeout = TimeSpan.FromSeconds(3) }; await using var rep = - await nats.RequestSubAsync("foo", 3, replyOpts: opts, cancellationToken: cts.Token); + await nats.CreateRequestSubAsync("foo", 3, replyOpts: opts, cancellationToken: cts.Token); await foreach (var msg in rep.Msgs.ReadAllAsync(cts.Token)) { Assert.Equal(results[count++], msg.Data); @@ -214,7 +214,7 @@ public async Task Request_reply_many_test_start_up_timeout() var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); var opts = new NatsSubOpts { StartUpTimeout = TimeSpan.FromSeconds(1) }; await using var rep = - await nats.RequestSubAsync("foo", 2, replyOpts: opts, cancellationToken: cts.Token); + await nats.CreateRequestSubAsync("foo", 2, replyOpts: opts, cancellationToken: cts.Token); await foreach (var msg in rep.Msgs.ReadAllAsync(cts.Token)) { count++; @@ -247,7 +247,7 @@ public async Task Request_reply_many_test_max_count() var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); var opts = new NatsSubOpts { MaxMsgs = 2 }; await using var rep = - await nats.RequestSubAsync("foo", 1, replyOpts: opts, cancellationToken: cts.Token); + await nats.CreateRequestSubAsync("foo", 1, replyOpts: opts, cancellationToken: cts.Token); await foreach (var msg in rep.Msgs.ReadAllAsync(cts.Token)) { Assert.Equal(results[count++], msg.Data); diff --git a/tests/NATS.Client.JetStream.Tests/NatsJsContextFactoryTest.cs b/tests/NATS.Client.JetStream.Tests/NatsJsContextFactoryTest.cs index 879b27d8e..13ac111e2 100644 --- a/tests/NATS.Client.JetStream.Tests/NatsJsContextFactoryTest.cs +++ b/tests/NATS.Client.JetStream.Tests/NatsJsContextFactoryTest.cs @@ -1,3 +1,5 @@ +using System.Text; +using System.Threading.Channels; using NATS.Client.Core.Tests; namespace NATS.Client.JetStream.Tests; @@ -100,6 +102,10 @@ public class MockConnection : INatsConnection public NatsConnectionState ConnectionState { get; } = NatsConnectionState.Closed; + public INatsSubscriptionManager SubscriptionManager { get; } = new TestSubscriptionManager(); + + public NatsHeaderParser HeaderParser { get; } = new NatsHeaderParser(Encoding.UTF8); + public ValueTask PingAsync(CancellationToken cancellationToken = default) => throw new NotImplementedException(); public ValueTask PublishAsync(string subject, T data, NatsHeaders? headers = default, string? replyTo = default, INatsSerialize? serializer = default, NatsPubOpts? opts = default, CancellationToken cancellationToken = default) => throw new NotImplementedException(); @@ -138,8 +144,22 @@ public IAsyncEnumerable> RequestManyAsync( CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public void OnMessageDropped(NatsSubBase natsSub, int pending, NatsMsg msg) => throw new NotImplementedException(); + + public ValueTask AddSubAsync(NatsSubBase sub, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + + public BoundedChannelOptions GetBoundedChannelOpts(NatsSubChannelOpts? subChannelOpts) => throw new NotImplementedException(); + + public ValueTask> CreateRequestSubAsync(string subject, TRequest? data, NatsHeaders? headers = default, INatsSerialize? requestSerializer = default, INatsDeserialize? replySerializer = default, NatsPubOpts? requestOpts = default, NatsSubOpts? replyOpts = default, CancellationToken cancellationToken = default) => + throw new NotImplementedException(); + public ValueTask ConnectAsync() => throw new NotImplementedException(); public ValueTask DisposeAsync() => throw new NotImplementedException(); } } + +public class TestSubscriptionManager : INatsSubscriptionManager +{ + public ValueTask RemoveAsync(NatsSubBase sub) => throw new NotImplementedException(); +} diff --git a/tests/NATS.Client.KeyValueStore.Tests/NatsKVContextFactoryTest.cs b/tests/NATS.Client.KeyValueStore.Tests/NatsKVContextFactoryTest.cs index 591e0dbf7..0270ae9f9 100644 --- a/tests/NATS.Client.KeyValueStore.Tests/NatsKVContextFactoryTest.cs +++ b/tests/NATS.Client.KeyValueStore.Tests/NatsKVContextFactoryTest.cs @@ -48,6 +48,8 @@ public void Create_Context_WithMockConnection_Test() public class MockJsContext : INatsJSContext { + public INatsConnection Connection { get; } = new NatsConnection(); + public ValueTask CreateOrderedConsumerAsync(string stream, NatsJSOrderedConsumerOpts? opts = default, CancellationToken cancellationToken = default) => throw new NotImplementedException(); public ValueTask CreateOrUpdateConsumerAsync(string stream, ConsumerConfig config, CancellationToken cancellationToken = default) => throw new NotImplementedException(); diff --git a/tests/NATS.Client.ObjectStore.Tests/NatsObjContextFactoryTest.cs b/tests/NATS.Client.ObjectStore.Tests/NatsObjContextFactoryTest.cs index 1b4053260..2ed5bb6f8 100644 --- a/tests/NATS.Client.ObjectStore.Tests/NatsObjContextFactoryTest.cs +++ b/tests/NATS.Client.ObjectStore.Tests/NatsObjContextFactoryTest.cs @@ -48,6 +48,8 @@ public void Create_Context_WithMockConnection_Test() public class MockJsContext : INatsJSContext { + public INatsConnection Connection { get; } = new NatsConnection(); + public ValueTask CreateOrderedConsumerAsync(string stream, NatsJSOrderedConsumerOpts? opts = default, CancellationToken cancellationToken = default) => throw new NotImplementedException(); public ValueTask CreateOrUpdateConsumerAsync(string stream, ConsumerConfig config, CancellationToken cancellationToken = default) => throw new NotImplementedException(); diff --git a/tests/NATS.Client.Simplified.Tests/ClientTest.cs b/tests/NATS.Client.Simplified.Tests/ClientTest.cs index 28b6875e5..6d1a2e478 100644 --- a/tests/NATS.Client.Simplified.Tests/ClientTest.cs +++ b/tests/NATS.Client.Simplified.Tests/ClientTest.cs @@ -1,5 +1,7 @@ using System.Text; using NATS.Client.Core.Tests; +using NATS.Client.JetStream; +using NATS.Client.JetStream.Models; using NATS.Net; // ReSharper disable AccessToDisposedClosure @@ -10,7 +12,7 @@ public class ClientTest [Fact] public async Task Client_works_with_all_expected_types_and_falls_back_to_JSON() { - await using var server = NatsServer.Start(); + await using var server = NatsServer.StartJS(); await using var client = new NatsClient(server.ClientUrl); CancellationTokenSource ctsTestTimeout = new(TimeSpan.FromSeconds(10)); @@ -194,7 +196,13 @@ await Retry.Until( } // Use JetStream by referencing NATS.Client.JetStream package - // var js = client.GetJetStream(); + var js = client.CreateJetStreamContext(); + await js.CreateStreamAsync(new StreamConfig("test", ["test.>"]), ctsTestTimeout.Token); + await foreach (var stream in js.ListStreamsAsync(cancellationToken: ctsTestTimeout.Token)) + { + Assert.Equal("test", stream.Info.Config.Name); + } + ctsStop.Cancel(); await Task.WhenAll(task1, task2, task3, task4, task5, task6); diff --git a/tests/NATS.Client.Simplified.Tests/NATS.Client.Simplified.Tests.csproj b/tests/NATS.Client.Simplified.Tests/NATS.Client.Simplified.Tests.csproj index b505b1ca8..e189806b8 100644 --- a/tests/NATS.Client.Simplified.Tests/NATS.Client.Simplified.Tests.csproj +++ b/tests/NATS.Client.Simplified.Tests/NATS.Client.Simplified.Tests.csproj @@ -34,6 +34,7 @@ +