From be4d997b0eb740134a7dc1f29dacfe955640954a Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Thu, 22 Jan 2026 12:54:34 +0000 Subject: [PATCH 1/2] Add TermWithReason support to AckTerminateAsync Add optional reason parameter that sends "+TERM " to the server, matching the Go client's TermWithReason functionality. The reason appears in JetStream advisory events and requires NATS Server 2.10.4+. Implementation uses ArrayPool to avoid allocations. Platform-specific code paths handle netstandard2.0 vs newer targets for optimal performance. --- src/NATS.Client.JetStream/NatsJSMsg.cs | 42 ++++++++++- .../DoubleAckNakDelayTests.cs | 74 +++++++++++++++++++ 2 files changed, 114 insertions(+), 2 deletions(-) diff --git a/src/NATS.Client.JetStream/NatsJSMsg.cs b/src/NATS.Client.JetStream/NatsJSMsg.cs index 401c0b0b4..335587919 100644 --- a/src/NATS.Client.JetStream/NatsJSMsg.cs +++ b/src/NATS.Client.JetStream/NatsJSMsg.cs @@ -125,9 +125,10 @@ public interface INatsJSMsg : INatsMsg /// Instructs the server to stop redelivery of the message without acknowledging it as successfully processed. /// /// Ack options. + /// Optional reason for termination, included in JetStream advisory events. Requires NATS Server 2.10.4+. /// A used to cancel the call. /// A representing the async call. - ValueTask AckTerminateAsync(AckOpts? opts = default, CancellationToken cancellationToken = default); + ValueTask AckTerminateAsync(AckOpts? opts = default, string? reason = null, CancellationToken cancellationToken = default); } /// @@ -136,6 +137,10 @@ public interface INatsJSMsg : INatsMsg /// User message type public readonly struct NatsJSMsg : INatsJSMsg { +#if NETSTANDARD2_0 + private static readonly byte[] TermPrefix = { (byte)'+', (byte)'T', (byte)'E', (byte)'R', (byte)'M', (byte)' ' }; +#endif + private readonly INatsJSContext _context; private readonly NatsMsg _msg; private readonly Lazy _replyToDateTimeAndSeq; @@ -262,9 +267,42 @@ public ValueTask NakAsync(AckOpts? opts = default, TimeSpan delay = default, Can /// Instructs the server to stop redelivery of the message without acknowledging it as successfully processed. /// /// Ack options. + /// Optional reason for termination, included in JetStream advisory events. Requires NATS Server 2.10.4+. /// A used to cancel the call. /// A representing the async call. - public ValueTask AckTerminateAsync(AckOpts? opts = default, CancellationToken cancellationToken = default) => SendAckAsync(NatsJSConstants.AckTerminate, opts, cancellationToken); + public ValueTask AckTerminateAsync(AckOpts? opts = default, string? reason = null, CancellationToken cancellationToken = default) + { + if (string.IsNullOrEmpty(reason)) + { + return SendAckAsync(NatsJSConstants.AckTerminate, opts, cancellationToken); + } + + return AckTerminateWithReasonAsync(reason!, opts, cancellationToken); + } + + private async ValueTask AckTerminateWithReasonAsync(string reason, AckOpts? opts, CancellationToken cancellationToken) + { + var reasonByteCount = Encoding.ASCII.GetByteCount(reason); + var totalLength = 6 + reasonByteCount; // "+TERM " is 6 bytes + + var buffer = ArrayPool.Shared.Rent(totalLength); + try + { +#if NETSTANDARD2_0 + Buffer.BlockCopy(TermPrefix, 0, buffer, 0, 6); + Encoding.ASCII.GetBytes(reason, 0, reason.Length, buffer, 6); +#else + "+TERM "u8.CopyTo(buffer.AsSpan()); + Encoding.ASCII.GetBytes(reason.AsSpan(), buffer.AsSpan(6)); +#endif + + await SendAckAsync(new ReadOnlySequence(buffer, 0, totalLength), opts, cancellationToken); + } + finally + { + ArrayPool.Shared.Return(buffer); + } + } private async ValueTask SendAckAsync(ReadOnlySequence payload, AckOpts? opts = default, CancellationToken cancellationToken = default) { diff --git a/tests/NATS.Client.JetStream.Tests/DoubleAckNakDelayTests.cs b/tests/NATS.Client.JetStream.Tests/DoubleAckNakDelayTests.cs index fa19e9ac9..c9eae911e 100644 --- a/tests/NATS.Client.JetStream.Tests/DoubleAckNakDelayTests.cs +++ b/tests/NATS.Client.JetStream.Tests/DoubleAckNakDelayTests.cs @@ -91,4 +91,78 @@ public async Task Delay_nak_received_messages(NatsRequestReplyMode mode) Assert.Fail("No message received"); } } + + [Theory] + [InlineData(NatsRequestReplyMode.Direct)] + [InlineData(NatsRequestReplyMode.SharedInbox)] + public async Task Terminate_without_reason(NatsRequestReplyMode mode) + { + var proxy = _server.CreateProxy(); + await using var nats = proxy.CreateNatsConnection(mode); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); + + var js = new NatsJSContext(nats); + + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + + await js.CreateStreamAsync($"{prefix}s1", [$"{prefix}s1.*"], cts.Token); + var consumer = await js.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token); + + var ack = await js.PublishAsync($"{prefix}s1.foo", 42, cancellationToken: cts.Token); + ack.EnsureSuccess(); + var next = await consumer.NextAsync(cancellationToken: cts.Token); + if (next is { } msg) + { + await msg.AckTerminateAsync(cancellationToken: cts.Token); + Assert.Equal(42, msg.Data); + + await Retry.Until("seen TERM", () => proxy.Frames.Any(f => f.Message.StartsWith("PUB $JS.ACK"))); + + var termFrame = proxy.Frames.Single(f => f.Message.StartsWith("PUB $JS.ACK")); + + Assert.Matches(@"\+TERM\s*$", termFrame.Message); + } + else + { + Assert.Fail("No message received"); + } + } + + [Theory] + [InlineData(NatsRequestReplyMode.Direct)] + [InlineData(NatsRequestReplyMode.SharedInbox)] + public async Task Terminate_with_reason(NatsRequestReplyMode mode) + { + var proxy = _server.CreateProxy(); + await using var nats = proxy.CreateNatsConnection(mode); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); + + var js = new NatsJSContext(nats); + + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + + await js.CreateStreamAsync($"{prefix}s1", [$"{prefix}s1.*"], cts.Token); + var consumer = await js.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token); + + var ack = await js.PublishAsync($"{prefix}s1.foo", 42, cancellationToken: cts.Token); + ack.EnsureSuccess(); + var next = await consumer.NextAsync(cancellationToken: cts.Token); + if (next is { } msg) + { + await msg.AckTerminateAsync(reason: "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"))); + + var termFrame = proxy.Frames.Single(f => f.Message.StartsWith("PUB $JS.ACK")); + + Assert.Contains("+TERM test failure reason", termFrame.Message); + } + else + { + Assert.Fail("No message received"); + } + } } From 48d1fe30159dafc052cf0fb1cd8d0661a6e7158f Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Fri, 23 Jan 2026 11:57:30 +0000 Subject: [PATCH 2/2] use overload to not break compat also code cleanup --- src/NATS.Client.JetStream/NatsJSMsg.cs | 155 +++++++++---------------- 1 file changed, 55 insertions(+), 100 deletions(-) diff --git a/src/NATS.Client.JetStream/NatsJSMsg.cs b/src/NATS.Client.JetStream/NatsJSMsg.cs index 335587919..078331568 100644 --- a/src/NATS.Client.JetStream/NatsJSMsg.cs +++ b/src/NATS.Client.JetStream/NatsJSMsg.cs @@ -8,7 +8,7 @@ namespace NATS.Client.JetStream; /// /// This interface provides an optional contract when passing -/// messages to processing methods which is usually helpful in +/// messages to processing methods, which is usually helpful in /// creating test doubles in unit testing. /// /// @@ -30,12 +30,12 @@ namespace NATS.Client.JetStream; public interface INatsJSMsg : INatsMsg { /// - /// Subject of the user message. + /// Get subject of the user message. /// string Subject { get; } /// - /// Message size in bytes. + /// Get message size in bytes. /// /// /// Message size is calculated using the same method NATS server uses: @@ -46,12 +46,12 @@ public interface INatsJSMsg : INatsMsg int Size { get; } /// - /// Deserialized user data. + /// Get deserialized user data. /// T? Data { get; } /// - /// The connection messages was delivered on. + /// Get the connection messages were delivered on. /// INatsConnection? Connection { get; } @@ -80,7 +80,7 @@ public interface INatsJSMsg : INatsMsg /// A used to cancel the command. /// A that represents the asynchronous send operation. [Obsolete("ReplyAsync is not valid when using JetStream. The reply message will never reach the Requestor as NATS will reply to all JetStream publish by default.")] - ValueTask ReplyAsync(NatsHeaders? headers = default, string? replyTo = default, NatsPubOpts? opts = default, CancellationToken cancellationToken = default); + ValueTask ReplyAsync(NatsHeaders? headers = null, string? replyTo = null, NatsPubOpts? opts = null, CancellationToken cancellationToken = default); /// /// Acknowledges the message was completely handled. @@ -88,7 +88,7 @@ public interface INatsJSMsg : INatsMsg /// Ack options. /// A used to cancel the call. /// A representing the async call. - ValueTask AckAsync(AckOpts? opts = default, CancellationToken cancellationToken = default); + ValueTask AckAsync(AckOpts? opts = null, CancellationToken cancellationToken = default); /// /// Signals that the message will not be processed now and processing can move onto the next message. @@ -101,7 +101,7 @@ public interface INatsJSMsg : INatsMsg /// Messages rejected using -NAK will be resent by the NATS JetStream server after the configured timeout /// or the delay parameter if it's specified. /// - ValueTask NakAsync(AckOpts? opts = default, TimeSpan delay = default, CancellationToken cancellationToken = default); + ValueTask NakAsync(AckOpts? opts = null, TimeSpan delay = default, CancellationToken cancellationToken = default); /// /// Indicates that work is ongoing and the wait period should be extended. @@ -112,23 +112,31 @@ public interface INatsJSMsg : INatsMsg /// /// /// Time period is defined by the consumer's ack_wait configuration on the server which is - /// defined as how long to allow messages to remain un-acknowledged before attempting redelivery. + /// defined as how long to allow messages to remain unacknowledged before attempting redelivery. /// /// /// This message must be sent before the ack_wait period elapses. The period should be extended /// by another amount of time equal to ack_wait by the NATS JetStream server. /// /// - ValueTask AckProgressAsync(AckOpts? opts = default, CancellationToken cancellationToken = default); + ValueTask AckProgressAsync(AckOpts? opts = null, CancellationToken cancellationToken = default); /// /// Instructs the server to stop redelivery of the message without acknowledging it as successfully processed. /// /// Ack options. + /// A used to cancel the call. + /// A representing the async call. + ValueTask AckTerminateAsync(AckOpts? opts = null, CancellationToken cancellationToken = default); + + /// + /// Instructs the server to stop redelivery of the message without acknowledging it as successfully processed. + /// /// Optional reason for termination, included in JetStream advisory events. Requires NATS Server 2.10.4+. + /// Ack options. /// A used to cancel the call. /// A representing the async call. - ValueTask AckTerminateAsync(AckOpts? opts = default, string? reason = null, CancellationToken cancellationToken = default); + ValueTask AckTerminateAsync(string reason, AckOpts? opts = null, CancellationToken cancellationToken = default); } /// @@ -137,10 +145,6 @@ public interface INatsJSMsg : INatsMsg /// User message type public readonly struct NatsJSMsg : INatsJSMsg { -#if NETSTANDARD2_0 - private static readonly byte[] TermPrefix = { (byte)'+', (byte)'T', (byte)'E', (byte)'R', (byte)'M', (byte)' ' }; -#endif - private readonly INatsJSContext _context; private readonly NatsMsg _msg; private readonly Lazy _replyToDateTimeAndSeq; @@ -152,45 +156,25 @@ public NatsJSMsg(NatsMsg msg, INatsJSContext context) _replyToDateTimeAndSeq = new Lazy(() => ReplyToDateTimeAndSeq.Parse(msg.ReplyTo)); } - /// - /// Subject of the user message. - /// + /// public string Subject => _msg.Subject; - /// - /// Message size in bytes. - /// - /// - /// Message size is calculated using the same method NATS server uses: - /// - /// int size = subject.Length + replyTo.Length + headers.Length + payload.Length; - /// - /// + /// public int Size => _msg.Size; - /// - /// Headers of the user message if set. - /// + /// public NatsHeaders? Headers => _msg.Headers; - /// - /// Deserialized user data. - /// + /// public T? Data => _msg.Data; - /// - /// The connection messages was delivered on. - /// + /// public INatsConnection? Connection => _msg.Connection; - /// - /// Additional metadata about the message. - /// + /// public NatsJSMsgMetadata? Metadata => _replyToDateTimeAndSeq.Value; - /// - /// The reply subject that subscribers can use to send a response back to the publisher/requester. - /// + /// public string? ReplyTo => _msg.ReplyTo; /// @@ -201,40 +185,18 @@ public NatsJSMsg(NatsMsg msg, INatsJSContext context) /// public void EnsureSuccess() => _msg.EnsureSuccess(); - /// - /// Reply with an empty message. - /// - /// Optional message headers. - /// Optional reply-to subject. - /// A for publishing options. - /// A used to cancel the command. - /// A that represents the asynchronous send operation. + /// [Obsolete("ReplyAsync is not valid when using JetStream. The reply message will never reach the Requestor as NATS will reply to all JetStream publish by default.")] - public ValueTask ReplyAsync(NatsHeaders? headers = default, string? replyTo = default, NatsPubOpts? opts = default, CancellationToken cancellationToken = default) => + public ValueTask ReplyAsync(NatsHeaders? headers = null, string? replyTo = null, NatsPubOpts? opts = null, CancellationToken cancellationToken = default) => _msg.ReplyAsync(headers, replyTo, opts, cancellationToken); - /// - /// Acknowledges the message was completely handled. - /// - /// Ack options. - /// A used to cancel the call. - /// A representing the async call. - public ValueTask AckAsync(AckOpts? opts = default, CancellationToken cancellationToken = default) => SendAckAsync(NatsJSConstants.Ack, opts, cancellationToken); + /// + public ValueTask AckAsync(AckOpts? opts = null, CancellationToken cancellationToken = default) => SendAckAsync(NatsJSConstants.Ack, opts, cancellationToken); - /// - /// Signals that the message will not be processed now and processing can move onto the next message. - /// - /// Delay redelivery of the message. - /// Ack options. - /// A used to cancel the call. - /// A representing the async call. - /// - /// Messages rejected using -NAK will be resent by the NATS JetStream server after the configured timeout - /// or the delay parameter if it's specified. - /// - public ValueTask NakAsync(AckOpts? opts = default, TimeSpan delay = default, CancellationToken cancellationToken = default) + /// + public ValueTask NakAsync(AckOpts? opts = null, TimeSpan delay = default, CancellationToken cancellationToken = default) { - if (delay == default) + if (delay == TimeSpan.Zero) { return SendAckAsync(NatsJSConstants.Nak, opts, cancellationToken); } @@ -245,32 +207,18 @@ public ValueTask NakAsync(AckOpts? opts = default, TimeSpan delay = default, Can } } - /// - /// Indicates that work is ongoing and the wait period should be extended. - /// - /// Ack options. - /// A used to cancel the call. - /// A representing the async call. - /// - /// - /// Time period is defined by the consumer's ack_wait configuration on the server which is - /// defined as how long to allow messages to remain un-acknowledged before attempting redelivery. - /// - /// - /// This message must be sent before the ack_wait period elapses. The period should be extended - /// by another amount of time equal to ack_wait by the NATS JetStream server. - /// - /// - public ValueTask AckProgressAsync(AckOpts? opts = default, CancellationToken cancellationToken = default) => SendAckAsync(NatsJSConstants.AckProgress, opts, cancellationToken); + /// + public ValueTask AckProgressAsync(AckOpts? opts = null, CancellationToken cancellationToken = default) => SendAckAsync(NatsJSConstants.AckProgress, opts, cancellationToken); - /// - /// Instructs the server to stop redelivery of the message without acknowledging it as successfully processed. - /// - /// Ack options. - /// Optional reason for termination, included in JetStream advisory events. Requires NATS Server 2.10.4+. - /// A used to cancel the call. - /// A representing the async call. - public ValueTask AckTerminateAsync(AckOpts? opts = default, string? reason = null, CancellationToken cancellationToken = default) + /// + public ValueTask AckTerminateAsync(AckOpts? opts = null, CancellationToken cancellationToken = default) + => AckTerminateInternalAsync(opts, null, cancellationToken); + + /// + 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)) { @@ -283,13 +231,13 @@ public ValueTask AckTerminateAsync(AckOpts? opts = default, string? reason = nul private async ValueTask AckTerminateWithReasonAsync(string reason, AckOpts? opts, CancellationToken cancellationToken) { var reasonByteCount = Encoding.ASCII.GetByteCount(reason); - var totalLength = 6 + reasonByteCount; // "+TERM " is 6 bytes + var totalLength = 6 + reasonByteCount; // `+TERM ` is 6 bytes var buffer = ArrayPool.Shared.Rent(totalLength); try { #if NETSTANDARD2_0 - Buffer.BlockCopy(TermPrefix, 0, buffer, 0, 6); + Buffer.BlockCopy(NatsJSMsgConstants.TermPrefix, 0, buffer, 0, 6); Encoding.ASCII.GetBytes(reason, 0, reason.Length, buffer, 6); #else "+TERM "u8.CopyTo(buffer.AsSpan()); @@ -304,7 +252,7 @@ private async ValueTask AckTerminateWithReasonAsync(string reason, AckOpts? opts } } - private async ValueTask SendAckAsync(ReadOnlySequence payload, AckOpts? opts = default, CancellationToken cancellationToken = default) + private async ValueTask SendAckAsync(ReadOnlySequence payload, AckOpts? opts = null, CancellationToken cancellationToken = default) { CheckPreconditions(); @@ -336,7 +284,7 @@ await _msg.ReplyAsync( [MemberNotNull(nameof(ReplyTo))] private void CheckPreconditions() { - if (Connection == default) + if (Connection == null) { throw new NatsException("unable to send acknowledgment; message did not originate from a consumer"); } @@ -361,3 +309,10 @@ public readonly record struct AckOpts /// public bool? DoubleAck { get; init; } } + +#if NETSTANDARD2_0 +internal static class NatsJSMsgConstants +{ + internal static readonly byte[] TermPrefix = "+TERM "u8.ToArray(); +} +#endif