Merge branch 'main' into kv-initial-impl

sandbox/ConsoleApp/Program.cs

@@ -84,7 +84,7 @@ public Runner(INatsConnection connection)
     [RootCommand]
     public async Task Run()
     {
-        var subscription = await _connection.SubscribeAsync("foo");
+        var subscription = await _connection.SubscribeAsync<string>("foo");
 
         _ = Task.Run(async () =>
         {

sandbox/Example.Core.SubscribeHeaders/Program.cs

@@ -13,11 +13,11 @@
 
 Print($"[SUB] Subscribing to subject '{subject}'...\n");
 
-var sub = await connection.SubscribeAsync(subject);
+var sub = await connection.SubscribeAsync<byte[]>(subject);
 
 await foreach (var msg in sub.Msgs.ReadAllAsync())
 {
-    Print($"[RCV] {msg.Subject}: {Encoding.UTF8.GetString(msg.Data.Span)}\n");
+    Print($"[RCV] {msg.Subject}: {Encoding.UTF8.GetString(msg.Data!)}\n");
     if (msg.Headers != null)
     {
         foreach (var (key, values) in msg.Headers)

sandbox/Example.Core.SubscribeQueueGroup/Program.cs

@@ -1,5 +1,4 @@
 // > nats pub foo.xyz --count=10 "my_message_{{ Count }}"
-using System.Text;
 using Microsoft.Extensions.Logging;
 using NATS.Client.Core;
 
@@ -12,13 +11,12 @@
 await using var connection1 = new NatsConnection(options);
 
 Print($"[1][SUB] Subscribing to subject '{subject}'...\n");
-var sub1 = await connection1.SubscribeAsync(subject, queueGroup: "My-Workers");
+var sub1 = await connection1.SubscribeAsync<string>(subject, queueGroup: "My-Workers");
 var task1 = Task.Run(async () =>
 {
     await foreach (var msg in sub1.Msgs.ReadAllAsync())
     {
-        var data = Encoding.UTF8.GetString(msg.Data.ToArray());
-        Print($"[1][RCV] {msg.Subject}: {data}\n");
+        Print($"[1][RCV] {msg.Subject}: {msg.Data}\n");
     }
 });
 
@@ -28,13 +26,12 @@
 await using var connection2 = new NatsConnection(options);
 
 Print($"[2][SUB] Subscribing to subject '{subject}'...\n");
-var sub2 = await connection2.SubscribeAsync(subject, queueGroup: "My-Workers");
+var sub2 = await connection2.SubscribeAsync<string>(subject, queueGroup: "My-Workers");
 var task2 = Task.Run(async () =>
 {
     await foreach (var msg in sub2.Msgs.ReadAllAsync())
     {
-        var data = Encoding.UTF8.GetString(msg.Data.ToArray());
-        Print($"[2][RCV] {msg.Subject}: {data}\n");
+        Print($"[2][RCV] {msg.Subject}: {msg.Data}\n");
     }
 });
 

sandbox/Example.Core.SubscribeRaw/Program.cs

@@ -11,11 +11,11 @@
 
 Print($"[SUB] Subscribing to subject '{subject}'...\n");
 
-var sub = await connection.SubscribeAsync(subject);
+var sub = await connection.SubscribeAsync<byte[]>(subject);
 
 await foreach (var msg in sub.Msgs.ReadAllAsync())
 {
-    var data = Encoding.UTF8.GetString(msg.Data.ToArray());
+    var data = Encoding.UTF8.GetString(msg.Data!);
     Print($"[RCV] {msg.Subject}: {data}\n");
 }
 

sandbox/Example.JetStream.PullConsumer/RawDataSerializer.cs

@@ -5,6 +5,8 @@ namespace Example.JetStream.PullConsumer;
 
 public class RawDataSerializer : INatsSerializer
 {
+    public INatsSerializer? Next => default;
+
     public int Serialize<T>(ICountableBufferWriter bufferWriter, T? value)
     {
         if (value is RawData data)
@@ -16,13 +18,11 @@ public int Serialize<T>(ICountableBufferWriter bufferWriter, T? value)
         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)
+    public T? Deserialize<T>(in ReadOnlySequence<byte> buffer)
     {
-        if (type != typeof(RawData))
+        if (typeof(T) != typeof(RawData))
             throw new Exception($"Can only work with '{typeof(RawData)}'");
 
-        return new RawData(buffer.ToArray());
+        return (T)(object)new RawData(buffer.ToArray());
     }
 }

sandbox/MinimumWebApp/Program.cs

@@ -1,4 +1,3 @@
-using System.Text;
 using NATS.Client.Core;
 using NATS.Client.Hosting;
 
@@ -11,13 +10,13 @@
 
 app.MapGet("/subscribe", async (INatsConnection command) =>
 {
-    var subscription = await command.SubscribeAsync("foo");
+    var subscription = await command.SubscribeAsync<int>("foo");
 
     _ = Task.Run(async () =>
     {
         await foreach (var msg in subscription.Msgs.ReadAllAsync())
         {
-            Console.WriteLine($"Received {Encoding.UTF8.GetString(msg.Data.ToArray())}");
+            Console.WriteLine($"Received {msg.Data}");
         }
     });
 });

sandbox/NatsBenchmark/Program.cs

@@ -77,7 +77,7 @@ private void RunPubSubBenchmark(string testName, long testCount, long testSize,
             pubConn.ConnectAsync().AsTask().Wait();
             subConn.ConnectAsync().AsTask().Wait();
 
-            var d = subConn.SubscribeAsync(_subject).AsTask().Result.Register(_ =>
+            var d = subConn.SubscribeAsync<byte[]>(_subject).AsTask().Result.Register(_ =>
             {
                 Interlocked.Increment(ref subCount);
 
@@ -152,7 +152,7 @@ private void RunPubSubBenchmarkBatch(string testName, long testCount, long testS
             pubConn.ConnectAsync().AsTask().Wait();
             subConn.ConnectAsync().AsTask().Wait();
 
-            var d = subConn.SubscribeAsync(_subject).AsTask().Result.Register(_ =>
+            var d = subConn.SubscribeAsync<byte[]>(_subject).AsTask().Result.Register(_ =>
             {
                 Interlocked.Increment(ref subCount);
 
@@ -236,7 +236,7 @@ private void ProfilingRunPubSubBenchmarkAsync(string testName, long testCount, l
             pubConn.ConnectAsync().AsTask().Wait();
             subConn.ConnectAsync().AsTask().Wait();
 
-            var d = subConn.SubscribeAsync(_subject).AsTask().Result.Register(_ =>
+            var d = subConn.SubscribeAsync<byte[]>(_subject).AsTask().Result.Register(_ =>
             {
                 Interlocked.Increment(ref subCount);
 
@@ -316,7 +316,7 @@ private void RunPubSubBenchmarkBatchRaw(string testName, long testCount, long te
             pubConn.ConnectAsync().AsTask().Wait();
             subConn.ConnectAsync().AsTask().Wait();
 
-            var d = subConn.SubscribeAsync(_subject).AsTask().Result.Register(_ =>
+            var d = subConn.SubscribeAsync<byte[]>(_subject).AsTask().Result.Register(_ =>
             {
                 Interlocked.Increment(ref subCount);
 
@@ -421,7 +421,7 @@ private void RunPubSubBenchmarkPubSub2(string testName, long testCount, long tes
             pubConn2.ConnectAsync().AsTask().Wait();
             subConn2.ConnectAsync().AsTask().Wait();
 
-            var d = subConn.SubscribeAsync(_subject).AsTask().Result.Register(_ =>
+            var d = subConn.SubscribeAsync<byte[]>(_subject).AsTask().Result.Register(_ =>
             {
                 Interlocked.Increment(ref subCount);
 
@@ -435,7 +435,7 @@ private void RunPubSubBenchmarkPubSub2(string testName, long testCount, long tes
                     }
                 }
             });
-            var d2 = subConn2.SubscribeAsync(_subject).AsTask().Result.Register(_ =>
+            var d2 = subConn2.SubscribeAsync<byte[]>(_subject).AsTask().Result.Register(_ =>
             {
                 Interlocked.Increment(ref subCount2);
 
@@ -824,18 +824,4 @@ internal static class NatsMsgTestUtils
         });
         return sub;
     }
-
-    internal static INatsSub? Register(this INatsSub? sub, Action<NatsMsg> action)
-    {
-        if (sub == null)
-            return null;
-        Task.Run(async () =>
-        {
-            await foreach (var natsMsg in sub.Msgs.ReadAllAsync())
-            {
-                action(natsMsg);
-            }
-        });
-        return sub;
-    }
 }

src/NATS.Client.Core/INatsConnection.cs

@@ -12,38 +12,32 @@ public interface INatsConnection
     ValueTask<TimeSpan> PingAsync(CancellationToken cancellationToken = default);
 
     /// <summary>
-    /// Publishes the message payload to the given subject name, optionally supplying a reply subject.
+    /// Publishes a serializable message payload to the given subject name, optionally supplying a reply subject.
     /// </summary>
     /// <param name="subject">The destination subject to publish to.</param>
-    /// <param name="payload">The message payload data.</param>
+    /// <param name="data">Serializable data object.</param>
     /// <param name="headers">Optional message headers.</param>
     /// <param name="replyTo">Optional reply-to subject.</param>
     /// <param name="opts">A <see cref="NatsPubOpts"/> for publishing options.</param>
     /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the command.</param>
+    /// <typeparam name="T">Specifies the type of data that may be sent to the NATS Server.</typeparam>
     /// <returns>A <see cref="ValueTask"/> that represents the asynchronous send operation.</returns>
-    ValueTask PublishAsync(string subject, ReadOnlySequence<byte> payload = default, NatsHeaders? headers = default, string? replyTo = default, NatsPubOpts? opts = default, CancellationToken cancellationToken = default);
-
-    /// <summary>
-    /// Publishes the message payload to the given subject name, optionally supplying a reply subject.
-    /// </summary>
-    /// <param name="msg">A <see cref="NatsMsg"/> representing message details.</param>
-    /// <param name="opts">A <see cref="NatsPubOpts"/> for publishing options.</param>
-    /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the command.</param>
-    /// <returns>A <see cref="ValueTask"/> that represents the asynchronous send operation.</returns>
-    ValueTask PublishAsync(in NatsMsg msg, NatsPubOpts? opts = default, CancellationToken cancellationToken = default);
+    ValueTask PublishAsync<T>(string subject, T data, NatsHeaders? headers = default, string? replyTo = default, NatsPubOpts? opts = default, CancellationToken cancellationToken = default);
 
     /// <summary>
-    /// Publishes a serializable message payload to the given subject name, optionally supplying a reply subject.
+    /// Publishes an empty message payload to the given subject name, optionally supplying a reply subject.
     /// </summary>
     /// <param name="subject">The destination subject to publish to.</param>
-    /// <param name="data">Serializable data object.</param>
     /// <param name="headers">Optional message headers.</param>
     /// <param name="replyTo">Optional reply-to subject.</param>
     /// <param name="opts">A <see cref="NatsPubOpts"/> for publishing options.</param>
     /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the command.</param>
-    /// <typeparam name="T">Specifies the type of data that may be sent to the NATS Server.</typeparam>
     /// <returns>A <see cref="ValueTask"/> that represents the asynchronous send operation.</returns>
-    ValueTask PublishAsync<T>(string subject, T data, NatsHeaders? headers = default, string? replyTo = default, NatsPubOpts? opts = default, CancellationToken cancellationToken = default);
+    /// <remarks>
+    /// Publishing a sentinel usually means a signal to the given subject which could be used to trigger an action
+    /// or indicate an event for example and of messages.
+    /// </remarks>
+    ValueTask PublishAsync(string subject, NatsHeaders? headers = default, string? replyTo = default, NatsPubOpts? opts = default, CancellationToken cancellationToken = default);
 
     /// <summary>
     /// Publishes a serializable message payload to the given subject name, optionally supplying a reply subject.
@@ -55,21 +49,6 @@ public interface INatsConnection
     /// <returns>A <see cref="ValueTask"/> that represents the asynchronous send operation.</returns>
     ValueTask PublishAsync<T>(in NatsMsg<T> msg, NatsPubOpts? opts = default, CancellationToken cancellationToken = default);
 
-    /// <summary>
-    /// Initiates a subscription to a subject, optionally joining a distributed queue group.
-    /// </summary>
-    /// <param name="subject">The subject name to subscribe to.</param>
-    /// <param name="queueGroup">If specified, the subscriber will join this queue group.</param>
-    /// <param name="opts">A <see cref="NatsSubOpts"/> for subscription options.</param>
-    /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the command.</param>
-    /// <returns>A <see cref="ValueTask{TResult}"/> that represents the asynchronous send operation.</returns>
-    /// <remarks>
-    /// Subscribers with the same queue group name, become a queue group,
-    /// and only one randomly chosen subscriber of the queue group will
-    /// consume a message each time a message is received by the queue group.
-    /// </remarks>
-    ValueTask<INatsSub> SubscribeAsync(string subject, string? queueGroup = default, NatsSubOpts? opts = default, CancellationToken cancellationToken = default);
-
     /// <summary>
     /// Initiates a subscription to a subject, optionally joining a distributed queue group.
     /// </summary>
@@ -106,7 +85,7 @@ public interface INatsConnection
     /// <returns>Returns the <see cref="NatsMsg{T}"/> received from the responder as reply.</returns>
     /// <exception cref="OperationCanceledException">Raised when cancellation token is used</exception>
     /// <remarks>
-    /// Response can be (null) or one <see cref="NatsMsg"/>.
+    /// Response can be (null) or one <see cref="NatsMsg{T}"/>.
     /// Reply option's max messages will be set to 1.
     /// if reply option's timeout is not defined then it will be set to NatsOpts.RequestTimeout.
     /// </remarks>
@@ -118,30 +97,6 @@ public interface INatsConnection
         NatsSubOpts? replyOpts = default,
         CancellationToken cancellationToken = default);
 
-    /// <summary>
-    /// Request and receive a single reply from a responder.
-    /// </summary>
-    /// <param name="subject">Subject of the responder</param>
-    /// <param name="payload">Payload to send to responder</param>
-    /// <param name="headers">Optional message headers</param>
-    /// <param name="requestOpts">Request publish options</param>
-    /// <param name="replyOpts">Reply handler subscription options</param>
-    /// <param name="cancellationToken">Cancel this request</param>
-    /// <returns>Returns the <see cref="NatsMsg"/> received from the responder as reply.</returns>
-    /// <exception cref="OperationCanceledException">Raised when cancellation token is used</exception>
-    /// <remarks>
-    /// Response can be (null) or one <see cref="NatsMsg"/>.
-    /// Reply option's max messages will be set to 1 (one).
-    /// if reply option's timeout is not defined then it will be set to NatsOpts.RequestTimeout.
-    /// </remarks>
-    ValueTask<NatsMsg?> RequestAsync(
-        string subject,
-        ReadOnlySequence<byte> payload = default,
-        NatsHeaders? headers = default,
-        NatsPubOpts? requestOpts = default,
-        NatsSubOpts? replyOpts = default,
-        CancellationToken cancellationToken = default);
-
     /// <summary>
     /// Request and receive zero or more replies from a responder.
     /// </summary>
@@ -153,7 +108,7 @@ public interface INatsConnection
     /// <param name="cancellationToken">Cancel this request</param>
     /// <typeparam name="TRequest">Request type</typeparam>
     /// <typeparam name="TReply">Reply type</typeparam>
-    /// <returns>An asynchronous enumerable of <see cref="NatsMsg"/> objects</returns>
+    /// <returns>An asynchronous enumerable of <see cref="NatsMsg{T}"/> objects</returns>
     /// <exception cref="OperationCanceledException">Raised when cancellation token is used</exception>
     /// <remarks>
     /// if reply option's timeout is not defined then it will be set to NatsOpts.RequestTimeout.
@@ -165,26 +120,4 @@ public interface INatsConnection
         NatsPubOpts? requestOpts = default,
         NatsSubOpts? replyOpts = default,
         CancellationToken cancellationToken = default);
-
-    /// <summary>
-    /// Request and receive zero or more replies from a responder.
-    /// </summary>
-    /// <param name="subject">Subject of the responder</param>
-    /// <param name="payload">Payload to send to responder</param>
-    /// <param name="headers">Optional message headers</param>
-    /// <param name="requestOpts">Request publish options</param>
-    /// <param name="replyOpts">Reply handler subscription options</param>
-    /// <param name="cancellationToken">Cancel this request</param>
-    /// <returns>An asynchronous enumerable of <see cref="NatsMsg"/> objects</returns>
-    /// <exception cref="OperationCanceledException">Raised when cancellation token is used</exception>
-    /// <remarks>
-    /// if reply option's timeout is not defined then it will be set to NatsOpts.RequestTimeout.
-    /// </remarks>
-    IAsyncEnumerable<NatsMsg> RequestManyAsync(
-        string subject,
-        ReadOnlySequence<byte> payload = default,
-        NatsHeaders? headers = default,
-        NatsPubOpts? requestOpts = default,
-        NatsSubOpts? replyOpts = default,
-        CancellationToken cancellationToken = default);
 }