Low-level APIs and last clean-up before first release (#65)

* Implemented Pub/Sub/INatsSub low-level APIs

* INatsConnection rename and namespace changes

Also removed flush, observable and sharding.

* NatsMsg as struct

* Made low-level api internal

sandbox/BlazorWasm/Server/NatsServices/WeatherForecastService.cs

@@ -11,18 +11,18 @@ public class WeatherForecastService : IHostedService, IAsyncDisposable
     };
 
     private readonly ILogger<WeatherForecastService> _logger;
-    private readonly INatsCommand _natsCommand;
+    private readonly INatsConnection _natsConnection;
     private NatsReplyHandle? _replyHandle;
 
-    public WeatherForecastService(ILogger<WeatherForecastService> logger, INatsCommand natsCommand)
+    public WeatherForecastService(ILogger<WeatherForecastService> logger, INatsConnection natsConnection)
     {
         _logger = logger;
-        _natsCommand = natsCommand;
+        _natsConnection = natsConnection;
     }
 
     public async Task StartAsync(CancellationToken cancellationToken)
     {
-        _replyHandle = await _natsCommand.ReplyAsync<object, WeatherForecast[]>("weather", req =>
+        _replyHandle = await _natsConnection.ReplyAsync<object, WeatherForecast[]>("weather", req =>
         {
             return Enumerable.Range(1, 5).Select(index => new WeatherForecast
             {

sandbox/ConsoleApp/Program.cs

@@ -64,17 +64,17 @@ public record Person(int Age, string Name);
 
 public class Runner : ConsoleAppBase
 {
-    private readonly INatsCommand _command;
+    private readonly INatsConnection _connection;
 
-    public Runner(INatsCommand command)
+    public Runner(INatsConnection connection)
     {
-        _command = command;
+        _connection = connection;
     }
 
     [RootCommand]
     public async Task Run()
     {
-        var subscription = await _command.SubscribeAsync("foo");
+        var subscription = await _connection.SubscribeAsync("foo");
 
         _ = Task.Run(async () =>
         {
@@ -84,8 +84,8 @@ public async Task Run()
             }
         });
 
-        await _command.PingAsync();
-        await _command.PublishAsync("foo");
+        await _connection.PingAsync();
+        await _connection.PublishAsync("foo");
     }
 }
 

sandbox/MinimumWebApp/Program.cs

@@ -9,7 +9,7 @@
 
 var app = builder.Build();
 
-app.MapGet("/subscribe", async (INatsCommand command) =>
+app.MapGet("/subscribe", async (INatsConnection command) =>
 {
     var subscription = await command.SubscribeAsync("foo");
 
@@ -22,6 +22,6 @@
     });
 });
 
-app.MapGet("/publish", async (INatsCommand command) => await command.PublishAsync("foo", 99));
+app.MapGet("/publish", async (INatsConnection command) => await command.PublishAsync("foo", 99));
 
 app.Run();

src/NATS.Client.Core/Internal/HeaderParser.cs → src/NATS.Client.Core/HeaderParser.cs

@@ -4,10 +4,11 @@
 using System.Diagnostics;
 using System.Text;
 using Microsoft.Extensions.Primitives;
+using NATS.Client.Core.Internal;
 
-namespace NATS.Client.Core.Internal;
+namespace NATS.Client.Core;
 
-internal class HeaderParser
+public class HeaderParser
 {
     private const byte ByteCR = (byte)'\r';
     private const byte ByteLF = (byte)'\n';

src/NATS.Client.Core/INatsCommand.cs → src/NATS.Client.Core/INatsConnection.cs

@@ -2,10 +2,8 @@
 
 namespace NATS.Client.Core;
 
-public interface INatsCommand
+public interface INatsConnection
 {
-    ValueTask FlushAsync(CancellationToken cancellationToken = default);
-
     /// <summary>
     /// Send PING command and await PONG. Return value is similar as Round Trip Time (RTT).
     /// </summary>
@@ -69,6 +67,4 @@ public interface INatsCommand
     /// <typeparam name="T">Specifies the type of data that may be received from the NATS Server.</typeparam>
     /// <returns>A <see cref="ValueTask{TResult}"/> that represents the asynchronous send operation.</returns>
     ValueTask<NatsSub<T>> SubscribeAsync<T>(string subject, in NatsSubOpts? opts = default, CancellationToken cancellationToken = default);
-
-    IObservable<T> AsObservable<T>(string subject);
 }

src/NATS.Client.Core/INatsSub.cs

@@ -0,0 +1,49 @@
+using System.Buffers;
+using System.Collections.Concurrent;
+
+namespace NATS.Client.Core;
+
+internal interface INatsSub : IAsyncDisposable
+{
+    string Subject { get; }
+
+    string? QueueGroup { get; }
+
+    int Sid { get; }
+
+    ValueTask ReceiveAsync(string subject, string? replyTo, in ReadOnlySequence<byte>? headersBuffer, in ReadOnlySequence<byte> payloadBuffer);
+}
+
+internal interface INatsSubBuilder<out T>
+    where T : INatsSub
+{
+    T Build(string subject, string? queueGroup, NatsConnection connection, SubscriptionManager manager);
+}
+
+internal class NatsSubBuilder : INatsSubBuilder<NatsSub>
+{
+    public static readonly NatsSubBuilder Default = new();
+
+    public NatsSub Build(string subject, string? queueGroup, NatsConnection connection, SubscriptionManager manager)
+    {
+        var sid = manager.GetNextSid();
+        return new NatsSub(connection, manager, subject, queueGroup, sid);
+    }
+}
+
+internal class NatsSubModelBuilder<T> : INatsSubBuilder<NatsSub<T>>
+{
+    private static readonly ConcurrentDictionary<INatsSerializer, NatsSubModelBuilder<T>> Cache = new();
+    private readonly INatsSerializer _serializer;
+
+    public NatsSubModelBuilder(INatsSerializer serializer) => _serializer = serializer;
+
+    public static NatsSubModelBuilder<T> For(INatsSerializer serializer) =>
+        Cache.GetOrAdd(serializer, static s => new NatsSubModelBuilder<T>(s));
+
+    public NatsSub<T> Build(string subject, string? queueGroup, NatsConnection connection, SubscriptionManager manager)
+    {
+        var sid = manager.GetNextSid();
+        return new NatsSub<T>(connection, manager, subject, queueGroup, sid, _serializer);
+    }
+}

src/NATS.Client.Core/NatsPipeliningWriteProtocolProcessor.cs → src/NATS.Client.Core/Internal/NatsPipeliningWriteProtocolProcessor.cs

@@ -2,9 +2,8 @@
 using System.Threading.Channels;
 using Microsoft.Extensions.Logging;
 using NATS.Client.Core.Commands;
-using NATS.Client.Core.Internal;
 
-namespace NATS.Client.Core;
+namespace NATS.Client.Core.Internal;
 
 internal sealed class NatsPipeliningWriteProtocolProcessor : IAsyncDisposable
 {

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

@@ -8,9 +8,8 @@
 using System.Text.Json;
 using Microsoft.Extensions.Logging;
 using NATS.Client.Core.Commands;
-using NATS.Client.Core.Internal;
 
-namespace NATS.Client.Core;
+namespace NATS.Client.Core.Internal;
 
 internal sealed class NatsReadProtocolProcessor : IAsyncDisposable
 {

src/NATS.Client.Core/NatsUri.cs → src/NATS.Client.Core/Internal/NatsUri.cs

@@ -1,4 +1,4 @@
-namespace NATS.Client.Core;
+namespace NATS.Client.Core.Internal;
 
 internal sealed class NatsUri : IEquatable<NatsUri>
 {

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

@@ -0,0 +1,76 @@
+using System.Buffers;
+using NATS.Client.Core.Commands;
+
+namespace NATS.Client.Core;
+
+public partial class NatsConnection
+{
+    internal ValueTask PubAsync(string subject, string? replyTo = default, ReadOnlySequence<byte> payload = default, NatsHeaders? headers = default, CancellationToken cancellationToken = default)
+    {
+        headers?.SetReadOnly();
+
+        if (ConnectionState == NatsConnectionState.Open)
+        {
+            var command = AsyncPublishBytesCommand.Create(_pool, GetCommandTimer(cancellationToken), subject, replyTo, headers, payload);
+            if (TryEnqueueCommand(command))
+            {
+                return command.AsValueTask();
+            }
+            else
+            {
+                return EnqueueAndAwaitCommandAsync(command);
+            }
+        }
+        else
+        {
+            return WithConnectAsync(subject, replyTo, headers, payload, cancellationToken, static (self, s, r, h, p, token) =>
+            {
+                var command = AsyncPublishBytesCommand.Create(self._pool, self.GetCommandTimer(token), s, r, h, p);
+                return self.EnqueueAndAwaitCommandAsync(command);
+            });
+        }
+    }
+
+    internal ValueTask PubModelAsync<T>(string subject, T data, INatsSerializer serializer, string? replyTo = default, NatsHeaders? headers = default, CancellationToken cancellationToken = default)
+    {
+        headers?.SetReadOnly();
+
+        if (ConnectionState == NatsConnectionState.Open)
+        {
+            var command = AsyncPublishCommand<T>.Create(_pool, GetCommandTimer(cancellationToken), subject, replyTo, headers, data, serializer);
+            if (TryEnqueueCommand(command))
+            {
+                return command.AsValueTask();
+            }
+            else
+            {
+                return EnqueueAndAwaitCommandAsync(command);
+            }
+        }
+        else
+        {
+            return WithConnectAsync(subject, replyTo, headers, data, serializer, cancellationToken, static (self, s, r, h, v, ser, token) =>
+            {
+                var command = AsyncPublishCommand<T>.Create(self._pool, self.GetCommandTimer(token), s, r, h, v, ser);
+                return self.EnqueueAndAwaitCommandAsync(command);
+            });
+        }
+    }
+
+    internal ValueTask<T> SubAsync<T>(string subject, string? queueGroup, INatsSubBuilder<T> builder, CancellationToken cancellationToken = default)
+        where T : INatsSub
+    {
+        var sub = builder.Build(subject, queueGroup, this, _subscriptionManager);
+        if (ConnectionState == NatsConnectionState.Open)
+        {
+            return _subscriptionManager.SubscribeAsync(subject, queueGroup, sub, cancellationToken);
+        }
+        else
+        {
+            return WithConnectAsync(subject, queueGroup, sub, cancellationToken, static (self, s, qg, sb, token) =>
+            {
+                return self._subscriptionManager.SubscribeAsync(s, qg, sb, token);
+            });
+        }
+    }
+}