diff --git a/src/NATS.Client.Core/Internal/NatsReadProtocolProcessor.cs b/src/NATS.Client.Core/Internal/NatsReadProtocolProcessor.cs index a4a5dc77a..ae74bc30d 100644 --- a/src/NATS.Client.Core/Internal/NatsReadProtocolProcessor.cs +++ b/src/NATS.Client.Core/Internal/NatsReadProtocolProcessor.cs @@ -21,12 +21,14 @@ internal sealed class NatsReadProtocolProcessor : IAsyncDisposable private readonly Task _infoParsed; // wait for an upgrade private readonly ConcurrentQueue _pingCommands; // wait for pong private readonly ILogger _logger; + private readonly Encoding _subjectEncoding; private readonly bool _trace; private int _disposed; public NatsReadProtocolProcessor(SocketConnectionWrapper socketConnection, NatsConnection connection, TaskCompletionSource waitForInfoSignal, TaskCompletionSource waitForPongOrErrorSignal, Task infoParsed) { _connection = connection; + _subjectEncoding = connection.Opts.SubjectEncoding; _logger = connection.Opts.LoggerFactory.CreateLogger(); _trace = _logger.IsEnabled(LogLevel.Trace); _waitForInfoSignal = waitForInfoSignal; @@ -471,7 +473,7 @@ private async ValueTask> DispatchCommandAsync(int code, R msgHeader.Split(out var sidBytes, out msgHeader); msgHeader.Split(out var replyToOrSizeBytes, out msgHeader); - var subject = Encoding.ASCII.GetString(subjectBytes); + var subject = _subjectEncoding.GetString(subjectBytes); if (msgHeader.Length == 0) { @@ -486,7 +488,7 @@ private async ValueTask> DispatchCommandAsync(int code, R var sid = GetInt32(sidBytes); var payloadLength = GetInt32(bytesSlice); - var replyTo = Encoding.ASCII.GetString(replyToBytes); + var replyTo = _subjectEncoding.GetString(replyToBytes); return (subject, sid, payloadLength, replyTo); } } @@ -526,7 +528,7 @@ private async ValueTask> DispatchCommandAsync(int code, R msgHeader.Split(out var replyToOrHeaderLenBytes, out msgHeader); msgHeader.Split(out var headerLenOrTotalLenBytes, out msgHeader); - var subject = Encoding.ASCII.GetString(subjectBytes); + var subject = _subjectEncoding.GetString(subjectBytes); var sid = GetInt32(sidBytes); // We don't have the optional reply-to field @@ -541,7 +543,7 @@ private async ValueTask> DispatchCommandAsync(int code, R else { var replyToBytes = replyToOrHeaderLenBytes; - var replyTo = Encoding.ASCII.GetString(replyToBytes); + var replyTo = _subjectEncoding.GetString(replyToBytes); var headerLen = GetInt32(headerLenOrTotalLenBytes); diff --git a/src/NATS.Client.Core/NatsOpts.cs b/src/NATS.Client.Core/NatsOpts.cs index 1281cfb46..b75f87e50 100644 --- a/src/NATS.Client.Core/NatsOpts.cs +++ b/src/NATS.Client.Core/NatsOpts.cs @@ -113,9 +113,25 @@ public sealed record NatsOpts public TimeSpan SubscriptionCleanUpInterval { get; init; } = TimeSpan.FromMinutes(5); + /// + /// Gets or sets encoding used for NATS message header names and values. (default: ASCII) + /// + /// + /// NATS headers follow HTTP/1.1 conventions where header field values are + /// restricted to visible US-ASCII characters per RFC 9110. Use base64 encoding + /// for non-ASCII data in header values. + /// public Encoding HeaderEncoding { get; init; } = Encoding.ASCII; - public Encoding SubjectEncoding { get; init; } = Encoding.ASCII; + /// + /// Gets or sets encoding used for NATS subjects and reply-to addresses. (default: UTF-8) + /// + /// + /// The NATS protocol specifies that subjects are UTF-8 encoded on the wire. + /// UTF-8 is backwards compatible with ASCII, so existing ASCII-only subjects + /// are unaffected by this default. + /// + public Encoding SubjectEncoding { get; init; } = Encoding.UTF8; public bool WaitUntilSent { get; init; } = false; diff --git a/tests/NATS.Client.Core2.Tests/Utf8SubjectTest.cs b/tests/NATS.Client.Core2.Tests/Utf8SubjectTest.cs new file mode 100644 index 000000000..00704fee7 --- /dev/null +++ b/tests/NATS.Client.Core2.Tests/Utf8SubjectTest.cs @@ -0,0 +1,220 @@ +using System.Text; +using NATS.Client.Core2.Tests; +using NATS.Client.TestUtilities; + +namespace NATS.Client.Core.Tests; + +public class Utf8SubjectMockServerTest +{ + [Fact] + public async Task Utf8_subject_is_decoded_correctly() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + + // UTF-8 subject with multi-byte characters + var utf8Subject = "test.ñoño.日本語"; + + await using var server = new MockServer( + async (client, cmd) => + { + if (cmd is { Name: "SUB", Subject: ">" }) + { + // Build raw MSG protocol line with UTF-8 encoded subject bytes. + // MockServer uses ISO 8859-1 so we write raw bytes to the stream + // to ensure multi-byte UTF-8 characters are sent correctly. + var payload = "hello"u8.ToArray(); + var msgLine = $"MSG {utf8Subject} {cmd.Sid} {payload.Length}\r\n"; + var msgLineBytes = Encoding.UTF8.GetBytes(msgLine); + + await client.Writer.FlushAsync().ConfigureAwait(false); + var stream = client.Writer.BaseStream; + await stream.WriteAsync(msgLineBytes, 0, msgLineBytes.Length).ConfigureAwait(false); + await stream.WriteAsync(payload, 0, payload.Length).ConfigureAwait(false); + var crlf = "\r\n"u8.ToArray(); + await stream.WriteAsync(crlf, 0, crlf.Length).ConfigureAwait(false); + await stream.FlushAsync().ConfigureAwait(false); + } + }, + cancellationToken: cts.Token); + + await using var nats = new NatsConnection(new NatsOpts { Url = server.Url }); + await nats.ConnectAsync(); + + await foreach (var msg in nats.SubscribeAsync(">", cancellationToken: cts.Token)) + { + Assert.Equal(utf8Subject, msg.Subject); + break; + } + } + + [Fact] + public async Task Emoji_subject_and_reply_to_are_decoded_correctly() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + + var emojiSubject = "events.🔥.📬"; + var emojiReplyTo = "_INBOX.🎉.reply"; + + await using var server = new MockServer( + async (client, cmd) => + { + if (cmd is { Name: "SUB", Subject: ">" }) + { + var payload = "data"u8.ToArray(); + + // MSG <#bytes>\r\n[payload]\r\n + var msgLine = $"MSG {emojiSubject} {cmd.Sid} {emojiReplyTo} {payload.Length}\r\n"; + var msgLineBytes = Encoding.UTF8.GetBytes(msgLine); + + await client.Writer.FlushAsync().ConfigureAwait(false); + var stream = client.Writer.BaseStream; + await stream.WriteAsync(msgLineBytes, 0, msgLineBytes.Length).ConfigureAwait(false); + await stream.WriteAsync(payload, 0, payload.Length).ConfigureAwait(false); + var crlf = "\r\n"u8.ToArray(); + await stream.WriteAsync(crlf, 0, crlf.Length).ConfigureAwait(false); + await stream.FlushAsync().ConfigureAwait(false); + } + }, + cancellationToken: cts.Token); + + await using var nats = new NatsConnection(new NatsOpts { Url = server.Url }); + await nats.ConnectAsync(); + + await foreach (var msg in nats.SubscribeAsync(">", cancellationToken: cts.Token)) + { + Assert.Equal(emojiSubject, msg.Subject); + Assert.Equal(emojiReplyTo, msg.ReplyTo); + break; + } + } + + [Fact] + public async Task Utf8_subject_with_hmsg_is_decoded_correctly_mock() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + + var utf8Subject = "hmsg.über.café"; + var utf8ReplyTo = "_INBOX.naïve.reply"; + var headers = "NATS/1.0\r\nX-Test: value\r\n\r\n"; + var payload = "payload"; + + await using var server = new MockServer( + async (client, cmd) => + { + if (cmd is { Name: "SUB", Subject: ">" }) + { + var headersBytes = Encoding.UTF8.GetBytes(headers); + var payloadBytes = Encoding.UTF8.GetBytes(payload); + var totalLen = headersBytes.Length + payloadBytes.Length; + + // HMSG <#header-bytes> <#total-bytes>\r\n[headers]\r\n\r\n[payload]\r\n + var msgLine = $"HMSG {utf8Subject} {cmd.Sid} {utf8ReplyTo} {headersBytes.Length} {totalLen}\r\n"; + var msgLineBytes = Encoding.UTF8.GetBytes(msgLine); + + await client.Writer.FlushAsync().ConfigureAwait(false); + var stream = client.Writer.BaseStream; + await stream.WriteAsync(msgLineBytes, 0, msgLineBytes.Length).ConfigureAwait(false); + await stream.WriteAsync(headersBytes, 0, headersBytes.Length).ConfigureAwait(false); + await stream.WriteAsync(payloadBytes, 0, payloadBytes.Length).ConfigureAwait(false); + var crlf = "\r\n"u8.ToArray(); + await stream.WriteAsync(crlf, 0, crlf.Length).ConfigureAwait(false); + await stream.FlushAsync().ConfigureAwait(false); + } + }, + cancellationToken: cts.Token); + + await using var nats = new NatsConnection(new NatsOpts { Url = server.Url }); + await nats.ConnectAsync(); + + await foreach (var msg in nats.SubscribeAsync(">", cancellationToken: cts.Token)) + { + Assert.Equal(utf8Subject, msg.Subject); + Assert.Equal(utf8ReplyTo, msg.ReplyTo); + Assert.Equal("value", msg.Headers?["X-Test"]); + break; + } + } +} + +[Collection("nats-server")] +public class Utf8SubjectServerTest +{ + private readonly NatsServerFixture _server; + + public Utf8SubjectServerTest(NatsServerFixture server) => _server = server; + + [Fact] + public async Task Utf8_subject_pub_sub_with_real_server() + { + await using var nats = new NatsConnection(new NatsOpts { Url = _server.Url }); + + var subject = "test.café.🔥"; + var sync = 0; + NatsMsg received = default; + + var sub = Task.Run(async () => + { + await foreach (var msg in nats.SubscribeAsync("test.>")) + { + if (msg.Subject == "test.sync") + { + Interlocked.Increment(ref sync); + continue; + } + + received = msg; + break; + } + }); + + await Retry.Until( + reason: "subscription is ready", + condition: () => Volatile.Read(ref sync) > 0, + action: async () => await nats.PublishAsync("test.sync"), + retryDelay: TimeSpan.FromSeconds(1)); + + await nats.PublishAsync(subject: subject, data: "hello"); + await sub; + + Assert.Equal(subject, received.Subject); + Assert.Equal("hello", received.Data); + } + + [Fact] + public async Task Utf8_subject_request_reply_with_real_server() + { + await using var nats = new NatsConnection(new NatsOpts { Url = _server.Url }); + + var subject = "svc.ñoño.日本語"; + var sync = 0; + + var responder = Task.Run(async () => + { + await foreach (var msg in nats.SubscribeAsync("svc.>")) + { + if (msg.Subject == "svc.sync") + { + Interlocked.Increment(ref sync); + continue; + } + + Assert.Equal(subject, msg.Subject); + Assert.NotNull(msg.ReplyTo); + await msg.ReplyAsync("pong"); + break; + } + }); + + await Retry.Until( + reason: "responder is ready", + condition: () => Volatile.Read(ref sync) > 0, + action: async () => await nats.PublishAsync("svc.sync"), + retryDelay: TimeSpan.FromSeconds(1)); + + var reply = await nats.RequestAsync(subject: subject, data: "ping"); + + Assert.Equal("pong", reply.Data); + + await responder; + } +}