diff --git a/Directory.Build.props b/Directory.Build.props index 35d960128..3f3104456 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -9,6 +9,9 @@ false + + + 2.6.0 diff --git a/README.md b/README.md index ba13d029c..a5a9452d5 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,10 @@ # NATS .NET +> [!IMPORTANT] +> **NATS .NET v3 is in development on the [`release/3.0`](https://github.com/nats-io/nats.net/tree/release/3.0) branch!** +> v3 adds .NET 10 support and drops .NET 6.0. Target frameworks: `netstandard2.0`, `netstandard2.1`, `net8.0`, `net10.0`. + NATS .NET is the .NET client for NATS, a distributed messaging system. It provides pub/sub and request/reply (Core NATS), streaming and persistence (JetStream), Key-Value Store, Object Store, and Services. @@ -83,11 +87,6 @@ var obj = nc.CreateObjectStoreContext(); var svc = nc.CreateServicesContext(); ``` -> [!NOTE] -> **We are not testing with .NET 6.0 target anymore** even though it is still targeted by the library. -> This is to reduce the number of test runs and speed up the CI process as well as to prepare for -> the next major version, possibly dropping .NET 6.0 support. - > [!NOTE] > **Don't confuse NuGet packages!** > NATS .NET package on NuGet is called [NATS.Net](https://www.nuget.org/packages/NATS.Net). @@ -95,7 +94,7 @@ var svc = nc.CreateServicesContext(); > and will be deprecated eventually. > [!TIP] -> NATS .NET now supports **.NET Standard** 2.0 and 2.1 along with .NET 6.0 and 8.0, +> NATS .NET supports **.NET Standard** 2.0 and 2.1 along with .NET 8.0 and 10.0, > which means you can also use it with **.NET Framework** 4.6.2+ and **Unity** 2018.1+. ### What is NATS? @@ -111,7 +110,7 @@ Head over to [NATS documentation](https://docs.nats.io/nats-concepts/overview) f ## NATS .NET Goals - Only support Async I/O (async/await) -- Target .NET Standard 2.0, 2.1, and the two most recent .NET LTS releases (currently .NET 6.0 & .NET 8.0) +- Target .NET Standard 2.0, 2.1, and the two most recent .NET LTS releases (currently .NET 8.0 & .NET 10.0) ## Packages @@ -123,7 +122,8 @@ Head over to [NATS documentation](https://docs.nats.io/nats-concepts/overview) f - **NATS.Client.Services**: [Services](https://docs.nats.io/using-nats/developer/services) - **NATS.Client.Simplified**: simplify common use cases especially for beginners - **NATS.Client.Serializers.Json**: JSON serializer for ad-hoc types -- **NATS.Extensions.Microsoft.DependencyInjection**: extension to configure DI container +- **NATS.Client.Hosting**: DI integration for AOT-compatible and minimal-dependency scenarios +- **NATS.Extensions.Microsoft.DependencyInjection**: DI integration with ad-hoc JSON serialization enabled by default ## Client and Orbit diff --git a/sandbox/MicroBenchmark/EncodingPolyfillBench.cs b/sandbox/MicroBenchmark/EncodingPolyfillBench.cs new file mode 100644 index 000000000..a9e08557b --- /dev/null +++ b/sandbox/MicroBenchmark/EncodingPolyfillBench.cs @@ -0,0 +1,143 @@ +using System.Buffers; +using System.Text; +using BenchmarkDotNet.Attributes; + +namespace MicroBenchmark; + +/// +/// Benchmarks comparing old (allocating) vs new (zero-alloc) encoding polyfills +/// used in NETSTANDARD builds for ProtocolWriter, HeaderWriter, and ReadProtocolProcessor. +/// +[MemoryDiagnoser] +public class EncodingPolyfillBench +{ + private const string ShortSubject = "foo.bar.baz"; + private const string MediumSubject = "my-service.requests.user.profile.update"; + private const string LongHeader = "X-Nats-Stream: my-persistent-stream-name-that-is-quite-long"; + + private static readonly Encoding Utf8 = Encoding.UTF8; + private static readonly Encoding Ascii = Encoding.ASCII; + + private byte[] _shortBytes = null!; + private byte[] _mediumBytes = null!; + private byte[] _longBytes = null!; + private byte[] _destBuffer = null!; + + [GlobalSetup] + public void Setup() + { + _shortBytes = Ascii.GetBytes(ShortSubject); + _mediumBytes = Ascii.GetBytes(MediumSubject); + _longBytes = Utf8.GetBytes(LongHeader); + _destBuffer = new byte[256]; + } + + [Benchmark(Description = "GetBytes_Old_Short")] + public int GetBytes_Old_Short() => GetBytesOld(Ascii, ShortSubject, _destBuffer); + + [Benchmark(Description = "GetBytes_New_Short")] + public int GetBytes_New_Short() => GetBytesNew(Ascii, ShortSubject, _destBuffer); + + [Benchmark(Description = "GetBytes_Old_Medium")] + public int GetBytes_Old_Medium() => GetBytesOld(Ascii, MediumSubject, _destBuffer); + + [Benchmark(Description = "GetBytes_New_Medium")] + public int GetBytes_New_Medium() => GetBytesNew(Ascii, MediumSubject, _destBuffer); + + [Benchmark(Description = "GetBytes_Old_Long")] + public int GetBytes_Old_Long() => GetBytesOld(Utf8, LongHeader, _destBuffer); + + [Benchmark(Description = "GetBytes_New_Long")] + public int GetBytes_New_Long() => GetBytesNew(Utf8, LongHeader, _destBuffer); + + [Benchmark(Description = "GetString_Old_Short")] + public string GetString_Old_Short() => GetStringOld(Ascii, _shortBytes); + + [Benchmark(Description = "GetString_New_Short")] + public string GetString_New_Short() => GetStringNew(Ascii, _shortBytes); + + [Benchmark(Description = "GetString_Old_Medium")] + public string GetString_Old_Medium() => GetStringOld(Ascii, _mediumBytes); + + [Benchmark(Description = "GetString_New_Medium")] + public string GetString_New_Medium() => GetStringNew(Ascii, _mediumBytes); + + [Benchmark(Description = "GetString_Old_Long")] + public string GetString_Old_Long() => GetStringOld(Utf8, _longBytes); + + [Benchmark(Description = "GetString_New_Long")] + public string GetString_New_Long() => GetStringNew(Utf8, _longBytes); + + [Benchmark(Description = "GetBytesWriter_Old_Medium")] + public void GetBytesWriter_Old_Medium() + { + var writer = new ArrayBufferWriter(256); + GetBytesWriterOld(Utf8, MediumSubject, writer); + } + + [Benchmark(Description = "GetBytesWriter_New_Medium")] + public void GetBytesWriter_New_Medium() + { + var writer = new ArrayBufferWriter(256); + GetBytesWriterNew(Utf8, MediumSubject, writer); + } + + private static int GetBytesOld(Encoding encoding, string chars, Span bytes) + { + var buffer = encoding.GetBytes(chars); + buffer.AsSpan().CopyTo(bytes); + return buffer.Length; + } + + private static string GetStringOld(Encoding encoding, ReadOnlySpan buffer) + { + return encoding.GetString(buffer.ToArray()); + } + + private static void GetBytesWriterOld(Encoding encoding, string chars, IBufferWriter bw) + { + var buffer = encoding.GetBytes(chars); + bw.Write(buffer); + } + + private static unsafe int GetBytesNew(Encoding encoding, string chars, Span bytes) + { + if (chars.Length == 0) + { + return 0; + } + + fixed (char* charPtr = chars) + { + fixed (byte* bytePtr = bytes) + { + return encoding.GetBytes(charPtr, chars.Length, bytePtr, bytes.Length); + } + } + } + + private static unsafe string GetStringNew(Encoding encoding, ReadOnlySpan buffer) + { + if (buffer.IsEmpty) + { + return string.Empty; + } + + fixed (byte* ptr = buffer) + { + return encoding.GetString(ptr, buffer.Length); + } + } + + private static void GetBytesWriterNew(Encoding encoding, string chars, IBufferWriter bw) + { + var byteCount = encoding.GetByteCount(chars); + var span = bw.GetSpan(byteCount); +#if NET8_0_OR_GREATER + encoding.GetBytes(chars.AsSpan(), span); +#else + GetBytesNew(encoding, chars, span); +#endif + bw.Advance(byteCount); + } +} diff --git a/sandbox/MicroBenchmark/LockBench.cs b/sandbox/MicroBenchmark/LockBench.cs new file mode 100644 index 000000000..f4dffcc5a --- /dev/null +++ b/sandbox/MicroBenchmark/LockBench.cs @@ -0,0 +1,61 @@ +using BenchmarkDotNet.Attributes; +using BenchmarkDotNet.Jobs; + +namespace MicroBenchmark; + +// Compares object lock (all TFMs) vs System.Threading.Lock (NET9+). +// On net8.0 both methods use object, so they serve as a same-TFM baseline. +// On net10.0 SystemLock uses System.Threading.Lock whose EnterScope() path +// avoids the Monitor sync-block machinery. +[SimpleJob(RuntimeMoniker.Net80)] +[SimpleJob(RuntimeMoniker.Net10_0)] +[MemoryDiagnoser] +public class LockBench +{ + private readonly object _objectLock = new(); + +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _systemLock = new(); +#else + private readonly object _systemLock = new(); +#endif + + private int _value; + + // Uncontended empty lock/unlock -- measures raw locking overhead. + [Benchmark(Baseline = true)] + public void ObjectLock_Empty() + { + lock (_objectLock) + { + } + } + + [Benchmark] + public void SystemLock_Empty() + { + lock (_systemLock) + { + } + } + + // Uncontended lock with a trivial increment -- closer to real usage where + // the lock body does a small amount of work. + [Benchmark] + public void ObjectLock_Increment() + { + lock (_objectLock) + { + _value++; + } + } + + [Benchmark] + public void SystemLock_Increment() + { + lock (_systemLock) + { + _value++; + } + } +} diff --git a/sandbox/MicroBenchmark/MicroBenchmark.csproj b/sandbox/MicroBenchmark/MicroBenchmark.csproj index 414afa07d..5cff5ba9c 100644 --- a/sandbox/MicroBenchmark/MicroBenchmark.csproj +++ b/sandbox/MicroBenchmark/MicroBenchmark.csproj @@ -2,19 +2,19 @@ Exe - net8.0 + net8.0;net10.0 $(TargetFrameworks);net481 enable enable false true - 12 + latest - + - + diff --git a/sandbox/MicroBenchmark/PublishRuntimeBench.cs b/sandbox/MicroBenchmark/PublishRuntimeBench.cs new file mode 100644 index 000000000..fe127d6cb --- /dev/null +++ b/sandbox/MicroBenchmark/PublishRuntimeBench.cs @@ -0,0 +1,44 @@ +using BenchmarkDotNet.Attributes; +using BenchmarkDotNet.Jobs; +using NATS.Client.Core; + +#pragma warning disable CS8618 + +namespace MicroBenchmark; + +// Tight serial publish loop across runtimes. +// Measures end-to-end throughput including buffering, pipe writes, and socket I/O. +// The Lock change affects CommandWriter._lock on each publish, but the body under +// the lock dominates, so any visible delta here reflects runtime improvements broadly. +// Requires a running NATS server on localhost:4222. +[SimpleJob(RuntimeMoniker.Net80)] +[SimpleJob(RuntimeMoniker.Net10_0)] +[MemoryDiagnoser] +public class PublishRuntimeBench +{ + private const int Msgs = 100_000; + private static readonly byte[] Payload = new byte[128]; + + private NatsConnection _nats; + + [GlobalSetup] + public async Task Setup() + { + _nats = new NatsConnection(); + await _nats.ConnectAsync(); + } + + [GlobalCleanup] + public async Task Cleanup() => await _nats.DisposeAsync(); + + [Benchmark] + public async Task Publish100k() + { + for (var i = 0; i < Msgs; i++) + { + await _nats.PublishAsync("bench", Payload); + } + + await _nats.PingAsync(); + } +} diff --git a/src/Directory.Build.props b/src/Directory.Build.props index 571b8972d..b67be741d 100644 --- a/src/Directory.Build.props +++ b/src/Directory.Build.props @@ -2,14 +2,14 @@ - netstandard2.0;netstandard2.1;net6.0;net8.0 + netstandard2.0;netstandard2.1;net8.0;net10.0 enable enable true - true - true - + true + true + $([System.IO.File]::ReadAllText("$(MSBuildThisFileDirectory)..\version.txt")) $(Version) diff --git a/src/NATS.Client.Abstractions/INatsHeaders.cs b/src/NATS.Client.Abstractions/INatsHeaders.cs new file mode 100644 index 000000000..4c1cd6c53 --- /dev/null +++ b/src/NATS.Client.Abstractions/INatsHeaders.cs @@ -0,0 +1,10 @@ +using Microsoft.Extensions.Primitives; + +namespace NATS.Client.Core; + +/// +/// Represents NATS message headers as a dictionary of string keys and values. +/// +public interface INatsHeaders : IDictionary +{ +} diff --git a/src/NATS.Client.Abstractions/INatsSerialize.cs b/src/NATS.Client.Abstractions/INatsSerialize.cs index 343f50880..c901b4289 100644 --- a/src/NATS.Client.Abstractions/INatsSerialize.cs +++ b/src/NATS.Client.Abstractions/INatsSerialize.cs @@ -44,9 +44,89 @@ public interface INatsDeserialize T? Deserialize(in ReadOnlySequence buffer); } +/// +/// Extended serializer interface with access to message context during serialization. +/// +/// Serialized object type +public interface INatsSerializeWithContext : INatsSerialize +{ + /// + /// Serialize value to buffer with message context. + /// + /// Buffer to write the serialized data. + /// Object to be serialized. + /// Message envelope metadata. + void Serialize(IBufferWriter bufferWriter, T value, in NatsMsgContext context); +} + +/// +/// Extended deserializer interface with access to message context during deserialization. +/// +/// Deserialized object type +public interface INatsDeserializeWithContext : INatsDeserialize +{ + /// + /// Deserialize value from buffer with message context. + /// + /// Buffer with the serialized data. + /// Message envelope metadata. + /// Deserialized object + T? Deserialize(in ReadOnlySequence buffer, in NatsMsgContext context); +} + +/// +/// Combined context-aware serializer interface that supports both serialization and deserialization +/// with access to message context. +/// +/// Object type +public interface INatsSerializerWithContext : INatsSerializeWithContext, INatsDeserializeWithContext +{ +} + +/// +/// Registry for serializers and deserializers. +/// public interface INatsSerializerRegistry { INatsSerialize GetSerializer(); INatsDeserialize GetDeserializer(); } + +/// +/// Extension methods to support context-aware serialization with fallback to standard serialization. +/// +/// +/// Context-aware serializers that want to mutate must check +/// for null before doing so: the library passes whatever headers the caller supplied to +/// PublishAsync, and does not allocate a new for serializers that +/// opt in to context. Callers who want header-mutating behavior must pass a non-null +/// headers instance to PublishAsync. +/// +public static class NatsSerializationExtensions +{ + /// + /// Serializes the value with message context, falling back to standard serialization if not supported. + /// + public static void Serialize(this INatsSerialize serializer, IBufferWriter bufferWriter, T value, in NatsMsgContext context) + { + if (serializer is INatsSerializeWithContext withContext) + { + withContext.Serialize(bufferWriter, value, in context); + return; + } + + serializer.Serialize(bufferWriter, value); + } + + /// + /// Deserializes the value with message context, falling back to standard deserialization if not supported. + /// + public static T? Deserialize(this INatsDeserialize deserializer, in ReadOnlySequence buffer, in NatsMsgContext context) + { + if (deserializer is INatsDeserializeWithContext withContext) + return withContext.Deserialize(buffer, in context); + + return deserializer.Deserialize(buffer); + } +} diff --git a/src/NATS.Client.Abstractions/NATS.Client.Abstractions.csproj b/src/NATS.Client.Abstractions/NATS.Client.Abstractions.csproj index 6ddef1fc0..e4e38515b 100644 --- a/src/NATS.Client.Abstractions/NATS.Client.Abstractions.csproj +++ b/src/NATS.Client.Abstractions/NATS.Client.Abstractions.csproj @@ -8,6 +8,7 @@ + all runtime; build; native; contentfiles; analyzers; buildtransitive diff --git a/src/NATS.Client.Abstractions/NatsMsgContext.cs b/src/NATS.Client.Abstractions/NatsMsgContext.cs new file mode 100644 index 000000000..420600d33 --- /dev/null +++ b/src/NATS.Client.Abstractions/NatsMsgContext.cs @@ -0,0 +1,32 @@ +namespace NATS.Client.Core; + +/// +/// Provides message envelope metadata available during serialization and deserialization. +/// +/// +/// Use the constructor to create instances. A default(NatsMsgContext) value has a null +/// and is not a valid context; the library never passes such a value to +/// user code. +/// +public readonly struct NatsMsgContext +{ + /// Creates a new . + /// Subject the message was published to. + /// Optional reply-to subject. + /// Optional message headers. + public NatsMsgContext(string subject, string? replyTo = null, INatsHeaders? headers = null) + { + Subject = subject ?? throw new ArgumentNullException(nameof(subject)); + ReplyTo = replyTo; + Headers = headers; + } + + /// Subject the message was published to. + public string Subject { get; } + + /// Optional reply-to subject. + public string? ReplyTo { get; } + + /// Optional message headers. May be null if the caller did not supply headers. + public INatsHeaders? Headers { get; } +} diff --git a/src/NATS.Client.Core/Commands/CommandWriter.cs b/src/NATS.Client.Core/Commands/CommandWriter.cs index 2178c0f5a..790238dee 100644 --- a/src/NATS.Client.Core/Commands/CommandWriter.cs +++ b/src/NATS.Client.Core/Commands/CommandWriter.cs @@ -36,7 +36,11 @@ internal sealed class CommandWriter : IAsyncDisposable private readonly NatsConnection _connection; private readonly ObjectPool _pool; private readonly int _arrayPoolInitialSize; +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _lock = new(); +#else private readonly object _lock = new(); +#endif private readonly CancellationTokenSource _cts; private readonly ConnectionStatsCounter _counter; private readonly Memory _consolidateMem = new byte[SendMemSize].AsMemory(); @@ -334,12 +338,12 @@ public ValueTask PublishAsync(string subject, T? value, NatsHeaders? headers, try { + if (value != null) + serializer.Serialize(payloadBuffer, value, new NatsMsgContext(subject, replyTo, headers)); + if (headers != null) _headerWriter.Write(headersBuffer!, headers); - if (value != null) - serializer.Serialize(payloadBuffer, value); - var size = payloadBuffer.WrittenMemory.Length + (headersBuffer?.WrittenMemory.Length ?? 0); if (_connection.ServerInfo is { } info && size > info.MaxPayload) { @@ -957,7 +961,11 @@ private async ValueTask UnsubscribeStateMachineAsync(bool lockHeld, int sid, int private class PartialSendFailureCounter { private const int MaxRetry = 1; +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _gate = new(); +#else private readonly object _gate = new(); +#endif private int _count; public bool Failed() diff --git a/src/NATS.Client.Core/Internal/NatsPipeliningWriteProtocolProcessor.cs b/src/NATS.Client.Core/Internal/NatsPipeliningWriteProtocolProcessor.cs deleted file mode 100644 index 922cb5b95..000000000 --- a/src/NATS.Client.Core/Internal/NatsPipeliningWriteProtocolProcessor.cs +++ /dev/null @@ -1,309 +0,0 @@ -// using System.Buffers; -// using System.Diagnostics; -// using System.IO.Pipelines; -// using System.Threading.Channels; -// using Microsoft.Extensions.Logging; -// using NATS.Client.Core.Commands; -// -// namespace NATS.Client.Core.Internal; -// -// internal sealed class NatsPipeliningWriteProtocolProcessor : IAsyncDisposable -// { -// private readonly CancellationTokenSource _cts; -// private readonly ConnectionStatsCounter _counter; -// private readonly NatsOpts _opts; -// private readonly PipeReader _pipeReader; -// private readonly Queue _inFlightCommands; -// private readonly ChannelReader _queuedCommandReader; -// private readonly ISocketConnection _socketConnection; -// private readonly Stopwatch _stopwatch = new Stopwatch(); -// private int _disposed; -// -// public NatsPipeliningWriteProtocolProcessor(ISocketConnection socketConnection, CommandWriter commandWriter, NatsOpts opts, ConnectionStatsCounter counter) -// { -// _cts = new CancellationTokenSource(); -// _counter = counter; -// _inFlightCommands = commandWriter.InFlightCommands; -// _opts = opts; -// _pipeReader = commandWriter.PipeReader; -// _queuedCommandReader = commandWriter.QueuedCommandsReader; -// _socketConnection = socketConnection; -// WriteLoop = Task.Run(WriteLoopAsync); -// } -// -// public Task WriteLoop { get; } -// -// public async ValueTask DisposeAsync() -// { -// if (Interlocked.Increment(ref _disposed) == 1) -// { -// #if NET6_0 -// _cts.Cancel(); -// #else -// await _cts.CancelAsync().ConfigureAwait(false); -// #endif -// try -// { -// await WriteLoop.ConfigureAwait(false); // wait to drain writer -// } -// catch -// { -// // ignore -// } -// } -// } -// -// private async Task WriteLoopAsync() -// { -// var logger = _opts.LoggerFactory.CreateLogger(); -// var isEnabledTraceLogging = logger.IsEnabled(LogLevel.Trace); -// var cancellationToken = _cts.Token; -// var pending = 0; -// var trimming = 0; -// var examinedOffset = 0; -// -// // memory segment used to consolidate multiple small memory chunks -// // should <= (minimumSegmentSize * 0.5) in CommandWriter -// // 8520 should fit into 6 packets on 1500 MTU TLS connection or 1 packet on 9000 MTU TLS connection -// // assuming 40 bytes TCP overhead + 40 bytes TLS overhead per packet -// var consolidateMem = new Memory(new byte[8520]); -// -// // add up in flight command sum -// var inFlightSum = 0; -// foreach (var command in _inFlightCommands) -// { -// inFlightSum += command.Size; -// } -// -// try -// { -// while (true) -// { -// var result = await _pipeReader.ReadAsync(cancellationToken).ConfigureAwait(false); -// if (result.IsCanceled) -// { -// break; -// } -// -// var consumedPos = result.Buffer.Start; -// var examinedPos = result.Buffer.Start; -// try -// { -// var buffer = result.Buffer.Slice(examinedOffset); -// while (inFlightSum < result.Buffer.Length) -// { -// QueuedCommand queuedCommand; -// while (!_queuedCommandReader.TryRead(out queuedCommand)) -// { -// await _queuedCommandReader.WaitToReadAsync(cancellationToken).ConfigureAwait(false); -// } -// -// _inFlightCommands.Enqueue(queuedCommand); -// inFlightSum += queuedCommand.Size; -// } -// -// while (buffer.Length > 0) -// { -// if (pending == 0) -// { -// var peek = _inFlightCommands.Peek(); -// pending = peek.Size; -// trimming = peek.Trim; -// } -// -// if (trimming > 0) -// { -// var trimmed = Math.Min(trimming, (int)buffer.Length); -// consumedPos = buffer.GetPosition(trimmed); -// examinedPos = buffer.GetPosition(trimmed); -// examinedOffset = 0; -// buffer = buffer.Slice(trimmed); -// pending -= trimmed; -// trimming -= trimmed; -// if (pending == 0) -// { -// // the entire command was trimmed (canceled) -// inFlightSum -= _inFlightCommands.Dequeue().Size; -// } -// -// continue; -// } -// -// var sendMem = buffer.First; -// var maxSize = 0; -// var maxSizeCap = Math.Max(sendMem.Length, consolidateMem.Length); -// var doTrim = false; -// foreach (var command in _inFlightCommands) -// { -// if (maxSize == 0) -// { -// // first command; set to pending -// maxSize = pending; -// continue; -// } -// -// if (maxSize > maxSizeCap) -// { -// // over cap -// break; -// } -// -// if (command.Trim > 0) -// { -// // will have to trim -// doTrim = true; -// break; -// } -// -// maxSize += command.Size; -// } -// -// if (sendMem.Length > maxSize) -// { -// sendMem = sendMem[..maxSize]; -// } -// -// var bufferIter = buffer; -// if (doTrim || (bufferIter.Length > sendMem.Length && sendMem.Length < consolidateMem.Length)) -// { -// var memIter = consolidateMem; -// var trimmedSize = 0; -// foreach (var command in _inFlightCommands) -// { -// if (bufferIter.Length == 0 || memIter.Length == 0) -// { -// break; -// } -// -// int write; -// if (trimmedSize == 0) -// { -// // first command, only write pending data -// write = pending; -// } -// else if (command.Trim == 0) -// { -// write = command.Size; -// } -// else -// { -// if (bufferIter.Length < command.Trim + 1) -// { -// // not enough bytes to start writing the next command -// break; -// } -// -// bufferIter = bufferIter.Slice(command.Trim); -// write = command.Size - command.Trim; -// if (write == 0) -// { -// // the entire command was trimmed (canceled) -// continue; -// } -// } -// -// write = Math.Min(memIter.Length, write); -// write = Math.Min((int)bufferIter.Length, write); -// bufferIter.Slice(0, write).CopyTo(memIter.Span); -// memIter = memIter[write..]; -// bufferIter = bufferIter.Slice(write); -// trimmedSize += write; -// } -// -// sendMem = consolidateMem[..trimmedSize]; -// } -// -// // perform send -// _stopwatch.Restart(); -// var sent = await _socketConnection.SendAsync(sendMem).ConfigureAwait(false); -// _stopwatch.Stop(); -// Interlocked.Add(ref _counter.SentBytes, sent); -// if (isEnabledTraceLogging) -// { -// logger.LogTrace("Socket.SendAsync. Size: {0} Elapsed: {1}ms", sent, _stopwatch.Elapsed.TotalMilliseconds); -// } -// -// var consumed = 0; -// var sentAndTrimmed = sent; -// while (consumed < sentAndTrimmed) -// { -// if (pending == 0) -// { -// var peek = _inFlightCommands.Peek(); -// pending = peek.Size - peek.Trim; -// consumed += peek.Trim; -// sentAndTrimmed += peek.Trim; -// -// if (pending == 0) -// { -// // the entire command was trimmed (canceled) -// inFlightSum -= _inFlightCommands.Dequeue().Size; -// continue; -// } -// } -// -// if (pending <= sentAndTrimmed - consumed) -// { -// // the entire command was sent -// inFlightSum -= _inFlightCommands.Dequeue().Size; -// Interlocked.Add(ref _counter.PendingMessages, -1); -// Interlocked.Add(ref _counter.SentMessages, 1); -// -// // mark the bytes as consumed, and reset pending -// consumed += pending; -// pending = 0; -// } -// else -// { -// // the entire command was not sent; decrement pending by -// // the number of bytes from the command that was sent -// pending += consumed - sentAndTrimmed; -// break; -// } -// } -// -// if (consumed > 0) -// { -// // mark fully sent commands as consumed -// consumedPos = buffer.GetPosition(consumed); -// examinedOffset = sentAndTrimmed - consumed; -// } -// else -// { -// // no commands were consumed -// examinedOffset += sentAndTrimmed; -// } -// -// // lop off sent bytes for next iteration -// examinedPos = buffer.GetPosition(sentAndTrimmed); -// buffer = buffer.Slice(sentAndTrimmed); -// } -// } -// finally -// { -// _pipeReader.AdvanceTo(consumedPos, examinedPos); -// } -// -// if (result.IsCompleted) -// { -// break; -// } -// } -// } -// catch (OperationCanceledException) -// { -// // ignore, intentionally disposed -// } -// catch (SocketClosedException) -// { -// // ignore, will be handled in read loop -// } -// catch (Exception ex) -// { -// logger.LogError(ex, "Unexpected error occured in write loop"); -// throw; -// } -// -// logger.LogDebug("WriteLoop finished."); -// } -// } diff --git a/src/NATS.Client.Core/Internal/ObjectPool.cs b/src/NATS.Client.Core/Internal/ObjectPool.cs index 0658294f2..9bb5fa4df 100644 --- a/src/NATS.Client.Core/Internal/ObjectPool.cs +++ b/src/NATS.Client.Core/Internal/ObjectPool.cs @@ -12,7 +12,11 @@ internal sealed class ObjectPool { private static int typeId = -1; // Increment by IdentityGenerator +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _gate = new(); +#else private readonly object _gate = new object(); +#endif private readonly int _poolLimit; private object[] _poolNodes = new object[4]; // ObjectPool[] diff --git a/src/NATS.Client.Core/Internal/ReplyTask.cs b/src/NATS.Client.Core/Internal/ReplyTask.cs index 83110864d..260857e05 100644 --- a/src/NATS.Client.Core/Internal/ReplyTask.cs +++ b/src/NATS.Client.Core/Internal/ReplyTask.cs @@ -7,7 +7,11 @@ namespace NATS.Client.Core.Internal; internal sealed class ReplyTask : ReplyTaskBase, IDisposable { +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _gate; +#else private readonly object _gate; +#endif private readonly ReplyTaskFactory _factory; private readonly long _id; private readonly NatsConnection _connection; @@ -25,7 +29,11 @@ public ReplyTask(ReplyTaskFactory factory, long id, string subject, NatsConnecti _deserializer = deserializer; _requestTimeout = requestTimeout; _tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); +#if NET9_0_OR_GREATER + _gate = new System.Threading.Lock(); +#else _gate = new object(); +#endif } public string Subject { get; } diff --git a/src/NATS.Client.Core/Internal/SslClientAuthenticationOptionsExtensions.cs b/src/NATS.Client.Core/Internal/SslClientAuthenticationOptionsExtensions.cs index fce04db4a..5a394c0e2 100644 --- a/src/NATS.Client.Core/Internal/SslClientAuthenticationOptionsExtensions.cs +++ b/src/NATS.Client.Core/Internal/SslClientAuthenticationOptionsExtensions.cs @@ -40,7 +40,11 @@ public static SslClientAuthenticationOptions LoadClientCertFromPem(this SslClien if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows)) { var ephemeral = leafCert; +#if NET10_0_OR_GREATER + leafCert = X509CertificateLoader.LoadPkcs12(leafCert.Export(X509ContentType.Pfx), null); +#else leafCert = new X509Certificate2(leafCert.Export(X509ContentType.Pfx)); +#endif ephemeral.Dispose(); } @@ -53,6 +57,18 @@ public static SslClientAuthenticationOptions LoadClientCertFromPfxFile(this SslC X509Certificate2 leafCert; var intermediateCerts = new X509Certificate2Collection(); +#if NET10_0_OR_GREATER + if (password != null) + { + leafCert = X509CertificateLoader.LoadPkcs12FromFile(certBundleFile, password); + intermediateCerts = X509CertificateLoader.LoadPkcs12CollectionFromFile(certBundleFile, password); + } + else + { + leafCert = X509CertificateLoader.LoadPkcs12FromFile(certBundleFile, null); + intermediateCerts = X509CertificateLoader.LoadPkcs12CollectionFromFile(certBundleFile, null); + } +#else if (password != null) { leafCert = new X509Certificate2(certBundleFile, password); @@ -63,6 +79,7 @@ public static SslClientAuthenticationOptions LoadClientCertFromPfxFile(this SslC leafCert = new X509Certificate2(certBundleFile); intermediateCerts.Import(certBundleFile); } +#endif // Linux does not include the leaf by default, but Windows does // compare leaf to first intermediate just to be sure to catch all platform differences diff --git a/src/NATS.Client.Core/Internal/SubscriptionManager.cs b/src/NATS.Client.Core/Internal/SubscriptionManager.cs index 928d80acc..4504e29ea 100644 --- a/src/NATS.Client.Core/Internal/SubscriptionManager.cs +++ b/src/NATS.Client.Core/Internal/SubscriptionManager.cs @@ -15,7 +15,11 @@ internal sealed class SubscriptionManager : INatsSubscriptionManager, IAsyncDisp private readonly ILogger _logger; private readonly bool _trace; private readonly bool _debug; +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _gate = new(); +#else private readonly object _gate = new(); +#endif private readonly NatsConnection _connection; private readonly string _inboxPrefix; private readonly ConcurrentDictionary _bySid = new(); @@ -96,18 +100,17 @@ public ValueTask PublishToClientHandlersAsync(string subject, string? replyTo, i } int? orphanSid = null; + NatsSubBase? targetSub = null; lock (_gate) { if (_bySid.TryGetValue(sid, out var sidMetadata)) { - if (sidMetadata.WeakReference.TryGetTarget(out var sub)) + if (sidMetadata.WeakReference.TryGetTarget(out targetSub)) { if (_trace) { _logger.LogTrace(NatsLogEvents.Subscription, "Found subscription handler for {Subject}/{Sid}", subject, sid); } - - return sub.ReceiveAsync(subject, replyTo, headersBuffer, payloadBuffer); } else { @@ -126,6 +129,11 @@ public ValueTask PublishToClientHandlersAsync(string subject, string? replyTo, i } } + if (targetSub != null) + { + return targetSub.ReceiveAsync(subject, replyTo, headersBuffer, payloadBuffer); + } + if (orphanSid != null) { try diff --git a/src/NATS.Client.Core/Internal/Telemetry.cs b/src/NATS.Client.Core/Internal/Telemetry.cs index 575279044..87fd0bff6 100644 --- a/src/NATS.Client.Core/Internal/Telemetry.cs +++ b/src/NATS.Client.Core/Internal/Telemetry.cs @@ -104,11 +104,7 @@ public static void AddTraceContextHeaders(Activity? activity, ref NatsHeaders? h return; } - // There are cases where headers reused publicly (e.g. JetStream publish retry) - // there may also be cases where application can reuse the same header - // in which case we should still be able to overwrite headers with telemetry fields - // even though headers would be set to readonly before being passed down in publish methods. - headers.SetOverrideReadOnly(fieldName, fieldValue); + headers[fieldName] = fieldValue; }); } @@ -286,11 +282,7 @@ private static bool TryParseTraceContext(NatsHeaders headers, out ActivityContex }, out var traceParent, out var traceState); -#if NETSTANDARD2_0_OR_GREATER || NET7_0_OR_GREATER return ActivityContext.TryParse(traceParent, traceState, isRemote: true, out context); -#else - return ActivityContext.TryParse(traceParent, traceState, out context); -#endif } public class Constants diff --git a/src/NATS.Client.Core/Internal/netstandard.cs b/src/NATS.Client.Core/Internal/netstandard.cs index 24694abca..925fbf067 100644 --- a/src/NATS.Client.Core/Internal/netstandard.cs +++ b/src/NATS.Client.Core/Internal/netstandard.cs @@ -11,9 +11,9 @@ #pragma warning disable SA1405 // Enable init only setters -#if NET5_0_OR_GREATER +#if !NETSTANDARD [assembly: System.Runtime.CompilerServices.TypeForwardedTo(typeof(System.Runtime.CompilerServices.IsExternalInit))] -#elif NETSTANDARD +#else namespace System.Runtime.CompilerServices { @@ -160,6 +160,7 @@ internal class PeriodicTimer : IDisposable private readonly Timer _timer; private readonly TimeSpan _period; private TaskCompletionSource _tcs; + private CancellationTokenRegistration _ctr; private bool _disposed; public PeriodicTimer(TimeSpan period) @@ -176,17 +177,29 @@ public Task WaitForNextTickAsync(CancellationToken cancellationToken = def _timer.Change(_period, Timeout.InfiniteTimeSpan); +#pragma warning disable VSTHRD103 + _ctr.Dispose(); +#pragma warning restore VSTHRD103 + if (cancellationToken.IsCancellationRequested) return Task.FromCanceled(cancellationToken); - cancellationToken.Register(() => _tcs.TrySetCanceled(cancellationToken)); + var tcs = _tcs; + _ctr = cancellationToken.Register( + static state => + { + var source = (TaskCompletionSource)state!; + source.TrySetCanceled(); + }, + tcs); - return _tcs.Task; + return tcs.Task; } public void Dispose() { _disposed = true; + _ctr.Dispose(); _timer.Dispose(); _tcs.TrySetResult(false); // Signal no more ticks will occur } @@ -200,7 +213,7 @@ private void Callback(object state) internal static class ReadOnlySequenceExtensions { - // Adapted from .NET 6.0 implementation + // Adapted from .NET runtime implementation internal static long GetOffset(this in ReadOnlySequence sequence, SequencePosition position) { var positionSequenceObject = position.GetObject(); @@ -256,12 +269,29 @@ internal static long GetOffset(this in ReadOnlySequence sequence, Sequence internal static class EncodingExtensionsCommon { internal static string GetString(this Encoding encoding, in ReadOnlySequence buffer) - => encoding.GetString(buffer.ToArray()); + { + if (buffer.IsSingleSegment) + { +#if NETSTANDARD2_0 + return encoding.GetString(buffer.First.Span); +#else + return encoding.GetString(buffer.FirstSpan); +#endif + } + + return encoding.GetString(buffer.ToArray()); + } internal static void GetBytes(this Encoding encoding, string chars, IBufferWriter bw) { - var buffer = encoding.GetBytes(chars); - bw.Write(buffer); + var byteCount = encoding.GetByteCount(chars); + var span = bw.GetSpan(byteCount); +#if NETSTANDARD2_0 + encoding.GetBytes(chars, span); +#else + encoding.GetBytes(chars.AsSpan(), span); +#endif + bw.Advance(byteCount); } } @@ -304,16 +334,31 @@ namespace NATS.Client.Core.Internal.NetStandardExtensions internal static class EncodingExtensions { - internal static int GetBytes(this Encoding encoding, string chars, Span bytes) + internal static unsafe int GetBytes(this Encoding encoding, string chars, Span bytes) { - var buffer = encoding.GetBytes(chars); - buffer.AsSpan().CopyTo(bytes); - return buffer.Length; + if (chars.Length == 0) + { + return 0; + } + + fixed (char* charPtr = chars) + { + fixed (byte* bytePtr = bytes) + { + return encoding.GetBytes(charPtr, chars.Length, bytePtr, bytes.Length); + } + } } - internal static string GetString(this Encoding encoding, in ReadOnlySpan buffer) + internal static unsafe string GetString(this Encoding encoding, in ReadOnlySpan buffer) { - return encoding.GetString(buffer.ToArray()); + if (buffer.IsEmpty) + return string.Empty; + + fixed (byte* ptr = buffer) + { + return encoding.GetString(ptr, buffer.Length); + } } } diff --git a/src/NATS.Client.Core/NATS.Client.Core.csproj b/src/NATS.Client.Core/NATS.Client.Core.csproj index 669c60970..b83da55fc 100644 --- a/src/NATS.Client.Core/NATS.Client.Core.csproj +++ b/src/NATS.Client.Core/NATS.Client.Core.csproj @@ -1,73 +1,73 @@ - - - - - pubsub;messaging;cloud-native;distributed;async;real-time;CNCF;nats - Core client for NATS, the high-performance cloud-native messaging system. Provides async pub/sub, request-reply, and queue groups for .NET applications. - - - - - - - - - all - runtime; build; native; contentfiles; analyzers; buildtransitive - - - - - - - - - all - runtime; build; native; contentfiles; analyzers - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + pubsub;messaging;cloud-native;distributed;async;real-time;CNCF;nats + Core client for NATS, the high-performance cloud-native messaging system. Provides async pub/sub, request-reply, and queue groups for .NET applications. + + + + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + + all + runtime; build; native; contentfiles; analyzers + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/NATS.Client.Core/NatsConnection.Publish.cs b/src/NATS.Client.Core/NatsConnection.Publish.cs index 68eb59e3d..8dd86e629 100644 --- a/src/NATS.Client.Core/NatsConnection.Publish.cs +++ b/src/NATS.Client.Core/NatsConnection.Publish.cs @@ -19,7 +19,6 @@ public ValueTask PublishAsync(string subject, NatsHeaders? headers = default, st Telemetry.AddTraceContextHeaders(activity, ref headers); try { - headers?.SetReadOnly(); return ConnectionState != NatsConnectionState.Open ? ConnectAndPublishAsync(subject, default, headers, replyTo, NatsRawSerializer.Default, cancellationToken) : CommandWriter.PublishAsync(subject, default, headers, replyTo, NatsRawSerializer.Default, cancellationToken); @@ -31,7 +30,6 @@ public ValueTask PublishAsync(string subject, NatsHeaders? headers = default, st } } - headers?.SetReadOnly(); return ConnectionState != NatsConnectionState.Open ? ConnectAndPublishAsync(subject, default, headers, replyTo, NatsRawSerializer.Default, cancellationToken) : CommandWriter.PublishAsync(subject, default, headers, replyTo, NatsRawSerializer.Default, cancellationToken); @@ -53,7 +51,6 @@ public ValueTask PublishAsync(string subject, T? data, NatsHeaders? headers = try { serializer ??= Opts.SerializerRegistry.GetSerializer(); - headers?.SetReadOnly(); return ConnectionState != NatsConnectionState.Open ? ConnectAndPublishAsync(subject, data, headers, replyTo, serializer, cancellationToken) : CommandWriter.PublishAsync(subject, data, headers, replyTo, serializer, cancellationToken); @@ -66,7 +63,6 @@ public ValueTask PublishAsync(string subject, T? data, NatsHeaders? headers = } serializer ??= Opts.SerializerRegistry.GetSerializer(); - headers?.SetReadOnly(); return ConnectionState != NatsConnectionState.Open ? ConnectAndPublishAsync(subject, data, headers, replyTo, serializer, cancellationToken) : CommandWriter.PublishAsync(subject, data, headers, replyTo, serializer, cancellationToken); diff --git a/src/NATS.Client.Core/NatsConnection.RequestReply.cs b/src/NATS.Client.Core/NatsConnection.RequestReply.cs index fd70eb082..0dde7c32d 100644 --- a/src/NATS.Client.Core/NatsConnection.RequestReply.cs +++ b/src/NATS.Client.Core/NatsConnection.RequestReply.cs @@ -149,7 +149,7 @@ static void WriteBuffer(Span buffer, (string prefix, int pfxLen) state) var totalLength = prefix.Length + (int)Nuid.NuidLength + separatorLength; var totalPrefixLength = prefix.Length + separatorLength; -#if NET6_0_OR_GREATER || NETSTANDARD2_1 +#if !NETSTANDARD2_0 return string.Create(totalLength, (prefix, totalPrefixLength), (buf, state) => WriteBuffer(buf, state)); #else Span buffer = new char[totalLength]; diff --git a/src/NATS.Client.Core/NatsConnection.cs b/src/NATS.Client.Core/NatsConnection.cs index d97564d24..5e72efbcd 100644 --- a/src/NATS.Client.Core/NatsConnection.cs +++ b/src/NATS.Client.Core/NatsConnection.cs @@ -37,7 +37,11 @@ public partial class NatsConnection : INatsConnection #pragma warning disable SA1401 internal readonly ConnectionStatsCounter Counter; // allow to call from external sources #pragma warning restore SA1401 +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _gate = new(); +#else private readonly object _gate = new object(); +#endif private readonly ILogger _logger; private readonly ObjectPool _pool; private readonly CancellationTokenSource _disposedCts; diff --git a/src/NATS.Client.Core/NatsHeaders.cs b/src/NATS.Client.Core/NatsHeaders.cs index 15d09be45..53324f541 100644 --- a/src/NATS.Client.Core/NatsHeaders.cs +++ b/src/NATS.Client.Core/NatsHeaders.cs @@ -10,12 +10,18 @@ namespace NATS.Client.Core; /// /// Represents a wrapper for RequestHeaders and ResponseHeaders. /// +/// +/// Not thread-safe. Do not share a single instance across concurrent +/// publishes: the writer may read and context-aware serializers may mutate the dictionary while +/// it is being written to the wire. Construct a fresh instance per publish, or wait for one +/// publish to complete before reusing the same instance. +/// [SuppressMessage("StyleCop.CSharp.OrderingRules", "SA1201:Elements should appear in the correct order", Justification = "Keep class format as is for reference")] [SuppressMessage("StyleCop.CSharp.LayoutRules", "SA1504:All accessors should be single-line or multi-line", Justification = "Keep class format as is for reference")] [SuppressMessage("StyleCop.CSharp.LayoutRules", "SA1516:Elements should be separated by blank line", Justification = "Keep class format as is for reference")] [SuppressMessage("StyleCop.CSharp.LayoutRules", "SA1513:Closing brace should be followed by blank line", Justification = "Keep class format as is for reference")] [SuppressMessage("StyleCop.CSharp.OrderingRules", "SA1214:Readonly fields should appear before non-readonly fields", Justification = "Keep class format as is for reference")] -public class NatsHeaders : IDictionary +public class NatsHeaders : INatsHeaders { public enum Messages { @@ -70,8 +76,6 @@ public enum Messages private static readonly IEnumerator> EmptyIEnumeratorType = default(Enumerator); private static readonly IEnumerator EmptyIEnumerator = default(Enumerator); - private int _readonly = 0; - public int Version => 1; public int Code { get; internal set; } @@ -144,7 +148,6 @@ public StringValues this[string key] { throw new ArgumentNullException(nameof(key)); } - ThrowIfReadOnly(); if (value.Count == 0) { @@ -161,11 +164,7 @@ public StringValues this[string key] StringValues IDictionary.this[string key] { get { return this[key]; } - set - { - ThrowIfReadOnly(); - this[key] = value; - } + set { this[key] = value; } } /// @@ -175,10 +174,9 @@ StringValues IDictionary.this[string key] public int Count => Store?.Count ?? 0; /// - /// Gets a value that indicates whether the is in read-only mode. + /// Gets a value that indicates whether the is read-only. Always false. /// - /// true if the is in read-only mode; otherwise, false. - public bool IsReadOnly => Volatile.Read(ref _readonly) == 1; + public bool IsReadOnly => false; /// /// Gets the collection of HTTP header names in this instance. @@ -222,7 +220,6 @@ public void Add(KeyValuePair item) { throw new ArgumentException("The key is null"); } - ThrowIfReadOnly(); EnsureStore(1); Store.Add(item.Key, item.Value); } @@ -238,7 +235,6 @@ public void Add(string key, StringValues value) { throw new ArgumentNullException(nameof(key)); } - ThrowIfReadOnly(); EnsureStore(1); Store.Add(key, value); } @@ -248,7 +244,6 @@ public void Add(string key, StringValues value) /// public void Clear() { - ThrowIfReadOnly(); Store?.Clear(); } @@ -322,7 +317,6 @@ public int GetBytesLength(Encoding? encoding = null) /// true if the specified object was removed from the collection; otherwise, false. public bool Remove(KeyValuePair item) { - ThrowIfReadOnly(); if (Store == null) { return false; @@ -342,7 +336,6 @@ public bool Remove(KeyValuePair item) /// true if the specified object was removed from the collection; otherwise, false. public bool Remove(string key) { - ThrowIfReadOnly(); if (Store == null) { return false; @@ -426,34 +419,6 @@ IEnumerator IEnumerable.GetEnumerator() return Store.GetEnumerator(); } - internal void SetReadOnly() => Interlocked.Exchange(ref _readonly, 1); - - internal void SetOverrideReadOnly(string key, StringValues value) - { - if (key == null) - { - throw new ArgumentNullException(nameof(key)); - } - - if (value.Count == 0) - { - Store?.Remove(key); - } - else - { - EnsureStore(1); - Store[key] = value; - } - } - - private void ThrowIfReadOnly() - { - if (IsReadOnly) - { - throw new InvalidOperationException("The response headers cannot be modified because the response has already started."); - } - } - /// /// Enumerates a . /// diff --git a/src/NATS.Client.Core/NatsMemoryOwner.cs b/src/NATS.Client.Core/NatsMemoryOwner.cs index d7813d6ff..6c2200966 100644 --- a/src/NATS.Client.Core/NatsMemoryOwner.cs +++ b/src/NATS.Client.Core/NatsMemoryOwner.cs @@ -189,10 +189,10 @@ public Span Span ref var r0 = ref array!.DangerousGetReferenceAt(_start); - // On .NET 6+ runtimes, we can manually create a span from the starting reference to + // On .NET 8+ runtimes, we can manually create a span from the starting reference to // skip the argument validations, which include an explicit null check, covariance check // for the array and the actual validation for the starting offset and target length. We - // only do this on .NET 6+ as we can leverage the runtime-specific array layout to get + // only do this on .NET 8+ as we can leverage the runtime-specific array layout to get // a fast access to the initial element, which makes this trick worth it. Otherwise, on // runtimes where we would need to at least access a static field to retrieve the base // byte offset within an SZ array object, we can get better performance by just using the diff --git a/src/NATS.Client.Core/NatsMsg.cs b/src/NATS.Client.Core/NatsMsg.cs index 8d7112831..3ca4c9c5e 100644 --- a/src/NATS.Client.Core/NatsMsg.cs +++ b/src/NATS.Client.Core/NatsMsg.cs @@ -364,8 +364,6 @@ public static NatsMsg Build( } } - headers?.SetReadOnly(); - var size = subject.Length + (replyTo?.Length ?? 0) + (headersBuffer?.Length ?? 0) @@ -401,7 +399,7 @@ public static NatsMsg Build( { try { - data = serializer.Deserialize(payloadBuffer); + data = serializer.Deserialize(payloadBuffer, new NatsMsgContext(subject, replyTo, headers)); } catch (Exception e) { @@ -519,7 +517,7 @@ public NatsMsg Msg if (Serializer != null && Data != null) { var bufferWriter = new NatsPooledBufferWriter(SerializationBufferSize); - Serializer.Serialize(bufferWriter, Data); + Serializer.Serialize(bufferWriter, Data, new NatsMsgContext(Subject, ReplyTo, Headers)); size = bufferWriter.WrittenMemory.Length; } diff --git a/src/NATS.Client.Core/NatsSerialize.cs b/src/NATS.Client.Core/NatsSerialize.cs index 3f889935b..6234508ac 100644 --- a/src/NATS.Client.Core/NatsSerialize.cs +++ b/src/NATS.Client.Core/NatsSerialize.cs @@ -37,7 +37,7 @@ public class NatsDefaultSerializerRegistry : INatsSerializerRegistry /// TimeSpan, bool, byte, decimal, double, float, /// int, long, sbyte, short, uint and ulong. /// -public class NatsUtf8PrimitivesSerializer : INatsSerializer +public class NatsUtf8PrimitivesSerializer : INatsSerializer, INatsSerializerWithContext { public static readonly NatsUtf8PrimitivesSerializer Default = new(); @@ -54,11 +54,58 @@ public class NatsUtf8PrimitivesSerializer : INatsSerializer /// public void Serialize(IBufferWriter bufferWriter, T value) + { + if (TrySerializePrimitive(bufferWriter, value)) + return; + + if (_next == null) + throw new NatsException($"Can't serialize {typeof(T)}"); + + _next.Serialize(bufferWriter, value); + } + + /// + public void Serialize(IBufferWriter bufferWriter, T value, in NatsMsgContext context) + { + if (TrySerializePrimitive(bufferWriter, value)) + return; + + if (_next == null) + throw new NatsException($"Can't serialize {typeof(T)}"); + + _next.Serialize(bufferWriter, value, in context); + } + + /// + public T? Deserialize(in ReadOnlySequence buffer) + { + if (TryDeserializePrimitive(buffer, out var result)) + return result; + + if (_next == null) + throw new NatsException($"Can't deserialize {typeof(T)}"); + + return _next.Deserialize(buffer); + } + + /// + public T? Deserialize(in ReadOnlySequence buffer, in NatsMsgContext context) + { + if (TryDeserializePrimitive(buffer, out var result)) + return result; + + if (_next == null) + throw new NatsException($"Can't deserialize {typeof(T)}"); + + return _next.Deserialize(buffer, in context); + } + + private static bool TrySerializePrimitive(IBufferWriter bufferWriter, T value) { if (value is string str) { Encoding.UTF8.GetBytes(str, bufferWriter); - return; + return true; } var span = bufferWriter.GetSpan(128); @@ -76,7 +123,7 @@ public void Serialize(IBufferWriter bufferWriter, T value) throw new NatsException($"Can't serialize {typeof(T)}, format error"); } - return; + return true; } } @@ -105,7 +152,7 @@ public void Serialize(IBufferWriter bufferWriter, T value) throw new NatsException($"Can't serialize {typeof(T)}, format error"); } - return; + return true; } } @@ -122,7 +169,7 @@ public void Serialize(IBufferWriter bufferWriter, T value) throw new NatsException($"Can't serialize {typeof(T)}, format error"); } - return; + return true; } } @@ -139,7 +186,7 @@ public void Serialize(IBufferWriter bufferWriter, T value) throw new NatsException($"Can't serialize {typeof(T)}, format error"); } - return; + return true; } } @@ -156,7 +203,7 @@ public void Serialize(IBufferWriter bufferWriter, T value) throw new NatsException($"Can't serialize {typeof(T)}, format error"); } - return; + return true; } } @@ -173,7 +220,7 @@ public void Serialize(IBufferWriter bufferWriter, T value) throw new NatsException($"Can't serialize {typeof(T)}, format error"); } - return; + return true; } } @@ -190,7 +237,7 @@ public void Serialize(IBufferWriter bufferWriter, T value) throw new NatsException($"Can't serialize {typeof(T)}, format error"); } - return; + return true; } } @@ -207,7 +254,7 @@ public void Serialize(IBufferWriter bufferWriter, T value) throw new NatsException($"Can't serialize {typeof(T)}, format error"); } - return; + return true; } } @@ -224,7 +271,7 @@ public void Serialize(IBufferWriter bufferWriter, T value) throw new NatsException($"Can't serialize {typeof(T)}, format error"); } - return; + return true; } } @@ -241,7 +288,7 @@ public void Serialize(IBufferWriter bufferWriter, T value) throw new NatsException($"Can't serialize {typeof(T)}, format error"); } - return; + return true; } } @@ -258,7 +305,7 @@ public void Serialize(IBufferWriter bufferWriter, T value) throw new NatsException($"Can't serialize {typeof(T)}, format error"); } - return; + return true; } } @@ -275,7 +322,7 @@ public void Serialize(IBufferWriter bufferWriter, T value) throw new NatsException($"Can't serialize {typeof(T)}, format error"); } - return; + return true; } } @@ -292,7 +339,7 @@ public void Serialize(IBufferWriter bufferWriter, T value) throw new NatsException($"Can't serialize {typeof(T)}, format error"); } - return; + return true; } } @@ -309,7 +356,7 @@ public void Serialize(IBufferWriter bufferWriter, T value) throw new NatsException($"Can't serialize {typeof(T)}, format error"); } - return; + return true; } } @@ -326,26 +373,25 @@ public void Serialize(IBufferWriter bufferWriter, T value) throw new NatsException($"Can't serialize {typeof(T)}, format error"); } - return; + return true; } } - if (_next == null) - { - throw new NatsException($"Can't serialize {typeof(T)}"); - } - - _next.Serialize(bufferWriter, value); + return false; } - /// - public T? Deserialize(in ReadOnlySequence buffer) + private static bool TryDeserializePrimitive(in ReadOnlySequence buffer, out T? result) { if (typeof(T) == typeof(string)) { if (buffer.Length == 0) - return default; - return (T)(object)Encoding.UTF8.GetString(buffer); + { + result = default; + return true; + } + + result = (T)(object)Encoding.UTF8.GetString(buffer); + return true; } var span = buffer.IsSingleSegment ? buffer.GetFirstSpan() : buffer.ToArray(); @@ -353,11 +399,15 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (typeof(T) == typeof(DateTime) || typeof(T) == typeof(DateTime?)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } if (Utf8Parser.TryParse(span, out DateTime value, out _, 'O')) { - return (T)(object)value; + result = (T)(object)value; + return true; } throw new NatsException($"Can't deserialize {typeof(T)}. Parsing error"); @@ -366,11 +416,15 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (typeof(T) == typeof(DateTimeOffset) || typeof(T) == typeof(DateTimeOffset?)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } if (Utf8Parser.TryParse(span, out DateTimeOffset value, out _, 'O')) { - return (T)(object)value; + result = (T)(object)value; + return true; } throw new NatsException($"Can't deserialize {typeof(T)}. Parsing error"); @@ -379,11 +433,15 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (typeof(T) == typeof(Guid) || typeof(T) == typeof(Guid?)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } if (Utf8Parser.TryParse(span, out Guid value, out _)) { - return (T)(object)value; + result = (T)(object)value; + return true; } throw new NatsException($"Can't deserialize {typeof(T)}. Parsing error"); @@ -392,11 +450,15 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (typeof(T) == typeof(TimeSpan) || typeof(T) == typeof(TimeSpan?)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } if (Utf8Parser.TryParse(span, out TimeSpan value, out _)) { - return (T)(object)value; + result = (T)(object)value; + return true; } throw new NatsException($"Can't deserialize {typeof(T)}. Parsing error"); @@ -405,11 +467,15 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (typeof(T) == typeof(bool) || typeof(T) == typeof(bool?)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } if (Utf8Parser.TryParse(span, out bool value, out _)) { - return (T)(object)value; + result = (T)(object)value; + return true; } throw new NatsException($"Can't deserialize {typeof(T)}. Parsing error"); @@ -418,11 +484,15 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (typeof(T) == typeof(byte) || typeof(T) == typeof(byte?)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } if (Utf8Parser.TryParse(span, out byte value, out _)) { - return (T)(object)value; + result = (T)(object)value; + return true; } throw new NatsException($"Can't deserialize {typeof(T)}. Parsing error"); @@ -431,11 +501,15 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (typeof(T) == typeof(decimal) || typeof(T) == typeof(decimal?)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } if (Utf8Parser.TryParse(span, out decimal value, out _)) { - return (T)(object)value; + result = (T)(object)value; + return true; } throw new NatsException($"Can't deserialize {typeof(T)}. Parsing error"); @@ -444,11 +518,15 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (typeof(T) == typeof(double) || typeof(T) == typeof(double?)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } if (Utf8Parser.TryParse(span, out double value, out _)) { - return (T)(object)value; + result = (T)(object)value; + return true; } throw new NatsException($"Can't deserialize {typeof(T)}. Parsing error"); @@ -457,11 +535,15 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (typeof(T) == typeof(float) || typeof(T) == typeof(float?)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } if (Utf8Parser.TryParse(span, out float value, out _)) { - return (T)(object)value; + result = (T)(object)value; + return true; } throw new NatsException($"Can't deserialize {typeof(T)}. Parsing error"); @@ -470,11 +552,15 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (typeof(T) == typeof(int) || typeof(T) == typeof(int?)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } if (Utf8Parser.TryParse(span, out int value, out _)) { - return (T)(object)value; + result = (T)(object)value; + return true; } throw new NatsException($"Can't deserialize {typeof(T)}. Parsing error"); @@ -483,11 +569,15 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (typeof(T) == typeof(long) || typeof(T) == typeof(long?)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } if (Utf8Parser.TryParse(span, out long value, out _)) { - return (T)(object)value; + result = (T)(object)value; + return true; } throw new NatsException($"Can't deserialize {typeof(T)}. Parsing error"); @@ -496,11 +586,15 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (typeof(T) == typeof(sbyte) || typeof(T) == typeof(sbyte?)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } if (Utf8Parser.TryParse(span, out sbyte value, out _)) { - return (T)(object)value; + result = (T)(object)value; + return true; } throw new NatsException($"Can't deserialize {typeof(T)}. Parsing error"); @@ -509,11 +603,15 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (typeof(T) == typeof(short) || typeof(T) == typeof(short?)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } if (Utf8Parser.TryParse(span, out short value, out _)) { - return (T)(object)value; + result = (T)(object)value; + return true; } throw new NatsException($"Can't deserialize {typeof(T)}. Parsing error"); @@ -522,11 +620,15 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (typeof(T) == typeof(uint) || typeof(T) == typeof(uint?)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } if (Utf8Parser.TryParse(span, out uint value, out _)) { - return (T)(object)value; + result = (T)(object)value; + return true; } throw new NatsException($"Can't deserialize {typeof(T)}. Parsing error"); @@ -535,29 +637,29 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (typeof(T) == typeof(ulong) || typeof(T) == typeof(ulong?)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } if (Utf8Parser.TryParse(span, out ulong value, out _)) { - return (T)(object)value; + result = (T)(object)value; + return true; } throw new NatsException($"Can't deserialize {typeof(T)}. Parsing error"); } - if (_next == null) - { - throw new NatsException($"Can't deserialize {typeof(T)}"); - } - - return _next.Deserialize(buffer); + result = default; + return false; } } /// /// Serializer for binary data. /// -public class NatsRawSerializer : INatsSerializer +public class NatsRawSerializer : INatsSerializer, INatsSerializerWithContext { public static readonly NatsRawSerializer Default = new(NatsUtf8PrimitivesSerializer.Default); @@ -574,23 +676,70 @@ public class NatsRawSerializer : INatsSerializer /// public void Serialize(IBufferWriter bufferWriter, T value) + { + if (TrySerializeRaw(bufferWriter, value)) + return; + + if (_next == null) + throw new NatsException($"Can't serialize {typeof(T)}"); + + _next.Serialize(bufferWriter, value); + } + + /// + public void Serialize(IBufferWriter bufferWriter, T value, in NatsMsgContext context) + { + if (TrySerializeRaw(bufferWriter, value)) + return; + + if (_next == null) + throw new NatsException($"Can't serialize {typeof(T)}"); + + _next.Serialize(bufferWriter, value, in context); + } + + /// + public T? Deserialize(in ReadOnlySequence buffer) + { + if (TryDeserializeRaw(buffer, out var result)) + return result; + + if (_next == null) + throw new NatsException($"Can't deserialize {typeof(T)}"); + + return _next.Deserialize(buffer); + } + + /// + public T? Deserialize(in ReadOnlySequence buffer, in NatsMsgContext context) + { + if (TryDeserializeRaw(buffer, out var result)) + return result; + + if (_next == null) + throw new NatsException($"Can't deserialize {typeof(T)}"); + + return _next.Deserialize(buffer, in context); + } + + private static bool TrySerializeRaw(IBufferWriter bufferWriter, T value) { if (value is byte[] bytes) { bufferWriter.Write(bytes); - return; + return true; } if (value is Memory memory) { bufferWriter.Write(memory.Span); - return; + return true; } if (value is ReadOnlyMemory readOnlyMemory) { bufferWriter.Write(readOnlyMemory.Span); - return; + return true; } if (value is ReadOnlySequence readOnlySequence) @@ -607,7 +756,7 @@ public void Serialize(IBufferWriter bufferWriter, T value) } } - return; + return true; } if (value is IMemoryOwner memoryOwner) @@ -621,64 +770,79 @@ public void Serialize(IBufferWriter bufferWriter, T value) bufferWriter.Advance(length); - return; + return true; } } - if (_next == null) - { - throw new NatsException($"Can't serialize {typeof(T)}"); - } - - _next.Serialize(bufferWriter, value); + return false; } - /// - public T? Deserialize(in ReadOnlySequence buffer) + private static bool TryDeserializeRaw(in ReadOnlySequence buffer, out T? result) { if (typeof(T) == typeof(byte[])) { if (buffer.Length == 0) - return default; - return (T)(object)buffer.ToArray(); + { + result = default; + return true; + } + + result = (T)(object)buffer.ToArray(); + return true; } if (typeof(T) == typeof(Memory)) { if (buffer.Length == 0) - return default; - return (T)(object)new Memory(buffer.ToArray()); + { + result = default; + return true; + } + + result = (T)(object)new Memory(buffer.ToArray()); + return true; } if (typeof(T) == typeof(ReadOnlyMemory)) { if (buffer.Length == 0) - return default; - return (T)(object)new ReadOnlyMemory(buffer.ToArray()); + { + result = default; + return true; + } + + result = (T)(object)new ReadOnlyMemory(buffer.ToArray()); + return true; } if (typeof(T) == typeof(ReadOnlySequence)) { if (buffer.Length == 0) - return default; - return (T)(object)new ReadOnlySequence(buffer.ToArray()); + { + result = default; + return true; + } + + result = (T)(object)new ReadOnlySequence(buffer.ToArray()); + return true; } if (typeof(T) == typeof(IMemoryOwner) || typeof(T) == typeof(NatsMemoryOwner)) { if (buffer.Length == 0) - return default; + { + result = default; + return true; + } + var memoryOwner = NatsMemoryOwner.Allocate((int)buffer.Length); buffer.CopyTo(memoryOwner.Memory.Span); - return (T)(object)memoryOwner; + result = (T)(object)memoryOwner; + return true; } - if (_next == null) - { - throw new NatsException($"Can't deserialize {typeof(T)}"); - } - - return _next.Deserialize(buffer); + result = default; + return false; } } @@ -696,7 +860,7 @@ public sealed class NatsJsonContextSerializerRegistry : INatsSerializerRegistry /// /// Serializer with support for . /// -public sealed class NatsJsonContextSerializer : INatsSerializer +public sealed class NatsJsonContextSerializer : INatsSerializer, INatsSerializerWithContext { // ReSharper disable once StaticMemberInGenericType private static readonly JsonWriterOptions JsonWriterOpts = new() { Indented = false, SkipValidation = true }; @@ -729,6 +893,53 @@ public NatsJsonContextSerializer(JsonSerializerContext context, INatsSerializer< /// public void Serialize(IBufferWriter bufferWriter, T value) + { + if (TrySerializeJson(bufferWriter, value)) + return; + + if (_next == null) + throw new NatsException($"Can't serialize {typeof(T)}"); + + _next.Serialize(bufferWriter, value); + } + + /// + public void Serialize(IBufferWriter bufferWriter, T value, in NatsMsgContext context) + { + if (TrySerializeJson(bufferWriter, value)) + return; + + if (_next == null) + throw new NatsException($"Can't serialize {typeof(T)}"); + + _next.Serialize(bufferWriter, value, in context); + } + + /// + public T? Deserialize(in ReadOnlySequence buffer) + { + if (TryDeserializeJson(buffer, out var result)) + return result; + + if (_next == null) + throw new NatsException($"Can't deserialize {typeof(T)}"); + + return _next.Deserialize(buffer); + } + + /// + public T? Deserialize(in ReadOnlySequence buffer, in NatsMsgContext context) + { + if (TryDeserializeJson(buffer, out var result)) + return result; + + if (_next == null) + throw new NatsException($"Can't deserialize {typeof(T)}"); + + return _next.Deserialize(buffer, in context); + } + + private bool TrySerializeJson(IBufferWriter bufferWriter, T value) { foreach (var context in _contexts) { @@ -748,24 +959,19 @@ public void Serialize(IBufferWriter bufferWriter, T value) JsonSerializer.Serialize(writer, value, jsonTypeInfo); writer.Reset(NullBufferWriter.Instance); - return; + return true; } } - if (_next == null) - { - throw new NatsException($"Can't serialize {typeof(T)}"); - } - - _next.Serialize(bufferWriter, value); + return false; } - /// - public T? Deserialize(in ReadOnlySequence buffer) + private bool TryDeserializeJson(in ReadOnlySequence buffer, out T? result) { if (buffer.Length == 0) { - return default; + result = default; + return true; } foreach (var context in _contexts) @@ -773,14 +979,13 @@ public void Serialize(IBufferWriter bufferWriter, T value) if (context.GetTypeInfo(typeof(T)) is JsonTypeInfo jsonTypeInfo) { var reader = new Utf8JsonReader(buffer); // Utf8JsonReader is ref struct, no allocate. - return JsonSerializer.Deserialize(ref reader, jsonTypeInfo); + result = JsonSerializer.Deserialize(ref reader, jsonTypeInfo); + return true; } } - if (_next != null) - return _next.Deserialize(buffer); - - throw new NatsException($"Can't deserialize {typeof(T)}"); + result = default; + return false; } } diff --git a/src/NATS.Client.Core/NatsSubBase.cs b/src/NATS.Client.Core/NatsSubBase.cs index afa3feb70..dddd3c545 100644 --- a/src/NATS.Client.Core/NatsSubBase.cs +++ b/src/NATS.Client.Core/NatsSubBase.cs @@ -32,7 +32,11 @@ public abstract class NatsSubBase { private static readonly byte[] NoRespondersHeaderSequence = { (byte)' ', (byte)'5', (byte)'0', (byte)'3' }; private readonly ILogger _logger; +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _gate = new(); +#else private readonly object _gate = new(); +#endif private readonly bool _debug; private readonly INatsSubscriptionManager _manager; private readonly Timer? _timeoutTimer; diff --git a/src/NATS.Client.Hosting/NATS.Client.Hosting.csproj b/src/NATS.Client.Hosting/NATS.Client.Hosting.csproj index e36938aae..8451a08f4 100644 --- a/src/NATS.Client.Hosting/NATS.Client.Hosting.csproj +++ b/src/NATS.Client.Hosting/NATS.Client.Hosting.csproj @@ -3,14 +3,14 @@ pubsub;messaging;cloud-native;aspnetcore;hosting;nats - ASP.NET Core and Generic Host integration for the NATS .NET core client. Registers NatsConnection in the .NET Generic Host service collection. + NATS .NET dependency injection with minimal dependencies and AOT support. Registers NatsConnection on the Microsoft service collection. For ad-hoc JSON serialization, see NATS.Extensions.Microsoft.DependencyInjection. - + - + diff --git a/src/NATS.Client.JetStream/Internal/NatsJSConsume.cs b/src/NATS.Client.JetStream/Internal/NatsJSConsume.cs index 09e295c36..ca276d7c1 100644 --- a/src/NATS.Client.JetStream/Internal/NatsJSConsume.cs +++ b/src/NATS.Client.JetStream/Internal/NatsJSConsume.cs @@ -42,7 +42,11 @@ internal class NatsJSConsume : NatsSubBase private readonly long _thresholdBytes; private readonly int _maxConsecutive503Errors; +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _pendingGate = new(); +#else private readonly object _pendingGate = new(); +#endif private long _pendingMsgs; private long _pendingBytes; private int _consecutive503Errors; diff --git a/src/NATS.Client.JetStream/Internal/NatsJSOrderedConsume.cs b/src/NATS.Client.JetStream/Internal/NatsJSOrderedConsume.cs index 900275d85..cc2cacf4a 100644 --- a/src/NATS.Client.JetStream/Internal/NatsJSOrderedConsume.cs +++ b/src/NATS.Client.JetStream/Internal/NatsJSOrderedConsume.cs @@ -31,7 +31,11 @@ internal class NatsJSOrderedConsume : NatsSubBase private readonly long _maxBytes; private readonly long _thresholdBytes; +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _pendingGate = new(); +#else private readonly object _pendingGate = new(); +#endif private long _pendingMsgs; private long _pendingBytes; diff --git a/src/NATS.Client.JetStream/NatsJSConsumer.cs b/src/NATS.Client.JetStream/NatsJSConsumer.cs index e10958563..dd66e0b43 100644 --- a/src/NATS.Client.JetStream/NatsJSConsumer.cs +++ b/src/NATS.Client.JetStream/NatsJSConsumer.cs @@ -14,7 +14,11 @@ public class NatsJSConsumer : INatsJSConsumer private readonly NatsJSContext _context; private readonly string _stream; private readonly string _consumer; +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _pinIdLock = new(); +#else private readonly object _pinIdLock = new(); +#endif private volatile bool _deleted; private string? _pinId; diff --git a/src/NATS.Extensions.Microsoft.DependencyInjection/NATS.Extensions.Microsoft.DependencyInjection.csproj b/src/NATS.Extensions.Microsoft.DependencyInjection/NATS.Extensions.Microsoft.DependencyInjection.csproj index 958310f71..2e48e7a4a 100644 --- a/src/NATS.Extensions.Microsoft.DependencyInjection/NATS.Extensions.Microsoft.DependencyInjection.csproj +++ b/src/NATS.Extensions.Microsoft.DependencyInjection/NATS.Extensions.Microsoft.DependencyInjection.csproj @@ -3,21 +3,21 @@ pubsub;messaging;cloud-native;aspnetcore;hosting;dependency-injection;nats - Microsoft dependency injection extension for NATS .NET. Configures NatsClient with the Generic Host and ASP.NET Core service collection. + NATS .NET dependency injection with ad-hoc JSON serialization enabled by default. Registers NatsConnection and INatsClient on the Microsoft service collection. For AOT or minimal dependencies, see NATS.Client.Hosting. - + - + - + diff --git a/src/NATS.Net/NATS.Net.csproj b/src/NATS.Net/NATS.Net.csproj index 088ff39ad..c0d441b97 100644 --- a/src/NATS.Net/NATS.Net.csproj +++ b/src/NATS.Net/NATS.Net.csproj @@ -9,6 +9,7 @@ + diff --git a/tests/NATS.Client.CheckAbi/Program.cs b/tests/NATS.Client.CheckAbi/Program.cs index 054a1c184..433b42ed0 100644 --- a/tests/NATS.Client.CheckAbi/Program.cs +++ b/tests/NATS.Client.CheckAbi/Program.cs @@ -6,24 +6,41 @@ // - AbiCheck references NATS.Net 2.7.0 (local source with type forwarders) // - TransientLib was compiled against NATS.Net 2.6.0 (types in NATS.Client.Core) // - At runtime, type forwarding should allow TransientLib to work with 2.7.0 -Console.WriteLine("ABI Compatibility Check (Transient Dependency Simulation)"); -Console.WriteLine("=========================================================="); -Console.WriteLine(); +var errors = new List(); + +void Check(string name, string expected, string actual) +{ + Console.WriteLine($" {name}: {actual}"); + if (actual != expected) + errors.Add($"{name}: expected '{expected}', got '{actual}'"); +} -// Check where THIS project sees the types (should be Abstractions since we use 2.7.0) Console.WriteLine("Types as seen by this project (compiled against 2.7.0):"); -Console.WriteLine($" INatsSerialize<> assembly: {typeof(INatsSerialize<>).Assembly.GetName().Name}"); -Console.WriteLine($" INatsDeserialize<> assembly: {typeof(INatsDeserialize<>).Assembly.GetName().Name}"); -Console.WriteLine($" INatsSerializer<> assembly: {typeof(INatsSerializer<>).Assembly.GetName().Name}"); -Console.WriteLine($" INatsSerializerRegistry assembly: {typeof(INatsSerializerRegistry).Assembly.GetName().Name}"); +Check("INatsSerialize<>", "NATS.Client.Abstractions", typeof(INatsSerialize<>).Assembly.GetName().Name!); +Check("INatsDeserialize<>", "NATS.Client.Abstractions", typeof(INatsDeserialize<>).Assembly.GetName().Name!); +Check("INatsSerializer<>", "NATS.Client.Abstractions", typeof(INatsSerializer<>).Assembly.GetName().Name!); +Check("INatsSerializerRegistry", "NATS.Client.Abstractions", typeof(INatsSerializerRegistry).Assembly.GetName().Name!); +Check("INatsSerializeWithContext<>", "NATS.Client.Abstractions", typeof(INatsSerializeWithContext<>).Assembly.GetName().Name!); +Check("INatsDeserializeWithContext<>", "NATS.Client.Abstractions", typeof(INatsDeserializeWithContext<>).Assembly.GetName().Name!); +Check("NatsMsgContext", "NATS.Client.Abstractions", typeof(NatsMsgContext).Assembly.GetName().Name!); Console.WriteLine(); -// Check where TransientLib sees the types (compiled against 2.6.0, but should resolve via forwarding) Console.WriteLine("Types as seen by TransientLib (compiled against 2.6.0):"); -Console.WriteLine($" INatsSerialize<> assembly: {MySerializer.GetSerializerInterfaceAssembly()}"); +Check("INatsSerialize<> (transient)", "NATS.Client.Abstractions", MySerializer.GetSerializerInterfaceAssembly()); +Console.WriteLine(); + +Console.WriteLine("Assembly versions at runtime:"); +Console.WriteLine($" INatsSerialize<> assembly version (from this project): {typeof(INatsSerialize<>).Assembly.GetName().Version}"); +Console.WriteLine($" INatsSerialize<> assembly version (from TransientLib): {MySerializer.GetSerializerInterfaceAssemblyVersion()}"); +foreach (var asm in AppDomain.CurrentDomain.GetAssemblies() + .Where(a => a.GetName().Name?.Contains("NATS") == true) + .OrderBy(a => a.GetName().Name)) +{ + Console.WriteLine($" {asm.GetName().Name} v{asm.GetName().Version} [{asm.Location}]"); +} + Console.WriteLine(); -// Use the serializer from TransientLib (compiled against 2.6.0) Console.WriteLine("Testing TransientLib.MySerializer (compiled against 2.6.0):"); var serializer = new MySerializer(); var buffer = new ArrayBufferWriter(); @@ -32,6 +49,18 @@ var deserialized = serializer.Deserialize(new ReadOnlySequence(buffer.WrittenSpan.ToArray())); Console.WriteLine($" Deserialized: '{deserialized}'"); +if (deserialized != "hello from transient dependency") + errors.Add($"Round-trip failed: got '{deserialized}'"); Console.WriteLine(); + +if (errors.Count > 0) +{ + Console.Error.WriteLine($"FAILED: {errors.Count} error(s):"); + foreach (var error in errors) + Console.Error.WriteLine($" {error}"); + return 1; +} + Console.WriteLine("SUCCESS: ABI compatibility verified with transient dependency!"); +return 0; diff --git a/tests/NATS.Client.CheckAbi/run-abi-check.sh b/tests/NATS.Client.CheckAbi/run-abi-check.sh index 94d1f25c3..ad352e9dd 100644 --- a/tests/NATS.Client.CheckAbi/run-abi-check.sh +++ b/tests/NATS.Client.CheckAbi/run-abi-check.sh @@ -5,30 +5,71 @@ set -e SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" TRANSIENT_LIB_CSPROJ="$SCRIPT_DIR/../NATS.Client.CheckAbiTransientLib/NATS.Client.CheckAbiTransientLib.csproj" CHECK_ABI_CSPROJ="$SCRIPT_DIR/NATS.Client.CheckAbi.csproj" +CHECK_ABI_OLD_CSPROJ="$SCRIPT_DIR/../NATS.Client.CheckAbiOld/NATS.Client.CheckAbiOld.csproj" + +VERSION=$(cat "$SCRIPT_DIR/../../version.txt" | tr -d '[:space:]') +OLD_VERSION=$(grep 'NatsAbiCheckVersion' "$SCRIPT_DIR/../../Directory.Build.props" | sed 's/.*>\([^<]*\)<.*/\1/') echo "=== NATS.Net ABI Compatibility Check (Transient Dependency Simulation) ===" echo "" echo "This test simulates:" -echo " - An app using NATS.Net 2.7.0 (with type forwarders)" -echo " - A transient dependency (TransientLib) compiled against NATS.Net 2.6.0" +echo " - An app using local NATS.Net $VERSION (with type forwarders)" +echo " - A transient dependency (TransientLib) compiled against NATS.Net $OLD_VERSION" echo " - Type forwarding should allow the old library to work with new NATS.Net" echo "" -# Step 1: Build CheckAbiTransientLib against 2.6.0 NuGet package -echo "[1/3] Building NATS.Client.CheckAbiTransientLib against NATS.Net 2.6.0..." +# Step 1: Build CheckAbiTransientLib against NuGet package +echo "[1/4] Building NATS.Client.CheckAbiTransientLib against NATS.Net $OLD_VERSION..." dotnet clean dotnet build "$TRANSIENT_LIB_CSPROJ" -c Release -# Step 2: Build AbiCheck (references local 2.7.0 source + TransientLib.dll) +# Step 2: Build and run CheckAbiOld (control: everything is 2.6.0) +echo "" +echo "[2/4] Building and running CheckAbiOld (control, NATS.Net $OLD_VERSION NuGet)..." +dotnet build "$CHECK_ABI_OLD_CSPROJ" -c Release + +OLD_OUTPUT=$(dotnet run --project "$CHECK_ABI_OLD_CSPROJ" -c Release --no-build) +echo "$OLD_OUTPUT" echo "" -echo "[2/3] Building AbiCheck against local NATS.Net 2.7.0 source + TransientLib..." + +# Verify old project sees types in NATS.Client.Core with the old version +if ! echo "$OLD_OUTPUT" | grep -q "NATS.Client.Core v${OLD_VERSION}.0"; then + echo "FAILED: CheckAbiOld should load NATS.Client.Core v${OLD_VERSION}.0" + exit 1 +fi +if echo "$OLD_OUTPUT" | grep -q "NATS.Client.Abstractions"; then + echo "FAILED: CheckAbiOld should NOT reference NATS.Client.Abstractions" + exit 1 +fi +echo "OK: CheckAbiOld correctly uses NATS.Client.Core v${OLD_VERSION}.0" + +# Step 3: Build AbiCheck (references local 2.7.0 source + TransientLib.dll) +echo "" +echo "[3/4] Building AbiCheck against local NATS.Net source + TransientLib..." dotnet build "$CHECK_ABI_CSPROJ" -c Release -# Step 3: Run the test +# Step 4: Run the test +echo "" +echo "[4/4] Running ABI compatibility check..." echo "" -echo "[3/3] Running ABI compatibility check..." + +NEW_OUTPUT=$(dotnet run --project "$CHECK_ABI_CSPROJ" -c Release --no-build) +echo "$NEW_OUTPUT" echo "" -dotnet run --project "$CHECK_ABI_CSPROJ" -c Release --no-build + +# Verify new project sees types forwarded to NATS.Client.Abstractions +if ! echo "$NEW_OUTPUT" | grep -q "NATS.Client.Abstractions"; then + echo "FAILED: CheckAbi should resolve types to NATS.Client.Abstractions" + exit 1 +fi +if echo "$NEW_OUTPUT" | grep -q "NATS.Client.Core v${OLD_VERSION}.0"; then + echo "FAILED: CheckAbi should NOT load NATS.Client.Core v${OLD_VERSION}.0" + exit 1 +fi +if ! echo "$NEW_OUTPUT" | grep -q "SUCCESS"; then + echo "FAILED: ABI compatibility check did not succeed" + exit 1 +fi echo "" echo "=== ABI Check Complete ===" diff --git a/tests/NATS.Client.CheckAbiOld/NATS.Client.CheckAbiOld.csproj b/tests/NATS.Client.CheckAbiOld/NATS.Client.CheckAbiOld.csproj new file mode 100644 index 000000000..7a5ac5499 --- /dev/null +++ b/tests/NATS.Client.CheckAbiOld/NATS.Client.CheckAbiOld.csproj @@ -0,0 +1,29 @@ + + + + Exe + net8.0 + enable + enable + + + + + + + + + + + false + false + + + + + ..\NATS.Client.CheckAbiTransientLib\bin\$(Configuration)\net8.0\NATS.Client.CheckAbiTransientLib.dll + true + + + + diff --git a/tests/NATS.Client.CheckAbiOld/Program.cs b/tests/NATS.Client.CheckAbiOld/Program.cs new file mode 100644 index 000000000..c2023778e --- /dev/null +++ b/tests/NATS.Client.CheckAbiOld/Program.cs @@ -0,0 +1,29 @@ +using NATS.Client.CheckAbiTransientLib; +using NATS.Client.Core; + +// This project references an older NATS.Net NuGet (same as TransientLib). +// It serves as a control: all types should resolve to NATS.Client.Core. +var coreVersion = typeof(INatsSerialize<>).Assembly.GetName().Version; +Console.WriteLine($"=== CheckAbiOld: running against NATS.Client.Core v{coreVersion} NuGet ==="); +Console.WriteLine(); + +Console.WriteLine($"Types as seen by this project:"); +Console.WriteLine($" INatsSerialize<>: {typeof(INatsSerialize<>).Assembly.GetName().Name}"); +Console.WriteLine($" INatsDeserialize<>: {typeof(INatsDeserialize<>).Assembly.GetName().Name}"); +Console.WriteLine($" INatsSerializer<>: {typeof(INatsSerializer<>).Assembly.GetName().Name}"); +Console.WriteLine($" INatsSerializerRegistry: {typeof(INatsSerializerRegistry).Assembly.GetName().Name}"); +Console.WriteLine(); + +Console.WriteLine($"Types as seen by TransientLib:"); +Console.WriteLine($" INatsSerialize<> (transient): {MySerializer.GetSerializerInterfaceAssembly()}"); +Console.WriteLine($" INatsSerialize<> version (transient): {MySerializer.GetSerializerInterfaceAssemblyVersion()}"); +Console.WriteLine(); + +Console.WriteLine("Assembly versions at runtime:"); +Console.WriteLine($" INatsSerialize<> assembly version: {typeof(INatsSerialize<>).Assembly.GetName().Version}"); +foreach (var asm in AppDomain.CurrentDomain.GetAssemblies() + .Where(a => a.GetName().Name?.Contains("NATS") == true) + .OrderBy(a => a.GetName().Name)) +{ + Console.WriteLine($" {asm.GetName().Name} v{asm.GetName().Version} [{asm.Location}]"); +} diff --git a/tests/NATS.Client.CheckAbiTransientLib/MySerializer.cs b/tests/NATS.Client.CheckAbiTransientLib/MySerializer.cs index 0d1e60648..7d70d9864 100644 --- a/tests/NATS.Client.CheckAbiTransientLib/MySerializer.cs +++ b/tests/NATS.Client.CheckAbiTransientLib/MySerializer.cs @@ -14,6 +14,11 @@ public static string GetSerializerInterfaceAssembly() return typeof(INatsSerialize<>).Assembly.GetName().Name!; } + public static string GetSerializerInterfaceAssemblyVersion() + { + return typeof(INatsSerialize<>).Assembly.GetName().Version?.ToString() ?? "(null)"; + } + public void Serialize(IBufferWriter bufferWriter, string value) { var bytes = System.Text.Encoding.UTF8.GetBytes(value); diff --git a/tests/NATS.Client.CheckAbiTransientLib/NATS.Client.CheckAbiTransientLib.csproj b/tests/NATS.Client.CheckAbiTransientLib/NATS.Client.CheckAbiTransientLib.csproj index 794a47fe6..9517cc2f7 100644 --- a/tests/NATS.Client.CheckAbiTransientLib/NATS.Client.CheckAbiTransientLib.csproj +++ b/tests/NATS.Client.CheckAbiTransientLib/NATS.Client.CheckAbiTransientLib.csproj @@ -7,7 +7,7 @@ - + diff --git a/tests/NATS.Client.Core.Tests/NATS.Client.Core.Tests.csproj b/tests/NATS.Client.Core.Tests/NATS.Client.Core.Tests.csproj index 6013ee793..fdbdefdc0 100644 --- a/tests/NATS.Client.Core.Tests/NATS.Client.Core.Tests.csproj +++ b/tests/NATS.Client.Core.Tests/NATS.Client.Core.Tests.csproj @@ -1,7 +1,7 @@  - net8.0 + net8.0;net10.0 $(TargetFrameworks);net481 any;win-x86 enable diff --git a/tests/NATS.Client.Core.Tests/NatsMsgTests.cs b/tests/NATS.Client.Core.Tests/NatsMsgTests.cs index 0e9d4905e..1d7977cce 100644 --- a/tests/NATS.Client.Core.Tests/NatsMsgTests.cs +++ b/tests/NATS.Client.Core.Tests/NatsMsgTests.cs @@ -1,3 +1,4 @@ +using System.Buffers; using System.Text; using NATS.Client.Serializers.Json; @@ -85,6 +86,38 @@ public void Create_nats_msg_by_object() msg.Size.Should().Be(expectedSize); } + [Fact] + public void Create_nats_msg_by_object_with_header_aware_serializer() + { + // Arrange + const string subject = "test"; + const string replyTo = "reply"; + var data = new TestData { Id = 1, Name = "example" }; + var headers = new NatsHeaders { { "X-Type", "test-data" } }; + + var serializer = new HeaderAwareSerializer(); + + // Act + var builder = new NatsMsgBuilder + { + Subject = subject, + Data = data, + Serializer = serializer, + Headers = headers, + ReplyTo = replyTo, + }; + var msg = builder.Msg; + + var bufferWriter = new NatsPooledBufferWriter(256); + ((INatsSerializeWithContext)serializer).Serialize(bufferWriter, data, new NatsMsgContext(subject, replyTo, headers)); + var serializedSize = bufferWriter.WrittenCount; + + var expectedSize = subject.Length + (replyTo?.Length ?? 0) + headers.GetBytesLength() + serializedSize; + + // Assert + msg.Size.Should().Be(expectedSize); + } + [Fact] public void Create_nats_msg_by_object_without_serializer() { @@ -110,4 +143,32 @@ private class TestData public string Name { get; set; } = null!; } + + private class HeaderAwareSerializer : INatsSerializer, INatsSerializerWithContext + { + private readonly NatsJsonSerializer _inner = new(); + + public void Serialize(IBufferWriter bufferWriter, T value) => _inner.Serialize(bufferWriter, value); + + public T? Deserialize(in ReadOnlySequence buffer) => _inner.Deserialize(buffer); + + public void Serialize(IBufferWriter bufferWriter, T value, in NatsMsgContext context) + { + // Write a header-based prefix before the JSON payload + if (context.Headers != null && context.Headers.TryGetValue("X-Type", out var values)) + { + var prefix = Encoding.UTF8.GetBytes(values.ToString() + ":"); + var span = bufferWriter.GetSpan(prefix.Length); + prefix.CopyTo(span); + bufferWriter.Advance(prefix.Length); + } + + _inner.Serialize(bufferWriter, value); + } + + public T? Deserialize(in ReadOnlySequence buffer, in NatsMsgContext context) => + _inner.Deserialize(buffer); + + public INatsSerializer CombineWith(INatsSerializer next) => this; + } } diff --git a/tests/NATS.Client.Core2.Tests/NATS.Client.Core2.Tests.csproj b/tests/NATS.Client.Core2.Tests/NATS.Client.Core2.Tests.csproj index 5d5a58b3f..6179c98db 100644 --- a/tests/NATS.Client.Core2.Tests/NATS.Client.Core2.Tests.csproj +++ b/tests/NATS.Client.Core2.Tests/NATS.Client.Core2.Tests.csproj @@ -1,7 +1,7 @@  - net8.0 + net8.0;net10.0 $(TargetFrameworks);net481 any;win-x86 enable @@ -28,8 +28,8 @@ runtime; build; native; contentfiles; analyzers; buildtransitive all - - + + diff --git a/tests/NATS.Client.Core2.Tests/SerializerTest.cs b/tests/NATS.Client.Core2.Tests/SerializerTest.cs index fbd987e79..89fedfaa8 100644 --- a/tests/NATS.Client.Core2.Tests/SerializerTest.cs +++ b/tests/NATS.Client.Core2.Tests/SerializerTest.cs @@ -324,6 +324,33 @@ public async Task Deserialize_using_json_stream_serializer_registry() Assert.Equal("two", result2.Data.Name); } + [Fact] + public async Task Serializer_can_mutate_headers_during_serialization() + { + await using var nats = new NatsConnection(new NatsOpts + { + Url = _server.Url, + SerializerRegistry = new HeaderMutatingSerializerRegistry(), + }); + + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + var cancellationToken = cts.Token; + + await nats.ConnectAsync(); + + var subject = _server.GetNextId(); + var sub = await nats.SubscribeCoreAsync(subject, cancellationToken: cancellationToken); + await nats.PingAsync(cancellationToken); + + await nats.PublishAsync(subject, "hello", headers: new NatsHeaders(), cancellationToken: cancellationToken); + + var msg = await sub.Msgs.ReadAsync(cancellationToken); + + Assert.Equal("hello", msg.Data); + Assert.NotNull(msg.Headers); + Assert.Equal("application/json", msg.Headers["Content-Type"].ToString()); + } + private static void AssertByteArray(byte[] expected, byte[] actual) { Assert.Equal(expected.Length, actual.Length); @@ -369,6 +396,30 @@ public record TestMessage1(string Name); public record TestMessage2(string Name); +public class HeaderMutatingSerializerRegistry : INatsSerializerRegistry +{ + public INatsSerialize GetSerializer() => new HeaderMutatingSerializer(); + + public INatsDeserialize GetDeserializer() => NatsDefaultSerializer.Default; +} + +public class HeaderMutatingSerializer : INatsSerialize, INatsSerializeWithContext +{ + public void Serialize(IBufferWriter bufferWriter, T value) + { + var bytes = Encoding.UTF8.GetBytes(value?.ToString() ?? string.Empty); + bufferWriter.Write(bytes); + } + + public void Serialize(IBufferWriter bufferWriter, T value, in NatsMsgContext context) + { + if (context.Headers != null) + context.Headers["Content-Type"] = "application/json"; + + Serialize(bufferWriter, value); + } +} + [JsonSerializable(typeof(TestMessage1))] [JsonSourceGenerationOptions(DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingDefault, WriteIndented = false)] internal partial class TestSerializerContext1 : JsonSerializerContext; diff --git a/tests/NATS.Client.CoreUnit.Tests/NATS.Client.CoreUnit.Tests.csproj b/tests/NATS.Client.CoreUnit.Tests/NATS.Client.CoreUnit.Tests.csproj index 9ea506692..eb8d97dbc 100644 --- a/tests/NATS.Client.CoreUnit.Tests/NATS.Client.CoreUnit.Tests.csproj +++ b/tests/NATS.Client.CoreUnit.Tests/NATS.Client.CoreUnit.Tests.csproj @@ -1,7 +1,7 @@  - net8.0 + net8.0;net10.0 $(TargetFrameworks);net481 any;win-x86 enable @@ -26,8 +26,8 @@ runtime; build; native; contentfiles; analyzers; buildtransitive all - - + + diff --git a/tests/NATS.Client.CoreUnit.Tests/NatsJsonSerializerTests.cs b/tests/NATS.Client.CoreUnit.Tests/NatsJsonSerializerTests.cs index b1d6d9454..166a4f8ec 100644 --- a/tests/NATS.Client.CoreUnit.Tests/NatsJsonSerializerTests.cs +++ b/tests/NATS.Client.CoreUnit.Tests/NatsJsonSerializerTests.cs @@ -24,11 +24,11 @@ public void RoundTrip_RequiredNullableProperty_ShouldSucceed() var bufferWriter = new ArrayBufferWriter(); // Act - Serialize - serializer.Serialize(bufferWriter, obj); + serializer.Serialize(bufferWriter, obj, default); // Deserialize back var buffer = new ReadOnlySequence(bufferWriter.WrittenMemory); - var result = serializer.Deserialize(buffer); + var result = serializer.Deserialize(buffer, default); // Assert - Round trip should succeed Assert.NotNull(result); @@ -44,7 +44,7 @@ public void Deserialize_RequiredNullableProperty_WithJsonSerializerOptionsDefaul var buffer = new ReadOnlySequence(json); // Act - var result = serializer.Deserialize(buffer); + var result = serializer.Deserialize(buffer, default); // Assert - This should work fine Assert.NotNull(result); @@ -60,7 +60,7 @@ public void RoundTrip_RequiredNullableProperty_WithJsonSerializerOptionsDefault_ var bufferWriter = new ArrayBufferWriter(); // Act - Serialize - serializer.Serialize(bufferWriter, obj); + serializer.Serialize(bufferWriter, obj, default); var json = Encoding.UTF8.GetString(bufferWriter.WrittenSpan); // With default options, null is included in JSON @@ -68,7 +68,7 @@ public void RoundTrip_RequiredNullableProperty_WithJsonSerializerOptionsDefault_ // Deserialize back var buffer = new ReadOnlySequence(bufferWriter.WrittenMemory); - var result = serializer.Deserialize(buffer); + var result = serializer.Deserialize(buffer, default); // Assert - Round trip succeeds Assert.NotNull(result); diff --git a/tests/NATS.Client.CoreUnit.Tests/NatsSerializationExtensionsTests.cs b/tests/NATS.Client.CoreUnit.Tests/NatsSerializationExtensionsTests.cs new file mode 100644 index 000000000..85a41936d --- /dev/null +++ b/tests/NATS.Client.CoreUnit.Tests/NatsSerializationExtensionsTests.cs @@ -0,0 +1,194 @@ +using System.Buffers; + +namespace NATS.Client.CoreUnit.Tests; + +public class NatsSerializationExtensionsTests +{ + [Fact] + public void Serialize_with_context_aware_serializer_calls_context_overload() + { + var serializer = new TrackingSerializerWithContext(); + var buffer = new NatsPooledBufferWriter(256); + var context = new NatsMsgContext("test", headers: new NatsHeaders { { "X-Test", "value" } }); + + ((INatsSerialize)serializer).Serialize(buffer, "test", in context); + + serializer.ContextSerializeCalled.Should().BeTrue(); + serializer.StandardSerializeCalled.Should().BeFalse(); + } + + [Fact] + public void Serialize_without_context_aware_serializer_falls_back() + { + var serializer = new TrackingSerializer(); + var buffer = new NatsPooledBufferWriter(256); + var context = new NatsMsgContext("test", headers: new NatsHeaders { { "X-Test", "value" } }); + + ((INatsSerialize)serializer).Serialize(buffer, "test", in context); + + serializer.StandardSerializeCalled.Should().BeTrue(); + } + + [Fact] + public void Serialize_with_null_headers_still_calls_context_overload() + { + var serializer = new TrackingSerializerWithContext(); + var buffer = new NatsPooledBufferWriter(256); + var context = new NatsMsgContext("test"); + + ((INatsSerialize)serializer).Serialize(buffer, "test", in context); + + serializer.ContextSerializeCalled.Should().BeTrue(); + serializer.StandardSerializeCalled.Should().BeFalse(); + } + + [Fact] + public void Deserialize_with_context_aware_deserializer_calls_context_overload() + { + var deserializer = new TrackingDeserializerWithContext(); + var buffer = new ReadOnlySequence(new byte[] { 1 }); + var context = new NatsMsgContext("test", headers: new NatsHeaders { { "X-Test", "value" } }); + + ((INatsDeserialize)deserializer).Deserialize(buffer, in context); + + deserializer.ContextDeserializeCalled.Should().BeTrue(); + deserializer.StandardDeserializeCalled.Should().BeFalse(); + } + + [Fact] + public void Deserialize_without_context_aware_deserializer_falls_back() + { + var deserializer = new TrackingDeserializer(); + var buffer = new ReadOnlySequence(new byte[] { 1 }); + var context = new NatsMsgContext("test", headers: new NatsHeaders { { "X-Test", "value" } }); + + ((INatsDeserialize)deserializer).Deserialize(buffer, in context); + + deserializer.StandardDeserializeCalled.Should().BeTrue(); + } + + [Fact] + public void Deserialize_with_null_headers_still_calls_context_overload() + { + var deserializer = new TrackingDeserializerWithContext(); + var buffer = new ReadOnlySequence(new byte[] { 1 }); + var context = new NatsMsgContext("test"); + + ((INatsDeserialize)deserializer).Deserialize(buffer, in context); + + deserializer.ContextDeserializeCalled.Should().BeTrue(); + deserializer.StandardDeserializeCalled.Should().BeFalse(); + } + + [Fact] + public void Serialize_built_in_chain_propagates_context_to_leaf() + { + // A non-context built-in (NatsRawSerializer) chained with a context-aware leaf. + // For a type it can't handle, the built-in delegates to _next; the context must arrive at the leaf. + var leaf = new TrackingContextSerializer(); + var chained = NatsRawSerializer.Default.CombineWith(leaf); + var buffer = new NatsPooledBufferWriter(256); + var context = new NatsMsgContext("test", headers: new NatsHeaders { { "X-Test", "value" } }); + + chained.Serialize(buffer, "hello", in context); + + leaf.ContextSerializeCalled.Should().BeTrue(); + leaf.StandardSerializeCalled.Should().BeFalse(); + } + + [Fact] + public void Deserialize_built_in_chain_propagates_context_to_leaf() + { + var leaf = new TrackingContextSerializer(); + var chained = NatsRawSerializer.Default.CombineWith(leaf); + var buffer = new ReadOnlySequence(new byte[] { 1 }); + var context = new NatsMsgContext("test", headers: new NatsHeaders { { "X-Test", "value" } }); + + chained.Deserialize(buffer, in context); + + leaf.ContextDeserializeCalled.Should().BeTrue(); + leaf.StandardDeserializeCalled.Should().BeFalse(); + } + + private class TrackingSerializer : INatsSerialize + { + public bool StandardSerializeCalled { get; private set; } + + public void Serialize(IBufferWriter bufferWriter, string value) => + StandardSerializeCalled = true; + } + + private class TrackingSerializerWithContext : INatsSerialize, INatsSerializeWithContext + { + public bool StandardSerializeCalled { get; private set; } + + public bool ContextSerializeCalled { get; private set; } + + public void Serialize(IBufferWriter bufferWriter, string value) => + StandardSerializeCalled = true; + + public void Serialize(IBufferWriter bufferWriter, string value, in NatsMsgContext context) => + ContextSerializeCalled = true; + } + + private class TrackingDeserializer : INatsDeserialize + { + public bool StandardDeserializeCalled { get; private set; } + + public string? Deserialize(in ReadOnlySequence buffer) + { + StandardDeserializeCalled = true; + return null; + } + } + + private class TrackingDeserializerWithContext : INatsDeserialize, INatsDeserializeWithContext + { + public bool StandardDeserializeCalled { get; private set; } + + public bool ContextDeserializeCalled { get; private set; } + + public string? Deserialize(in ReadOnlySequence buffer) + { + StandardDeserializeCalled = true; + return null; + } + + public string? Deserialize(in ReadOnlySequence buffer, in NatsMsgContext context) + { + ContextDeserializeCalled = true; + return null; + } + } + + private class TrackingContextSerializer : INatsSerializer, INatsSerializerWithContext + { + public bool StandardSerializeCalled { get; private set; } + + public bool ContextSerializeCalled { get; private set; } + + public bool StandardDeserializeCalled { get; private set; } + + public bool ContextDeserializeCalled { get; private set; } + + public void Serialize(IBufferWriter bufferWriter, string value) => + StandardSerializeCalled = true; + + public void Serialize(IBufferWriter bufferWriter, string value, in NatsMsgContext context) => + ContextSerializeCalled = true; + + public string? Deserialize(in ReadOnlySequence buffer) + { + StandardDeserializeCalled = true; + return null; + } + + public string? Deserialize(in ReadOnlySequence buffer, in NatsMsgContext context) + { + ContextDeserializeCalled = true; + return null; + } + + public INatsSerializer CombineWith(INatsSerializer next) => throw new NotSupportedException(); + } +} diff --git a/tests/NATS.Client.JetStream.Tests/EnumJsonTests.cs b/tests/NATS.Client.JetStream.Tests/EnumJsonTests.cs index 4eef1a5d4..b3eab5190 100644 --- a/tests/NATS.Client.JetStream.Tests/EnumJsonTests.cs +++ b/tests/NATS.Client.JetStream.Tests/EnumJsonTests.cs @@ -20,12 +20,12 @@ public void ConsumerConfigAckPolicy_test(ConsumerConfigAckPolicy value, string e var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new ConsumerConfig { AckPolicy = value }); + serializer.Serialize(bw, new ConsumerConfig { AckPolicy = value }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Contains(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(value, result.AckPolicy); } @@ -42,12 +42,12 @@ public void ConsumerConfigDeliverPolicy_test(ConsumerConfigDeliverPolicy value, var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new ConsumerConfig { DeliverPolicy = value }); + serializer.Serialize(bw, new ConsumerConfig { DeliverPolicy = value }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Contains(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(value, result.DeliverPolicy); } @@ -60,12 +60,12 @@ public void ConsumerConfigReplayPolicy_test(ConsumerConfigReplayPolicy value, st var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new ConsumerConfig { ReplayPolicy = value }); + serializer.Serialize(bw, new ConsumerConfig { ReplayPolicy = value }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Contains(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(value, result.ReplayPolicy); } @@ -78,7 +78,7 @@ public void StreamConfigCompression_test(StreamConfigCompression value, string e var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new StreamConfig { Compression = value }); + serializer.Serialize(bw, new StreamConfig { Compression = value }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); if (value == default) @@ -86,7 +86,7 @@ public void StreamConfigCompression_test(StreamConfigCompression value, string e else Assert.Contains(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(value, result.Compression); } @@ -99,7 +99,7 @@ public void StreamConfigDiscard_test(StreamConfigDiscard value, string expected) var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new StreamConfig { Discard = value }); + serializer.Serialize(bw, new StreamConfig { Discard = value }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); if (value == default) @@ -107,7 +107,7 @@ public void StreamConfigDiscard_test(StreamConfigDiscard value, string expected) else Assert.Contains(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(value, result.Discard); } @@ -121,12 +121,12 @@ public void StreamConfigRetention_test(StreamConfigRetention value, string expec var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new StreamConfig { Retention = value }); + serializer.Serialize(bw, new StreamConfig { Retention = value }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Contains(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(value, result.Retention); } @@ -139,12 +139,12 @@ public void StreamConfigStorage_test(StreamConfigStorage value, string expected) var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new StreamConfig { Storage = value }); + serializer.Serialize(bw, new StreamConfig { Storage = value }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Contains(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(value, result.Storage); } @@ -158,12 +158,12 @@ public void ConsumerCreateRequestAction_Test(ConsumerCreateAction value, string var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new ConsumerCreateRequest { Action = value, StreamName = string.Empty }); + serializer.Serialize(bw, new ConsumerCreateRequest { Action = value, StreamName = string.Empty }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Contains(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(value, result.Action); } @@ -176,13 +176,13 @@ public void StreamConfigPersistMode_null_not_serialized() // When PersistMode is null (not explicitly set), it should not be included in JSON var bw = new NatsBufferWriter(); var config = new StreamConfig { PersistMode = null }; - serializer.Serialize(bw, config); + serializer.Serialize(bw, config, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.DoesNotContain("persist_mode", json); // Deserialize and verify it remains null - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Null(result.PersistMode); } @@ -197,13 +197,13 @@ public void StreamConfigPersistMode_explicit_value_serialized(StreamConfigPersis // When PersistMode is explicitly set (even to Default), it should be included in JSON var bw = new NatsBufferWriter(); var config = new StreamConfig { PersistMode = value }; - serializer.Serialize(bw, config); + serializer.Serialize(bw, config, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Contains(expected, json); // Deserialize and verify the value is preserved - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(value, result.PersistMode); } @@ -216,39 +216,39 @@ public void StreamConfigPersistMode_roundtrip_preserves_server_values() // Test case 1: Server returns "default" - should preserve it var jsonWithDefault = "{\"persist_mode\":\"default\",\"retention\":\"limits\",\"storage\":\"file\"}"; var bytes = Encoding.UTF8.GetBytes(jsonWithDefault); - var configFromServer = serializer.Deserialize(new ReadOnlySequence(bytes)); + var configFromServer = serializer.Deserialize(new ReadOnlySequence(bytes), default); Assert.NotNull(configFromServer); Assert.Equal(StreamConfigPersistMode.Default, configFromServer.PersistMode); // When we serialize it again, it should include persist_mode var bw1 = new NatsBufferWriter(); - serializer.Serialize(bw1, configFromServer); + serializer.Serialize(bw1, configFromServer, default); var jsonOut1 = Encoding.UTF8.GetString(bw1.WrittenSpan.ToArray()); Assert.Contains("\"persist_mode\":\"default\"", jsonOut1); // Test case 2: Server returns "async" - should preserve it var jsonWithAsync = "{\"persist_mode\":\"async\",\"retention\":\"limits\",\"storage\":\"file\"}"; bytes = Encoding.UTF8.GetBytes(jsonWithAsync); - configFromServer = serializer.Deserialize(new ReadOnlySequence(bytes)); + configFromServer = serializer.Deserialize(new ReadOnlySequence(bytes), default); Assert.NotNull(configFromServer); Assert.Equal(StreamConfigPersistMode.Async, configFromServer.PersistMode); // When we serialize it again, it should include persist_mode var bw2 = new NatsBufferWriter(); - serializer.Serialize(bw2, configFromServer); + serializer.Serialize(bw2, configFromServer, default); var jsonOut2 = Encoding.UTF8.GetString(bw2.WrittenSpan.ToArray()); Assert.Contains("\"persist_mode\":\"async\"", jsonOut2); // Test case 3: Server doesn't return persist_mode - should remain null var jsonWithoutPersistMode = "{\"retention\":\"limits\",\"storage\":\"file\"}"; bytes = Encoding.UTF8.GetBytes(jsonWithoutPersistMode); - configFromServer = serializer.Deserialize(new ReadOnlySequence(bytes)); + configFromServer = serializer.Deserialize(new ReadOnlySequence(bytes), default); Assert.NotNull(configFromServer); Assert.Null(configFromServer.PersistMode); // When we serialize it again, it should NOT include persist_mode var bw3 = new NatsBufferWriter(); - serializer.Serialize(bw3, configFromServer); + serializer.Serialize(bw3, configFromServer, default); var jsonOut3 = Encoding.UTF8.GetString(bw3.WrittenSpan.ToArray()); Assert.DoesNotContain("persist_mode", jsonOut3); } @@ -262,12 +262,12 @@ public void ConsumerConfigPriorityPolicy_test(ConsumerConfigPriorityPolicy value var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new ConsumerConfig { PriorityPolicy = value }); + serializer.Serialize(bw, new ConsumerConfig { PriorityPolicy = value }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Contains(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(value, result.PriorityPolicy); } @@ -278,12 +278,12 @@ public void ConsumerConfigPriorityPolicy_null_test() var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new ConsumerConfig { PriorityPolicy = null }); + serializer.Serialize(bw, new ConsumerConfig { PriorityPolicy = null }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.DoesNotContain("priority_policy", json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Null(result.PriorityPolicy); } diff --git a/tests/NATS.Client.JetStream.Tests/JetStreamApiSerializerTest.cs b/tests/NATS.Client.JetStream.Tests/JetStreamApiSerializerTest.cs index 2bce33d27..58c0cdb2b 100644 --- a/tests/NATS.Client.JetStream.Tests/JetStreamApiSerializerTest.cs +++ b/tests/NATS.Client.JetStream.Tests/JetStreamApiSerializerTest.cs @@ -101,7 +101,7 @@ await Retry.Until( public void Deserialize_value() { var serializer = NatsJSJsonDocumentSerializer.Default; - var result = serializer.Deserialize(new ReadOnlySequence(Encoding.UTF8.GetBytes("""{"memory":1}"""))); + var result = serializer.Deserialize(new ReadOnlySequence(Encoding.UTF8.GetBytes("""{"memory":1}""")), default); result.Value.Memory.Should().Be(1); } @@ -109,7 +109,7 @@ public void Deserialize_value() public void Deserialize_empty_buffer() { var serializer = NatsJSJsonDocumentSerializer.Default; - var result = serializer.Deserialize(ReadOnlySequence.Empty); + var result = serializer.Deserialize(ReadOnlySequence.Empty, default); result.Exception.Message.Should().Be("Buffer is empty"); } @@ -117,7 +117,7 @@ public void Deserialize_empty_buffer() public void Deserialize_error() { var serializer = NatsJSJsonDocumentSerializer.Default; - var result = serializer.Deserialize(new ReadOnlySequence(Encoding.UTF8.GetBytes("""{"error":{"code":2}}"""))); + var result = serializer.Deserialize(new ReadOnlySequence(Encoding.UTF8.GetBytes("""{"error":{"code":2}}""")), default); result.Error.Code.Should().Be(2); } } diff --git a/tests/NATS.Client.JetStream.Tests/ParseJsonTests.cs b/tests/NATS.Client.JetStream.Tests/ParseJsonTests.cs index 4fae0ba01..b0a9acd57 100644 --- a/tests/NATS.Client.JetStream.Tests/ParseJsonTests.cs +++ b/tests/NATS.Client.JetStream.Tests/ParseJsonTests.cs @@ -24,12 +24,12 @@ public void Placement_properties_should_be_optional() var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new Placement()); + serializer.Serialize(bw, new Placement(), default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Equal("{}", json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Null(result.Cluster); Assert.Null(result.Tags); @@ -41,7 +41,7 @@ public void Default_consumer_ack_policy_should_be_explicit() var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new ConsumerConfig()); + serializer.Serialize(bw, new ConsumerConfig(), default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Matches("\"ack_policy\":\"explicit\"", json); diff --git a/tests/NATS.Client.JetStream.Tests/TimeSpanJsonTests.cs b/tests/NATS.Client.JetStream.Tests/TimeSpanJsonTests.cs index c382f5f67..ff1c36f04 100644 --- a/tests/NATS.Client.JetStream.Tests/TimeSpanJsonTests.cs +++ b/tests/NATS.Client.JetStream.Tests/TimeSpanJsonTests.cs @@ -48,12 +48,12 @@ public void ConsumerConfigAckWait_test(string value, string expected) var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new ConsumerConfig { AckWait = time }); + serializer.Serialize(bw, new ConsumerConfig { AckWait = time }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Matches(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(time, result.AckWait); } @@ -68,12 +68,12 @@ public void ConsumerConfigIdleHeartbeat_test(string value, string expected) var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new ConsumerConfig { IdleHeartbeat = time }); + serializer.Serialize(bw, new ConsumerConfig { IdleHeartbeat = time }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Matches(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(time, result.IdleHeartbeat); } @@ -88,12 +88,12 @@ public void ConsumerConfigInactiveThreshold_test(string value, string expected) var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new ConsumerConfig { InactiveThreshold = time }); + serializer.Serialize(bw, new ConsumerConfig { InactiveThreshold = time }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Matches(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(time, result.InactiveThreshold); } @@ -108,12 +108,12 @@ public void ConsumerConfigMaxExpires_test(string value, string expected) var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new ConsumerConfig { MaxExpires = time }); + serializer.Serialize(bw, new ConsumerConfig { MaxExpires = time }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Matches(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(time, result.MaxExpires); } @@ -128,12 +128,12 @@ public void ConsumerGetnextRequestExpires_test(string value, string expected) var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new ConsumerGetnextRequest { Expires = time }); + serializer.Serialize(bw, new ConsumerGetnextRequest { Expires = time }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Matches(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(time, result.Expires); } @@ -148,12 +148,12 @@ public void ConsumerGetnextRequestIdleHeartbeat_test(string value, string expect var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new ConsumerGetnextRequest { IdleHeartbeat = time }); + serializer.Serialize(bw, new ConsumerGetnextRequest { IdleHeartbeat = time }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Matches(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(time, result.IdleHeartbeat); } @@ -168,12 +168,12 @@ public void PeerInfoActive_test(string value, string expected) var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new PeerInfo { Name = "test", Active = time }); + serializer.Serialize(bw, new PeerInfo { Name = "test", Active = time }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Matches(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(time, result.Active); } @@ -188,12 +188,12 @@ public void StreamConfigMaxAge_test(string value, string expected) var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new StreamConfig { MaxAge = time }); + serializer.Serialize(bw, new StreamConfig { MaxAge = time }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Matches(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(time, result.MaxAge); } @@ -208,12 +208,12 @@ public void StreamConfigDuplicateWindow_test(string value, string expected) var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new StreamConfig { DuplicateWindow = time }); + serializer.Serialize(bw, new StreamConfig { DuplicateWindow = time }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Matches(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(time, result.DuplicateWindow); } @@ -228,12 +228,12 @@ public void StreamSourceInfoActive_test(string value, string expected) var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new StreamSourceInfo { Name = "test", Active = time }); + serializer.Serialize(bw, new StreamSourceInfo { Name = "test", Active = time }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Matches(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(time, result.Active); } @@ -246,12 +246,12 @@ public void StreamSourceInfoActive_minus_one_indicates_no_activity() var serializer = NatsJSJsonSerializer.Default; var jsonWithMinusOne = """{"name":"test","lag":0,"active":-1}"""u8; - var resultMinusOne = serializer.Deserialize(new ReadOnlySequence(jsonWithMinusOne.ToArray())); + var resultMinusOne = serializer.Deserialize(new ReadOnlySequence(jsonWithMinusOne.ToArray()), default); Assert.NotNull(resultMinusOne); Assert.Null(resultMinusOne.Active); var jsonWithZero = """{"name":"test","lag":0,"active":0}"""u8; - var resultZero = serializer.Deserialize(new ReadOnlySequence(jsonWithZero.ToArray())); + var resultZero = serializer.Deserialize(new ReadOnlySequence(jsonWithZero.ToArray()), default); Assert.NotNull(resultZero); Assert.Equal(TimeSpan.Zero, resultZero.Active); @@ -267,12 +267,12 @@ public void StreamSourceInfoActive_null_roundtrip() // Serialize null -> should produce -1 var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new StreamSourceInfo { Name = "test", Active = null }); + serializer.Serialize(bw, new StreamSourceInfo { Name = "test", Active = null }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Contains("\"active\":-1", json); // Deserialize -1 -> should produce null - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Null(result.Active); } @@ -288,7 +288,7 @@ public void ConsumerInfoPauseRemaining_test(string? value, string expected) var serializer = NatsJSJsonSerializer.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new ConsumerInfo { StreamName = "test", Name = "test", PauseRemaining = time }); + serializer.Serialize(bw, new ConsumerInfo { StreamName = "test", Name = "test", PauseRemaining = time }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); if (value != null) @@ -301,7 +301,7 @@ public void ConsumerInfoPauseRemaining_test(string? value, string expected) Assert.DoesNotMatch(expected, "pause_remaining"); } - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(time, result.PauseRemaining); } @@ -316,12 +316,12 @@ public void ConsumerConfigBackoff_test(int minimumFrameworkVersion, List.Default; var bw = new NatsBufferWriter(); - serializer.Serialize(bw, new ConsumerConfig { Backoff = timeSpans }); + serializer.Serialize(bw, new ConsumerConfig { Backoff = timeSpans }, default); var json = Encoding.UTF8.GetString(bw.WrittenSpan.ToArray()); Assert.Matches(expected, json); - var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory)); + var result = serializer.Deserialize(new ReadOnlySequence(bw.WrittenMemory), default); Assert.NotNull(result); Assert.Equal(timeSpans, result.Backoff); } diff --git a/tests/NATS.Net.DocsExamples/Advanced/SerializationPage.cs b/tests/NATS.Net.DocsExamples/Advanced/SerializationPage.cs index a8550387d..301ebc34a 100644 --- a/tests/NATS.Net.DocsExamples/Advanced/SerializationPage.cs +++ b/tests/NATS.Net.DocsExamples/Advanced/SerializationPage.cs @@ -15,6 +15,7 @@ using System.Text.Json.Serialization; using Google.Protobuf; using Google.Protobuf.Reflection; +using Microsoft.Extensions.Primitives; using NATS.Client.Core; namespace NATS.Net.DocsExamples.Advanced; @@ -278,6 +279,48 @@ public class MixedSerializerRegistry : INatsSerializerRegistry } #endregion +#region header-aware-serializer +public class MyHeaderAwareSerializer : INatsSerializer, INatsSerializerWithContext +{ + private readonly NatsJsonContextSerializer _jsonSerializer; + + public MyHeaderAwareSerializer(JsonSerializerContext context) + { + _jsonSerializer = new NatsJsonContextSerializer(context); + } + + public void Serialize(IBufferWriter bufferWriter, T value) => _jsonSerializer.Serialize(bufferWriter, value); + + public T? Deserialize(in ReadOnlySequence buffer) => _jsonSerializer.Deserialize(buffer); + + public void Serialize(IBufferWriter bufferWriter, T value, in NatsMsgContext context) + { + // Set a content-type header so the deserializer knows the format + if (context.Headers != null) + { + context.Headers["Content-Type"] = "application/json"; + } + + _jsonSerializer.Serialize(bufferWriter, value); + } + + public T? Deserialize(in ReadOnlySequence buffer, in NatsMsgContext context) + { + // Read the content-type header to determine how to deserialize + if (context.Headers != null + && context.Headers.TryGetValue("Content-Type", out StringValues contentType) + && contentType.ToString() == "application/json") + { + return _jsonSerializer.Deserialize(buffer); + } + + throw new NatsException($"Unsupported content type for {typeof(T)}"); + } + + public INatsSerializer CombineWith(INatsSerializer next) => throw new NotSupportedException(); +} +#endregion + // Fake protobuf message. // Normally, this would be generated using protobuf compiler. public class Greeting : IBufferMessage diff --git a/tests/NATS.Slow.Tests/NatsConnectionTest.Headers.cs b/tests/NATS.Slow.Tests/NatsConnectionTest.Headers.cs index ee3c95332..ebf65275a 100644 --- a/tests/NATS.Slow.Tests/NatsConnectionTest.Headers.cs +++ b/tests/NATS.Slow.Tests/NatsConnectionTest.Headers.cs @@ -37,17 +37,10 @@ await Retry.Until( ["Test-Header-Key"] = "test-header-value", ["Multi"] = new[] { "multi-value-0", "multi-value-1" }, }; - Assert.False(headers.IsReadOnly); // Send with headers await nats.PublishAsync("foo", 100, headers: headers); - Assert.True(headers.IsReadOnly); - Assert.Throws(() => - { - headers["should-not-set"] = "value"; - }); - var msg1 = await signal1; Assert.Equal(100, msg1.Data); Assert.NotNull(msg1.Headers); diff --git a/tools/site_src/documentation/advanced/aot.md b/tools/site_src/documentation/advanced/aot.md index 52a12b1e2..ede99cc39 100644 --- a/tools/site_src/documentation/advanced/aot.md +++ b/tools/site_src/documentation/advanced/aot.md @@ -13,3 +13,10 @@ NuGet packages that are compatible with AOT publishing are: - [NATS.Client.KeyValueStore](https://www.nuget.org/packages/NATS.Client.KeyValueStore) - [NATS.Client.ObjectStore](https://www.nuget.org/packages/NATS.Client.ObjectStore) - [NATS.Client.Services](https://www.nuget.org/packages/NATS.Client.Services) +- [NATS.Client.Hosting](https://www.nuget.org/packages/NATS.Client.Hosting) (for dependency injection) + +> [!NOTE] +> For dependency injection in AOT scenarios, use `NATS.Client.Hosting` (with its `AddNats()` method) +> instead of `NATS.Extensions.Microsoft.DependencyInjection`. The latter depends on `NATS.Client.Simplified` which +> includes the ad hoc JSON serializer and is not AOT-compatible. +> See [Dependency Injection](dependency-injection.md) for a full comparison of the two DI packages. diff --git a/tools/site_src/documentation/advanced/dependency-injection.md b/tools/site_src/documentation/advanced/dependency-injection.md new file mode 100644 index 000000000..f3828993a --- /dev/null +++ b/tools/site_src/documentation/advanced/dependency-injection.md @@ -0,0 +1,194 @@ +# Dependency Injection + +NATS .NET provides two packages for integrating with Microsoft's dependency injection framework. +We acknowledge this can be confusing — they exist because they serve slightly different use cases, +particularly around serialization defaults and AOT compatibility. + +## Which Package Should I Use? + +| | `NATS.Extensions.Microsoft.DependencyInjection` | `NATS.Client.Hosting` | +|---|---|---| +| **Best for** | Most applications | AOT deployments, minimal dependencies | +| **Entry method** | `AddNatsClient()` | `AddNats()` | +| **API style** | Builder pattern (fluent) | Direct parameters | +| **JSON serialization** | Enabled by default (ad hoc) | Not included | +| **AOT compatible** | No | Yes | +| **Depends on** | `NATS.Client.Simplified` (includes Core + JSON serializer) | `NATS.Client.Core` only | +| **Included in `NATS.Net`** | Yes | Yes | + +**In short:** +- Use **`NATS.Extensions.Microsoft.DependencyInjection`** if you want things to "just work" with JSON serialization + out of the box and you don't need AOT publishing. +- Use **`NATS.Client.Hosting`** if you need AOT compatibility, want minimal dependencies, + or prefer to configure serialization yourself. + +## NATS.Extensions.Microsoft.DependencyInjection + +Install the package: + +```shell +dotnet add package NATS.Extensions.Microsoft.DependencyInjection +``` + +This package provides `AddNatsClient()` with a builder pattern for configuration. +It is included in the `NATS.Net` meta-package, but can also be installed standalone. +It brings in the ad hoc JSON serializer (`NATS.Client.Serializers.Json` via `NATS.Client.Simplified`), +so you can publish and subscribe with data classes using JSON serialization without any extra setup. + +### Basic Usage + +```csharp +var builder = WebApplication.CreateBuilder(args); + +builder.Services.AddNatsClient(); +``` + +This registers `INatsConnection`, `NatsConnection`, and `INatsClient` in the DI container. + +### Configuration + +```csharp +builder.Services.AddNatsClient(nats => +{ + nats.ConfigureOptions(opts => + opts.Configure(o => o.Opts = o.Opts with { Url = "nats://myserver:4222" })); +}); +``` + +You can also configure the connection after it's created: + +```csharp +builder.Services.AddNatsClient(nats => +{ + nats.ConfigureConnection((provider, connection) => + { + // Access DI services and configure the connection + }); +}); +``` + +### Keyed Services (.NET 8+) + +Register multiple NATS connections with different keys: + +```csharp +builder.Services.AddNatsClient(nats => +{ + nats.WithKey("primary"); + nats.ConfigureOptions(opts => + opts.Configure(o => o.Opts = o.Opts with { Url = "nats://primary:4222" })); +}); + +builder.Services.AddNatsClient(nats => +{ + nats.WithKey("secondary"); + nats.ConfigureOptions(opts => + opts.Configure(o => o.Opts = o.Opts with { Url = "nats://secondary:4222" })); +}); +``` + +Inject with `[FromKeyedServices]`: + +```csharp +public class MyService([FromKeyedServices("primary")] INatsConnection primaryNats) +{ +} +``` + +> [!NOTE] +> `WithKey()` must be called before `ConfigureOptions()` or other configuration methods. + +## NATS.Client.Hosting + +Install the package: + +```shell +dotnet add package NATS.Client.Hosting +``` + +This package provides `AddNats()` with a simpler callback-based API. +It depends only on `NATS.Client.Core`, so it doesn't bring in the JSON serializer +or any other higher-level client packages. This makes it suitable for +[AOT deployments](aot.md) and scenarios where you want to control your dependency footprint. + +### Basic Usage + +```csharp +var builder = WebApplication.CreateBuilder(args); + +builder.Services.AddNats(); +``` + +This registers `INatsConnection` and `NatsConnection` in the DI container. + +### Configuration + +```csharp +builder.Services.AddNats( + configureOpts: opts => opts with { Url = "nats://myserver:4222" }, + configureConnection: conn => + { + // Configure the connection after creation + }); +``` + +### Keyed Services (.NET 8+) + +```csharp +builder.Services.AddNats(key: "primary", configureOpts: opts => opts with { Url = "nats://primary:4222" }); +builder.Services.AddNats(key: "secondary", configureOpts: opts => opts with { Url = "nats://secondary:4222" }); +``` + +### Serialization + +Since `NATS.Client.Hosting` does not include a JSON serializer, you need to set up serialization yourself +if you need to work with data types beyond the built-in `int`, `string`, and `byte[]` support. +See [Serialization](serialization.md) for details on configuring custom serializers, +and [AOT Deployments](aot.md) for AOT-specific guidance. + +```csharp +builder.Services.AddNats(configureOpts: opts => opts with +{ + SerializerRegistry = new MyCustomSerializerRegistry(), +}); +``` + +## Key Differences in Detail + +### Serialization Defaults + +`NATS.Extensions.Microsoft.DependencyInjection` automatically configures `NatsClientDefaultSerializerRegistry`, +which supports ad hoc JSON serialization using `System.Text.Json`. This means you can immediately +publish and subscribe with your own data classes: + +```csharp +// Works out of the box with NATS.Extensions.Microsoft.DependencyInjection +await connection.PublishAsync("orders", new Order { Id = 1, Item = "Widget" }); +``` + +`NATS.Client.Hosting` uses the default `NatsOpts` serializer registry, which only supports +`int`, `string`, `byte[]`, and other primitive types. For anything else, you must configure +a serializer. + +### Channel Full Mode + +`NATS.Extensions.Microsoft.DependencyInjection` sets `SubPendingChannelFullMode` to +`BoundedChannelFullMode.Wait` by default (matching `NatsClient` behavior). This means +a slow subscriber will apply backpressure instead of dropping messages. + +`NATS.Client.Hosting` uses the `NatsOpts` default of `BoundedChannelFullMode.DropNewest`, +which drops the newest messages when the subscriber's channel is full. + +### Options Pattern + +`NATS.Extensions.Microsoft.DependencyInjection` uses the Microsoft Options pattern +(`IOptionsMonitor`) under the hood, providing better integration with the configuration +system and support for named options. + +`NATS.Client.Hosting` uses simple `Func` callbacks for configuration. + +## What's Next + +- [Serialization](serialization.md) for configuring custom serializers. +- [AOT Deployments](aot.md) for native ahead-of-time publishing guidance. +- [Platform Compatibility](platform-compatibility.md) for API differences across target frameworks. diff --git a/tools/site_src/documentation/advanced/intro.md b/tools/site_src/documentation/advanced/intro.md index 5142eda87..bfd8ca64d 100644 --- a/tools/site_src/documentation/advanced/intro.md +++ b/tools/site_src/documentation/advanced/intro.md @@ -92,6 +92,7 @@ essentially sent back to back after they're picked up from internal queues and b ## What's Next +- [Dependency Injection](dependency-injection.md) explains the two DI packages (`NATS.Client.Hosting` and `NATS.Extensions.Microsoft.DependencyInjection`) and when to use each one. - [Slow Consumers](slow-consumers.md) explains how to detect and handle subscribers that can't keep up with the message rate, including the `MessageDropped` and `SlowConsumerDetected` events. - [Server Errors](server-errors.md) covers the `ServerError` event for observing `-ERR` messages from the server (permission denials, auth issues, server-side limits). - [Serialization](serialization.md) is the process of converting an object into a format that can be stored or transmitted. diff --git a/tools/site_src/documentation/advanced/serialization.md b/tools/site_src/documentation/advanced/serialization.md index a3dcf2c40..d1124c698 100644 --- a/tools/site_src/documentation/advanced/serialization.md +++ b/tools/site_src/documentation/advanced/serialization.md @@ -83,6 +83,22 @@ You can then use the custom serializer as the default for the connection: [!code-csharp[](../../../../tests/NATS.Net.DocsExamples/Advanced/SerializationPage.cs#custom)] +## Using Message Context in Serializers + +Serializers can opt into receiving message context (subject, reply-to, headers) by implementing +[`INatsSerializeWithContext`](xref:NATS.Client.Core.INatsSerializeWithContext`1) and/or +[`INatsDeserializeWithContext`](xref:NATS.Client.Core.INatsDeserializeWithContext`1). +These interfaces extend the base serialization interfaces, so existing serializers continue to work without changes. +When a context-aware serializer is detected, the library automatically dispatches to the context overload; +otherwise it falls back to the standard method. + +This can be used for scenarios like content-type negotiation, subject-based dispatch, encoding metadata, +or any other context-driven serialization logic. + +Here is an example of a serializer that writes a content-type header during serialization and uses it during deserialization: + +[!code-csharp[](../../../../tests/NATS.Net.DocsExamples/Advanced/SerializationPage.cs#header-aware-serializer)] + ## Using Multiple Serializers (chaining) You can also chain multiple serializers together to support multiple serialization formats. The first serializer in the diff --git a/tools/site_src/documentation/toc.yml b/tools/site_src/documentation/toc.yml index 44fc8566e..aac3bc2c7 100644 --- a/tools/site_src/documentation/toc.yml +++ b/tools/site_src/documentation/toc.yml @@ -36,6 +36,8 @@ - name: Advanced Options href: advanced/intro.md items: + - name: Dependency Injection + href: advanced/dependency-injection.md - name: Slow Consumers href: advanced/slow-consumers.md - name: Server Errors diff --git a/version.txt b/version.txt index 834f26295..7965b72a5 100644 --- a/version.txt +++ b/version.txt @@ -1 +1 @@ -2.8.0 +3.0.0-preview.6