Rework TermWithReason

src/NATS.Client.JetStream/NatsJSExtensions.cs

@@ -38,4 +38,10 @@ public static bool IsSuccess(this PubAckResponse ack)
 #endif
         return ack.Error == null && !ack.Duplicate;
     }
+
+    public static ValueTask NakAsync<T>(this INatsJSMsg<T> msg, TimeSpan delay, CancellationToken cancellationToken = default)
+        => msg.NakAsync(new AckOpts { NakDelay = delay }, cancellationToken);
+
+    public static ValueTask AckTerminateAsync<T>(this INatsJSMsg<T> msg, string reason, CancellationToken cancellationToken = default)
+        => msg.AckTerminateAsync(new AckOpts { TerminateReason = reason }, cancellationToken);
 }

src/NATS.Client.JetStream/NatsJSMsg.cs

@@ -93,15 +93,10 @@ public interface INatsJSMsg<out T> : INatsMsg
     /// <summary>
     /// Signals that the message will not be processed now and processing can move onto the next message.
     /// </summary>
-    /// <param name="delay">Delay redelivery of the message.</param>
-    /// <param name="opts">Ack options.</param>
+    /// <param name="opts">Ack options including <see cref="AckOpts.NakDelay"/>.</param>
     /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the call.</param>
     /// <returns>A <see cref="ValueTask"/> representing the async call.</returns>
-    /// <remarks>
-    /// Messages rejected using <c>-NAK</c> will be resent by the NATS JetStream server after the configured timeout
-    /// or the delay parameter if it's specified.
-    /// </remarks>
-    ValueTask NakAsync(AckOpts? opts = null, TimeSpan delay = default, CancellationToken cancellationToken = default);
+    ValueTask NakAsync(AckOpts? opts = null, CancellationToken cancellationToken = default);
 
     /// <summary>
     /// Indicates that work is ongoing and the wait period should be extended.
@@ -128,15 +123,6 @@ public interface INatsJSMsg<out T> : INatsMsg
     /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the call.</param>
     /// <returns>A <see cref="ValueTask"/> representing the async call.</returns>
     ValueTask AckTerminateAsync(AckOpts? opts = null, CancellationToken cancellationToken = default);
-
-    /// <summary>
-    /// Instructs the server to stop redelivery of the message without acknowledging it as successfully processed.
-    /// </summary>
-    /// <param name="reason">Optional reason for termination, included in JetStream advisory events. Requires NATS Server 2.10.4+.</param>
-    /// <param name="opts">Ack options.</param>
-    /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the call.</param>
-    /// <returns>A <see cref="ValueTask"/> representing the async call.</returns>
-    ValueTask AckTerminateAsync(string reason, AckOpts? opts = null, CancellationToken cancellationToken = default);
 }
 
 /// <summary>
@@ -194,39 +180,25 @@ public ValueTask ReplyAsync(NatsHeaders? headers = null, string? replyTo = null,
     public ValueTask AckAsync(AckOpts? opts = null, CancellationToken cancellationToken = default) => SendAckAsync(NatsJSConstants.Ack, opts, cancellationToken);
 
     /// <inheritdoc />
-    public ValueTask NakAsync(AckOpts? opts = null, TimeSpan delay = default, CancellationToken cancellationToken = default)
+    public ValueTask NakAsync(AckOpts? opts = null, CancellationToken cancellationToken = default)
     {
+        var delay = opts?.NakDelay ?? TimeSpan.Zero;
+
         if (delay == TimeSpan.Zero)
-        {
             return SendAckAsync(NatsJSConstants.Nak, opts, cancellationToken);
-        }
-        else
-        {
-            var nakDelayed = new ReadOnlySequence<byte>(Encoding.ASCII.GetBytes($"-NAK {{\"delay\": {delay.ToNanos()}}}"));
-            return SendAckAsync(nakDelayed, opts, cancellationToken);
-        }
+
+        var nakDelayed = new ReadOnlySequence<byte>(Encoding.ASCII.GetBytes($"-NAK {{\"delay\": {delay.ToNanos()}}}"));
+        return SendAckAsync(nakDelayed, opts, cancellationToken);
     }
 
     /// <inheritdoc />
     public ValueTask AckProgressAsync(AckOpts? opts = null, CancellationToken cancellationToken = default) => SendAckAsync(NatsJSConstants.AckProgress, opts, cancellationToken);
 
     /// <inheritdoc />
     public ValueTask AckTerminateAsync(AckOpts? opts = null, CancellationToken cancellationToken = default)
-        => AckTerminateInternalAsync(opts, null, cancellationToken);
-
-    /// <inheritdoc />
-    public ValueTask AckTerminateAsync(string reason, AckOpts? opts = null, CancellationToken cancellationToken = default)
-        => AckTerminateInternalAsync(opts, reason, cancellationToken);
-
-    private ValueTask AckTerminateInternalAsync(AckOpts? opts, string? reason, CancellationToken cancellationToken)
-    {
-        if (string.IsNullOrEmpty(reason))
-        {
-            return SendAckAsync(NatsJSConstants.AckTerminate, opts, cancellationToken);
-        }
-
-        return AckTerminateWithReasonAsync(reason!, opts, cancellationToken);
-    }
+        => string.IsNullOrEmpty(opts?.TerminateReason)
+            ? SendAckAsync(NatsJSConstants.AckTerminate, opts, cancellationToken)
+            : AckTerminateWithReasonAsync(opts!.Value.TerminateReason!, opts, cancellationToken);
 
     private async ValueTask AckTerminateWithReasonAsync(string reason, AckOpts? opts, CancellationToken cancellationToken)
     {
@@ -308,6 +280,17 @@ public readonly record struct AckOpts
     /// Ask server for an acknowledgment
     /// </summary>
     public bool? DoubleAck { get; init; }
+
+    /// <summary>
+    /// Delay for Nak before redelivery
+    /// </summary>
+    public TimeSpan? NakDelay { get; init; }
+
+    /// <summary>
+    /// Reason for AckTerminate
+    /// Requires NATS Server 2.10.4+.
+    /// </summary>
+    public string? TerminateReason { get; init; }
 }
 
 #if NETSTANDARD2_0

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

@@ -151,7 +151,7 @@ public async Task Terminate_with_reason(NatsRequestReplyMode mode)
         var next = await consumer.NextAsync<int>(cancellationToken: cts.Token);
         if (next is { } msg)
         {
-            await msg.AckTerminateAsync(reason: "test failure reason", cancellationToken: cts.Token);
+            await msg.AckTerminateAsync("test failure reason", cancellationToken: cts.Token);
             Assert.Equal(42, msg.Data);
 
             await Retry.Until("seen TERM", () => proxy.Frames.Any(f => f.Message.StartsWith("PUB $JS.ACK")));