Added support direct get batch API

src/NATS.Client.Core/NatsHeaderParser.cs

@@ -119,6 +119,11 @@ private bool TryParseHeaderLine(ReadOnlySpan<byte> headerLine, NatsHeaders heade
                 headers.Message = NatsHeaders.Messages.MessageSizeExceedsMaxBytes;
                 headers.MessageText = NatsHeaders.MessageMessageSizeExceedsMaxBytesStr;
             }
+            else if (headerLine.SequenceEqual(NatsHeaders.MessageEobCode))
+            {
+                headers.Message = NatsHeaders.Messages.EobCode;
+                headers.MessageText = NatsHeaders.MessageEobCodeStr;
+            }
             else
             {
                 headers.Message = NatsHeaders.Messages.Text;

src/NATS.Client.Core/NatsHeaders.cs

@@ -25,6 +25,7 @@ public enum Messages
         NoMessages,
         RequestTimeout,
         MessageSizeExceedsMaxBytes,
+        EobCode,
     }
 
     // Uses C# compiler's optimization for static byte[] data
@@ -56,6 +57,10 @@ public enum Messages
     internal static ReadOnlySpan<byte> MessageMessageSizeExceedsMaxBytes => new byte[] { 77, 101, 115, 115, 97, 103, 101, 32, 83, 105, 122, 101, 32, 69, 120, 99, 101, 101, 100, 115, 32, 77, 97, 120, 66, 121, 116, 101, 115 };
     internal static readonly string MessageMessageSizeExceedsMaxBytesStr = "Message Size Exceeds MaxBytes";
 
+    // EOB
+    internal static ReadOnlySpan<byte> MessageEobCode => new byte[] { 69, 79, 66 };
+    internal static readonly string MessageEobCodeStr = "EOB";
+
     private static readonly string[] EmptyKeys = Array.Empty<string>();
     private static readonly StringValues[] EmptyValues = Array.Empty<StringValues>();
 

src/NATS.Client.JetStream/INatsJSStream.cs

@@ -112,5 +112,43 @@ ValueTask UpdateAsync(
 
     ValueTask<NatsMsg<T>> GetDirectAsync<T>(StreamMsgGetRequest request, INatsDeserialize<T>? serializer = default, CancellationToken cancellationToken = default);
 
+    /// <summary>
+    /// Request a direct batch message
+    /// </summary>
+    /// <param name="request">Batch message request.</param>
+    /// <param name="serializer">Serializer to use for the message type.</param>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
+    /// <typeparam name="T">Message type to deserialize.</typeparam>
+    /// <exception cref="InvalidOperationException">There was an issue, stream must have allow direct set.</exception>
+    IAsyncEnumerable<NatsMsg<T>> GetBatchDirectAsync<T>(StreamMsgBatchGetRequest request, INatsDeserialize<T>? serializer = default, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Request a direct batch message
+    /// </summary>
+    /// <param name="multiLastBySubjects">Return last messages matching the subjects</param>
+    /// <param name="batch">The maximum amount of messages to be returned</param>
+    /// <param name="serializer">Serializer to use for the message type.</param>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
+    /// <typeparam name="T">Message type to deserialize.</typeparam>
+    /// <remarks>
+    /// Get up to batch number of messages for subject
+    /// </remarks>
+    /// <exception cref="InvalidOperationException">There was an issue, stream must have allow direct set.</exception>
+    IAsyncEnumerable<NatsMsg<T>> GetBatchDirectAsync<T>(string[] multiLastBySubjects, ulong batch, INatsDeserialize<T>? serializer = default, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Request a direct batch message
+    /// </summary>
+    /// <param name="nextBySubject">The subject used filter messages that should be returned</param>
+    /// <param name="batch">The maximum amount of messages to be returned</param>
+    /// <param name="serializer">Serializer to use for the message type.</param>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
+    /// <typeparam name="T">Message type to deserialize.</typeparam>
+    /// <remarks>
+    /// Get the last message for each subject in the list up to the batch size
+    /// </remarks>
+    /// <exception cref="InvalidOperationException">There was an issue, stream must have allow direct set.</exception>
+    IAsyncEnumerable<NatsMsg<T>> GetBatchDirectAsync<T>(string nextBySubject, ulong batch, INatsDeserialize<T>? serializer = default, CancellationToken cancellationToken = default);
+
     ValueTask<StreamMsgGetResponse> GetAsync(StreamMsgGetRequest request, CancellationToken cancellationToken = default);
 }

src/NATS.Client.JetStream/Models/StreamMsgBatchGetRequest.cs

@@ -0,0 +1,76 @@
+using System.Text.Json.Serialization;
+
+namespace NATS.Client.JetStream.Models;
+
+/// <summary>
+/// A request to the JetStream $JS.API.STREAM.MSG.GET API
+/// </summary>
+public record StreamMsgBatchGetRequest
+{
+    /// <summary>
+    /// The maximum amount of messages to be returned for this request
+    /// </summary>
+    [JsonPropertyName("batch")]
+    [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingDefault)]
+    [System.ComponentModel.DataAnnotations.Range(ulong.MinValue, ulong.MaxValue)]
+    public ulong Batch { get; set; }
+
+    /// <summary>
+    /// The maximum amount of returned bytes for this request.
+    /// </summary>
+    [JsonPropertyName("max_bytes")]
+    [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingDefault)]
+    [System.ComponentModel.DataAnnotations.Range(ulong.MinValue, long.MaxValue)]
+    public ulong MaxBytes { get; set; }
+
+    /// <summary>
+    /// The minimum sequence for returned message
+    /// </summary>
+    [JsonPropertyName("seq")]
+    [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingDefault)]
+    [System.ComponentModel.DataAnnotations.Range(ulong.MinValue, ulong.MaxValue)]
+    public ulong Seq { get; set; }
+
+    /// <summary>
+    /// The minimum start time for returned message
+    /// </summary>
+    [JsonPropertyName("start_time")]
+    [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingDefault)]
+    public DateTimeOffset StartTime { get; set; }
+
+    /// <summary>
+    /// The subject used filter messages that should be returned
+    /// </summary>
+    [JsonPropertyName("next_by_subj")]
+    [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)]
+#if NET6_0
+    public string? NextBySubject { get; set; } = default!;
+#else
+#pragma warning disable SA1206
+    public string? NextBySubject { get; set; }
+
+#pragma warning restore SA1206
+#endif
+
+    /// <summary>
+    /// Return last messages mathing the subjects
+    /// </summary>
+    [JsonPropertyName("multi_last")]
+    [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)]
+    public string[]? MultiLastBySubjects { get; set; }
+
+    /// <summary>
+    /// Return message after sequence
+    /// </summary>
+    [JsonPropertyName("up_to_seq")]
+    [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingDefault)]
+    [System.ComponentModel.DataAnnotations.Range(ulong.MinValue, ulong.MaxValue)]
+    public ulong UpToSequence { get; set; }
+
+    /// <summary>
+    /// Return message after time
+    /// </summary>
+    [JsonPropertyName("up_to_time")]
+    [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingDefault)]
+    public DateTimeOffset UpToTime { get; set; }
+}

src/NATS.Client.JetStream/NatsJSJsonSerializer.cs

@@ -78,6 +78,7 @@ public static class NatsJSJsonSerializer<T>
 [JsonSerializable(typeof(StreamMsgDeleteResponse))]
 [JsonSerializable(typeof(StreamMsgGetRequest))]
 [JsonSerializable(typeof(StreamMsgGetResponse))]
+[JsonSerializable(typeof(StreamMsgBatchGetRequest))]
 [JsonSerializable(typeof(StreamNamesRequest))]
 [JsonSerializable(typeof(StreamNamesResponse))]
 [JsonSerializable(typeof(StreamPurgeRequest))]

src/NATS.Client.JetStream/NatsJSStream.cs

@@ -1,5 +1,5 @@
+using System.Runtime.CompilerServices;
 using NATS.Client.Core;
-using NATS.Client.JetStream.Internal;
 using NATS.Client.JetStream.Models;
 
 namespace NATS.Client.JetStream;
@@ -181,15 +181,61 @@ public ValueTask<NatsMsg<T>> GetDirectAsync<T>(StreamMsgGetRequest request, INat
             cancellationToken: cancellationToken);
     }
 
+    public IAsyncEnumerable<NatsMsg<T>> GetBatchDirectAsync<T>(StreamMsgBatchGetRequest request, INatsDeserialize<T>? serializer = default, CancellationToken cancellationToken = default)
+        => GetBatchDirectInternalAsync<T>(request, serializer, cancellationToken);
+
+    public IAsyncEnumerable<NatsMsg<T>> GetBatchDirectAsync<T>(string[] multiLastBySubjects, ulong batch, INatsDeserialize<T>? serializer = default, CancellationToken cancellationToken = default)
+    {
+        var request = new StreamMsgBatchGetRequest { MultiLastBySubjects = multiLastBySubjects, Batch = batch };
+        return GetBatchDirectInternalAsync<T>(request, serializer, cancellationToken);
+    }
+
+    public IAsyncEnumerable<NatsMsg<T>> GetBatchDirectAsync<T>(string nextBySubject, ulong batch, INatsDeserialize<T>? serializer = default, CancellationToken cancellationToken = default)
+    {
+        var request = new StreamMsgBatchGetRequest { NextBySubject = nextBySubject, Batch = batch };
+        return GetBatchDirectInternalAsync<T>(request, serializer, cancellationToken);
+    }
+
     public ValueTask<StreamMsgGetResponse> GetAsync(StreamMsgGetRequest request, CancellationToken cancellationToken = default) =>
         _context.JSRequestResponseAsync<StreamMsgGetRequest, StreamMsgGetResponse>(
             subject: $"{_context.Opts.Prefix}.STREAM.MSG.GET.{_name}",
             request: request,
             cancellationToken);
 
+    private async IAsyncEnumerable<NatsMsg<T>> GetBatchDirectInternalAsync<T>(StreamMsgBatchGetRequest request, INatsDeserialize<T>? serializer = default, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        ValidateStream();
+
+        var requestManyAsync = _context.Connection.RequestManyAsync(
+            subject: $"{_context.Opts.Prefix}.DIRECT.GET.{_name}",
+            data: request,
+            requestSerializer: NatsJSJsonSerializer<StreamMsgBatchGetRequest>.Default,
+            replySerializer: serializer,
+            replyOpts: new NatsSubOpts { StopOnEmptyMsg = true, ThrowIfNoResponders = true },
+            cancellationToken: cancellationToken);
+
+        await foreach (var msg in requestManyAsync.ConfigureAwait(false))
+        {
+            if (msg.Error is { } error)
+            {
+                throw error;
+            }
+
+            yield return msg;
+        }
+    }
+
     private void ThrowIfDeleted()
     {
         if (_deleted)
             throw new NatsJSException($"Stream '{_name}' is deleted");
     }
+
+    private void ValidateStream()
+    {
+        if (!Info.Config.AllowDirect)
+        {
+            throw new InvalidOperationException("StreamMsgBatchGetRequest is not permitted when AllowDirect on stream disable");
+        }
+    }
 }

tests/NATS.Client.CoreUnit.Tests/NatsHeaderTest.cs

@@ -96,6 +96,7 @@ public void ParserTests()
     [InlineData("Request Timeout", NatsHeaders.Messages.RequestTimeout)]
     [InlineData("Message Size Exceeds MaxBytes", NatsHeaders.Messages.MessageSizeExceedsMaxBytes)]
     [InlineData("test message", NatsHeaders.Messages.Text)]
+    [InlineData("EOB", NatsHeaders.Messages.EobCode)]
     public void ParserMessageEnumTests(string message, NatsHeaders.Messages result)
     {
         var parser = new NatsHeaderParser(Encoding.UTF8);

tests/NATS.Client.JetStream.Tests/DirectGetTest.cs

@@ -0,0 +1,78 @@
+using NATS.Client.Core.Tests;
+using NATS.Client.JetStream.Models;
+
+namespace NATS.Client.JetStream.Tests;
+
+public class DirectGetTest
+{
+    [SkipIfNatsServer(versionEarlierThan: "2.11")]
+    public async Task Direct_get_when_stream_disable()
+    {
+        await using var server = NatsServer.StartJS();
+        await using var nats = server.CreateClientConnection();
+        var js = new NatsJSContext(nats);
+        var cts = new CancellationTokenSource();
+        var cancellationToken = cts.Token;
+        var streamConfig = new StreamConfig("stream_disable", new[] { "stream_disable.x" });
+
+        var stream = await js.CreateStreamAsync(streamConfig, cancellationToken);
+
+        async Task GetBatchAction()
+        {
+            var streamMsgBatchGetRequest = new StreamMsgBatchGetRequest { NextBySubject = "stream_disable.x" };
+            await foreach (var unused in stream.GetBatchDirectAsync<string>(streamMsgBatchGetRequest, cancellationToken: cancellationToken))
+            {
+            }
+        }
+
+        await Assert.ThrowsAsync<InvalidOperationException>(GetBatchAction);
+    }
+
+    [SkipIfNatsServer(versionEarlierThan: "2.11")]
+    public async Task Direct_get_when_stream_enable()
+    {
+        var testDataList = new List<TestData?>();
+        await using var server = NatsServer.StartJS();
+        await using var nats = server.CreateClientConnection();
+        var js = new NatsJSContext(nats);
+        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(100));
+        var cancellationToken = cts.Token;
+        var streamConfig = new StreamConfig("stream_enable", new[] { "stream_enable.x" }) { AllowDirect = true };
+
+        var stream = await js.CreateStreamAsync(streamConfig, cancellationToken);
+
+        await js.PublishAsync("stream_enable.x", new TestData { Test = 1 }, TestDataJsonSerializer<TestData>.Default, cancellationToken: cancellationToken);
+
+        var streamMsgBatchGetRequest = new StreamMsgBatchGetRequest { NextBySubject = "stream_enable.x", Batch = 3 };
+        await foreach (var msg in stream.GetBatchDirectAsync(streamMsgBatchGetRequest, TestDataJsonSerializer<TestData>.Default, cancellationToken: cancellationToken))
+        {
+            testDataList.Add(msg.Data);
+        }
+
+        Assert.Single(testDataList);
+    }
+
+    [SkipIfNatsServer(versionEarlierThan: "2.11")]
+    public async Task Direct_get_by_multi_last()
+    {
+        var testDataList = new List<TestData?>();
+        await using var server = NatsServer.StartJS();
+        await using var nats = server.CreateClientConnection();
+        var js = new NatsJSContext(nats);
+        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(100));
+        var cancellationToken = cts.Token;
+        var streamConfig = new StreamConfig("multiLast", ["multiLast.*"]) { AllowDirect = true };
+
+        var stream = await js.CreateStreamAsync(streamConfig, cancellationToken);
+
+        await js.PublishAsync("multiLast.x", new TestData { Test = 1 }, TestDataJsonSerializer<TestData>.Default, cancellationToken: cancellationToken);
+        await js.PublishAsync("multiLast.y", new TestData { Test = 2 }, TestDataJsonSerializer<TestData>.Default, cancellationToken: cancellationToken);
+
+        await foreach (var msg in stream.GetBatchDirectAsync(["multiLast.x", "multiLast.y"], 4, TestDataJsonSerializer<TestData>.Default, cancellationToken: cancellationToken))
+        {
+            testDataList.Add(msg.Data);
+        }
+
+        Assert.Equal(2, testDataList.Count);
+    }
+}