JetStream pull consumer redesign

NATS.Client.sln

@@ -65,6 +65,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Schema.Generation", "tools\
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NATS.Client.Perf", "tests\NATS.Client.Perf\NATS.Client.Perf.csproj", "{ADF66CBA-4F3E-4E91-9842-E194E3BC06A1}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Example.JetStream.PullConsumer", "sandbox\Example.JetStream.PullConsumer\Example.JetStream.PullConsumer.csproj", "{3A9FC281-3B81-4D63-A76B-E1127C1D2241}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -159,6 +161,10 @@ Global
 		{ADF66CBA-4F3E-4E91-9842-E194E3BC06A1}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{ADF66CBA-4F3E-4E91-9842-E194E3BC06A1}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{ADF66CBA-4F3E-4E91-9842-E194E3BC06A1}.Release|Any CPU.Build.0 = Release|Any CPU
+		{3A9FC281-3B81-4D63-A76B-E1127C1D2241}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{3A9FC281-3B81-4D63-A76B-E1127C1D2241}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{3A9FC281-3B81-4D63-A76B-E1127C1D2241}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{3A9FC281-3B81-4D63-A76B-E1127C1D2241}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -187,6 +193,7 @@ Global
 		{90E5BF38-70C1-460A-9177-CE42815BDBF5} = {C526E8AB-739A-48D7-8FC4-048978C9B650}
 		{B7DD4A9C-2D24-4772-951E-86A665C59ADF} = {BD234E2E-F51A-4B18-B8BE-8AF6D546BF87}
 		{ADF66CBA-4F3E-4E91-9842-E194E3BC06A1} = {C526E8AB-739A-48D7-8FC4-048978C9B650}
+		{3A9FC281-3B81-4D63-A76B-E1127C1D2241} = {95A69671-16CA-4133-981C-CC381B7AAA30}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {8CBB7278-D093-448E-B3DE-B5991209A1AA}

sandbox/Example.JetStream.PullConsumer/Example.JetStream.PullConsumer.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <OutputType>Exe</OutputType>
+        <TargetFramework>net6.0</TargetFramework>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+    </PropertyGroup>
+
+    <ItemGroup>
+      <ProjectReference Include="..\..\src\NATS.Client.JetStream\NATS.Client.JetStream.csproj" />
+    </ItemGroup>
+
+</Project>

sandbox/Example.JetStream.PullConsumer/Program.cs

@@ -0,0 +1,78 @@
+using System.Buffers;
+using System.Runtime.CompilerServices;
+using System.Text;
+using Microsoft.Extensions.Logging;
+using NATS.Client.Core;
+using NATS.Client.JetStream;
+using NATS.Client.JetStream.Internal;
+using NATS.Client.JetStream.Models;
+
+var options = NatsOptions.Default with { LoggerFactory = new MinimumConsoleLoggerFactory(LogLevel.Error) };
+
+await using var nats = new NatsConnection(options);
+
+var js = new NatsJSContext(nats);
+
+var stream = await js.CreateStreamAsync("s1", new[] { "s1.*" });
+var consumer = await js.CreateConsumerAsync("s1", "c1");
+await using var sub = await consumer.CreateSubscription<RawData>(
+    controlHandler: async (thisSub, msg) =>
+    {
+        Console.WriteLine($"____");
+        Console.WriteLine($"{msg.Headers?.Code} {msg.Headers?.MessageText}");
+        var error = msg.Error?.Error;
+        if (error != null)
+            Console.WriteLine($"Error: {error}");
+    },
+    reconnectRequestFactory: () => default,
+    heartBeat: TimeSpan.FromSeconds(5),
+    opts: new NatsSubOpts { Serializer = new RawDataSerializer() });
+
+await sub.CallMsgNextAsync(new ConsumerGetnextRequest
+{
+    Batch = 100,
+    IdleHeartbeat = TimeSpan.FromSeconds(3).ToNanos(),
+    Expires = TimeSpan.FromSeconds(12).ToNanos(),
+});
+
+await foreach (var jsMsg in sub.Msgs.ReadAllAsync())
+{
+    var msg = jsMsg.Msg;
+    await jsMsg.AckAsync();
+    Console.WriteLine($"____");
+    Console.WriteLine($"subject: {msg.Subject}");
+    Console.WriteLine($"data: {msg.Data}");
+}
+
+class RawData
+{
+    public RawData(byte[] buffer) => Buffer = buffer;
+
+    public byte[] Buffer { get; }
+
+    public override string ToString() => Encoding.ASCII.GetString(Buffer);
+}
+
+class RawDataSerializer : INatsSerializer
+{
+    public int Serialize<T>(ICountableBufferWriter bufferWriter, T? value)
+    {
+        if (value is RawData data)
+        {
+            bufferWriter.Write(data.Buffer);
+            return data.Buffer.Length;
+        }
+
+        throw new Exception($"Can only work with '{typeof(RawData)}'");
+    }
+
+    public T? Deserialize<T>(in ReadOnlySequence<byte> buffer) => (T?)Deserialize(buffer, typeof(T));
+
+    public object? Deserialize(in ReadOnlySequence<byte> buffer, Type type)
+    {
+        if (type != typeof(RawData))
+            throw new Exception($"Can only work with '{typeof(RawData)}'");
+
+        return new RawData(buffer.ToArray());
+    }
+}

src/NATS.Client.Core/Internal/NatsReadProtocolProcessor.cs

@@ -27,13 +27,13 @@ internal sealed class NatsReadProtocolProcessor : IAsyncDisposable
     public NatsReadProtocolProcessor(ISocketConnection socketConnection, NatsConnection connection, TaskCompletionSource waitForInfoSignal, TaskCompletionSource waitForPongOrErrorSignal, Task infoParsed)
     {
         _connection = connection;
-        _logger = connection.Options.LoggerFactory.CreateLogger<NatsReadProtocolProcessor>();
+        _logger = connection.Opts.LoggerFactory.CreateLogger<NatsReadProtocolProcessor>();
         _trace = _logger.IsEnabled(LogLevel.Trace);
         _waitForInfoSignal = waitForInfoSignal;
         _waitForPongOrErrorSignal = waitForPongOrErrorSignal;
         _infoParsed = infoParsed;
         _pingCommands = new ConcurrentQueue<AsyncPingCommand>();
-        _socketReader = new SocketReader(socketConnection, connection.Options.ReaderBufferSize, connection.Counter, connection.Options.LoggerFactory);
+        _socketReader = new SocketReader(socketConnection, connection.Opts.ReaderBufferSize, connection.Counter, connection.Opts.LoggerFactory);
         _readLoop = Task.Run(ReadLoopAsync);
     }
 

src/NATS.Client.Core/Internal/SubscriptionManager.cs

@@ -6,7 +6,7 @@
 
 namespace NATS.Client.Core.Internal;
 
-internal interface ISubscriptionManager
+public interface ISubscriptionManager
 {
     public ValueTask RemoveAsync(NatsSubBase sub);
 }
@@ -36,11 +36,11 @@ public SubscriptionManager(NatsConnection connection, string inboxPrefix)
     {
         _connection = connection;
         _inboxPrefix = inboxPrefix;
-        _logger = _connection.Options.LoggerFactory.CreateLogger<SubscriptionManager>();
+        _logger = _connection.Opts.LoggerFactory.CreateLogger<SubscriptionManager>();
         _cts = new CancellationTokenSource();
-        _cleanupInterval = _connection.Options.SubscriptionCleanUpInterval;
+        _cleanupInterval = _connection.Opts.SubscriptionCleanUpInterval;
         _timer = Task.Run(CleanupAsync);
-        InboxSubBuilder = new InboxSubBuilder(connection.Options.LoggerFactory.CreateLogger<InboxSubBuilder>());
+        InboxSubBuilder = new InboxSubBuilder(connection.Opts.LoggerFactory.CreateLogger<InboxSubBuilder>());
         _inboxSubSentinel = new InboxSub(InboxSubBuilder, nameof(_inboxSubSentinel), default, connection, this);
         _inboxSub = _inboxSubSentinel;
     }

src/NATS.Client.Core/NatsConnection.Publish.cs

@@ -26,7 +26,7 @@ public ValueTask PublishAsync(in NatsMsg msg, NatsPubOpts? opts = default, Cance
     /// <inheritdoc />
     public ValueTask PublishAsync<T>(string subject, T? data, NatsPubOpts? opts = default, CancellationToken cancellationToken = default)
     {
-        var serializer = opts?.Serializer ?? Options.Serializer;
+        var serializer = opts?.Serializer ?? Opts.Serializer;
         if (opts?.WaitUntilSent ?? false)
         {
             return PubModelAsync<T>(subject, data, serializer, opts?.ReplyTo, opts?.Headers, cancellationToken);

src/NATS.Client.Core/NatsConnection.RequestReply.cs

@@ -112,7 +112,7 @@ private NatsSubOpts SetReplyOptsDefaults(in NatsSubOpts? replyOpts)
 
         if ((opts.Timeout ?? default) == default)
         {
-            opts = opts with { Timeout = Options.RequestTimeout };
+            opts = opts with { Timeout = Opts.RequestTimeout };
         }
 
         return opts;

src/NATS.Client.Core/NatsConnection.RequestSub.cs

@@ -28,14 +28,14 @@ internal async ValueTask<INatsSub<TReply>> RequestSubAsync<TRequest, TReply>(
     {
         var replyTo = $"{InboxPrefix}{Guid.NewGuid():n}";
 
-        var replySerializer = replyOpts?.Serializer ?? Options.Serializer;
+        var replySerializer = replyOpts?.Serializer ?? Opts.Serializer;
         var sub = new NatsSub<TReply>(this, SubscriptionManager.InboxSubBuilder, replyTo, replyOpts, replySerializer);
         await SubAsync(replyTo, replyOpts, sub, cancellationToken).ConfigureAwait(false);
 
         await PubModelAsync(
             subject,
             data,
-            requestOpts?.Serializer ?? Options.Serializer,
+            requestOpts?.Serializer ?? Opts.Serializer,
             replyTo,
             requestOpts?.Headers,
             cancellationToken).ConfigureAwait(false);

src/NATS.Client.Core/NatsConnection.Subscribe.cs

@@ -16,7 +16,7 @@ public async ValueTask<INatsSub> SubscribeAsync(string subject, NatsSubOpts? opt
     /// <inheritdoc />
     public async ValueTask<INatsSub<T>> SubscribeAsync<T>(string subject, NatsSubOpts? opts = default, CancellationToken cancellationToken = default)
     {
-        var serializer = opts?.Serializer ?? Options.Serializer;
+        var serializer = opts?.Serializer ?? Opts.Serializer;
         var sub = new NatsSub<T>(this, SubscriptionManager.GetManagerFor(subject), subject, opts, serializer);
         await SubAsync(subject, opts, sub, cancellationToken).ConfigureAwait(false);
         return sub;

src/NATS.Client.Core/NatsConnection.Util.cs

@@ -70,7 +70,7 @@ internal void PostDirectWrite(DirectWriteCommand command)
     internal CancellationTimer GetCancellationTimer(CancellationToken cancellationToken, TimeSpan timeout = default)
     {
         if (timeout == default)
-            timeout = Options.CommandTimeout;
+            timeout = Opts.CommandTimeout;
         return _cancellationTimerPool.Start(timeout, cancellationToken);
     }
 

src/NATS.Client.Core/NatsConnection.cs

@@ -56,23 +56,23 @@ public NatsConnection()
     {
     }
 
-    public NatsConnection(NatsOptions options)
+    public NatsConnection(NatsOptions opts)
     {
-        Options = options;
+        Opts = opts;
         ConnectionState = NatsConnectionState.Closed;
         _waitForOpenConnection = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
         _disposedCancellationTokenSource = new CancellationTokenSource();
-        _pool = new ObjectPool(options.ObjectPoolSize);
+        _pool = new ObjectPool(opts.ObjectPoolSize);
         _cancellationTimerPool = new CancellationTimerPool(_pool, _disposedCancellationTokenSource.Token);
-        _name = options.Name;
+        _name = opts.Name;
         Counter = new ConnectionStatsCounter();
-        _writerState = new WriterState(options);
+        _writerState = new WriterState(opts);
         _commandWriter = _writerState.CommandBuffer.Writer;
-        InboxPrefix = $"{options.InboxPrefix}.{Guid.NewGuid():n}.";
+        InboxPrefix = $"{opts.InboxPrefix}.{Guid.NewGuid():n}.";
         SubscriptionManager = new SubscriptionManager(this, InboxPrefix);
-        _logger = options.LoggerFactory.CreateLogger<NatsConnection>();
-        _clientOptions = ClientOptions.Create(Options);
-        HeaderParser = new HeaderParser(options.HeaderEncoding);
+        _logger = opts.LoggerFactory.CreateLogger<NatsConnection>();
+        _clientOptions = ClientOptions.Create(Opts);
+        HeaderParser = new HeaderParser(opts.HeaderEncoding);
     }
 
     // events
@@ -82,7 +82,7 @@ public NatsConnection(NatsOptions options)
 
     public event EventHandler<string>? ReconnectFailed;
 
-    public NatsOptions Options { get; }
+    public NatsOptions Opts { get; }
 
     public NatsConnectionState ConnectionState { get; private set; }
 
@@ -221,15 +221,15 @@ private async ValueTask InitialConnectAsync()
     {
         Debug.Assert(ConnectionState == NatsConnectionState.Connecting, "Connection state");
 
-        var uris = Options.GetSeedUris();
-        if (Options.TlsOptions.Disabled && uris.Any(u => u.IsTls))
+        var uris = Opts.GetSeedUris();
+        if (Opts.TlsOptions.Disabled && uris.Any(u => u.IsTls))
             throw new NatsException($"URI {uris.First(u => u.IsTls)} requires TLS but TlsOptions.Disabled is set to true");
-        if (Options.TlsOptions.Required)
-            _tlsCerts = new TlsCerts(Options.TlsOptions);
+        if (Opts.TlsOptions.Required)
+            _tlsCerts = new TlsCerts(Opts.TlsOptions);
 
-        if (!Options.AuthOptions.IsAnonymous)
+        if (!Opts.AuthOptions.IsAnonymous)
         {
-            _userCredentials = new UserCredentials(Options.AuthOptions);
+            _userCredentials = new UserCredentials(Opts.AuthOptions);
         }
 
         foreach (var uri in uris)
@@ -247,13 +247,13 @@ private async ValueTask InitialConnectAsync()
                 if (uri.IsWebSocket)
                 {
                     var conn = new WebSocketConnection();
-                    await conn.ConnectAsync(uri.Uri, Options.ConnectTimeout).ConfigureAwait(false);
+                    await conn.ConnectAsync(uri.Uri, Opts.ConnectTimeout).ConfigureAwait(false);
                     _socket = conn;
                 }
                 else
                 {
                     var conn = new TcpConnection();
-                    await conn.ConnectAsync(target.Host, target.Port, Options.ConnectTimeout).ConfigureAwait(false);
+                    await conn.ConnectAsync(target.Host, target.Port, Opts.ConnectTimeout).ConfigureAwait(false);
                     _socket = conn;
                 }
 
@@ -331,19 +331,19 @@ private async ValueTask SetupReaderWriterAsync(bool reconnect)
             // check to see if we should upgrade to TLS
             if (_socket is TcpConnection tcpConnection)
             {
-                if (Options.TlsOptions.Disabled && WritableServerInfo!.TlsRequired)
+                if (Opts.TlsOptions.Disabled && WritableServerInfo!.TlsRequired)
                 {
                     throw new NatsException(
                         $"Server {_currentConnectUri} requires TLS but TlsOptions.Disabled is set to true");
                 }
 
-                if (Options.TlsOptions.Required && !WritableServerInfo!.TlsRequired && !WritableServerInfo.TlsAvailable)
+                if (Opts.TlsOptions.Required && !WritableServerInfo!.TlsRequired && !WritableServerInfo.TlsAvailable)
                 {
                     throw new NatsException(
                         $"Server {_currentConnectUri} does not support TLS but TlsOptions.Disabled is set to true");
                 }
 
-                if (Options.TlsOptions.Required || WritableServerInfo!.TlsRequired || WritableServerInfo.TlsAvailable)
+                if (Opts.TlsOptions.Required || WritableServerInfo!.TlsRequired || WritableServerInfo.TlsAvailable)
                 {
                     // do TLS upgrade
                     // if the current URI is not a seed URI and is not a DNS hostname, check the server cert against the
@@ -364,7 +364,7 @@ private async ValueTask SetupReaderWriterAsync(bool reconnect)
                     _socketReader = null;
 
                     // upgrade TcpConnection to SslConnection
-                    var sslConnection = tcpConnection.UpgradeToSslStreamConnection(Options.TlsOptions, _tlsCerts);
+                    var sslConnection = tcpConnection.UpgradeToSslStreamConnection(Opts.TlsOptions, _tlsCerts);
                     await sslConnection.AuthenticateAsClientAsync(targetHost).ConfigureAwait(false);
                     _socket = sslConnection;
 
@@ -433,12 +433,12 @@ private async void ReconnectLoop()
             await DisposeSocketAsync(true).ConfigureAwait(false);
 
             var defaultScheme = _currentConnectUri!.Uri.Scheme;
-            var urls = (Options.NoRandomize
+            var urls = (Opts.NoRandomize
                 ? WritableServerInfo?.ClientConnectUrls?.Select(x => new NatsUri(x, false, defaultScheme)).Distinct().ToArray()
                 : WritableServerInfo?.ClientConnectUrls?.Select(x => new NatsUri(x, false, defaultScheme)).OrderBy(_ => Guid.NewGuid()).Distinct().ToArray())
                     ?? Array.Empty<NatsUri>();
             if (urls.Length == 0)
-                urls = Options.GetSeedUris();
+                urls = Opts.GetSeedUris();
 
             // add last.
             urls = urls.Where(x => x != _currentConnectUri).Append(_currentConnectUri).ToArray();
@@ -463,13 +463,13 @@ private async void ReconnectLoop()
                     if (url.IsWebSocket)
                     {
                         var conn = new WebSocketConnection();
-                        await conn.ConnectAsync(url.Uri, Options.ConnectTimeout).ConfigureAwait(false);
+                        await conn.ConnectAsync(url.Uri, Opts.ConnectTimeout).ConfigureAwait(false);
                         _socket = conn;
                     }
                     else
                     {
                         var conn = new TcpConnection();
-                        await conn.ConnectAsync(target.Host, target.Port, Options.ConnectTimeout).ConfigureAwait(false);
+                        await conn.ConnectAsync(target.Host, target.Port, Opts.ConnectTimeout).ConfigureAwait(false);
                         _socket = conn;
                     }
 
@@ -517,8 +517,8 @@ private async void ReconnectLoop()
 
     private async Task WaitWithJitterAsync()
     {
-        var jitter = Random.Shared.NextDouble() * Options.ReconnectJitter.TotalMilliseconds;
-        var waitTime = Options.ReconnectWait + TimeSpan.FromMilliseconds(jitter);
+        var jitter = Random.Shared.NextDouble() * Opts.ReconnectJitter.TotalMilliseconds;
+        var waitTime = Opts.ReconnectWait + TimeSpan.FromMilliseconds(jitter);
         if (waitTime != TimeSpan.Zero)
         {
             _logger.LogTrace("Wait {0}ms to reconnect.", waitTime.TotalMilliseconds);
@@ -528,16 +528,16 @@ private async Task WaitWithJitterAsync()
 
     private async void StartPingTimer(CancellationToken cancellationToken)
     {
-        if (Options.PingInterval == TimeSpan.Zero)
+        if (Opts.PingInterval == TimeSpan.Zero)
             return;
 
-        var periodicTimer = new PeriodicTimer(Options.PingInterval);
+        var periodicTimer = new PeriodicTimer(Opts.PingInterval);
         ResetPongCount();
         try
         {
             while (!cancellationToken.IsCancellationRequested)
             {
-                if (Interlocked.Increment(ref _pongCount) > Options.MaxPingOut)
+                if (Interlocked.Increment(ref _pongCount) > Opts.MaxPingOut)
                 {
                     _logger.LogInformation("Detect MaxPingOut, try to connection abort.");
                     if (_socket != null)
@@ -559,7 +559,7 @@ private async void StartPingTimer(CancellationToken cancellationToken)
     [System.Runtime.CompilerServices.MethodImpl(System.Runtime.CompilerServices.MethodImplOptions.AggressiveInlining)]
     private CancellationTimer GetRequestCommandTimer(CancellationToken cancellationToken)
     {
-        return _cancellationTimerPool.Start(Options.RequestTimeout, cancellationToken);
+        return _cancellationTimerPool.Start(Opts.RequestTimeout, cancellationToken);
     }
 
     [System.Runtime.CompilerServices.MethodImpl(System.Runtime.CompilerServices.MethodImplOptions.AggressiveInlining)]