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/NATS.Net.sln b/NATS.Net.sln index 0cd496046..55575bdec 100644 --- a/NATS.Net.sln +++ b/NATS.Net.sln @@ -158,6 +158,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NATS.Client.CheckAbiTransie EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NATS.Client.CheckAbi", "tests\NATS.Client.CheckAbi\NATS.Client.CheckAbi.csproj", "{DA8E9EC2-990C-4DA2-A3F3-7E7796C71DD8}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NATS.Client.CheckAbiOld", "tests\NATS.Client.CheckAbiOld\NATS.Client.CheckAbiOld.csproj", "{93E92151-28BF-4176-92F4-65E1227C3269}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -372,6 +374,10 @@ Global {DA8E9EC2-990C-4DA2-A3F3-7E7796C71DD8}.Debug|Any CPU.Build.0 = Debug|Any CPU {DA8E9EC2-990C-4DA2-A3F3-7E7796C71DD8}.Release|Any CPU.ActiveCfg = Release|Any CPU {DA8E9EC2-990C-4DA2-A3F3-7E7796C71DD8}.Release|Any CPU.Build.0 = Release|Any CPU + {93E92151-28BF-4176-92F4-65E1227C3269}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {93E92151-28BF-4176-92F4-65E1227C3269}.Debug|Any CPU.Build.0 = Debug|Any CPU + {93E92151-28BF-4176-92F4-65E1227C3269}.Release|Any CPU.ActiveCfg = Release|Any CPU + {93E92151-28BF-4176-92F4-65E1227C3269}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -432,6 +438,7 @@ Global {71C2BC96-C104-4748-B6CF-E74BAC4E709D} = {4827B3EC-73D8-436D-AE2A-5E29AC95FD0C} {21649885-B40A-4F6B-9286-A7E378276C3E} = {C526E8AB-739A-48D7-8FC4-048978C9B650} {DA8E9EC2-990C-4DA2-A3F3-7E7796C71DD8} = {C526E8AB-739A-48D7-8FC4-048978C9B650} + {93E92151-28BF-4176-92F4-65E1227C3269} = {C526E8AB-739A-48D7-8FC4-048978C9B650} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {8CBB7278-D093-448E-B3DE-B5991209A1AA} 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..364faf241 100644 --- a/src/NATS.Client.Core/Commands/CommandWriter.cs +++ b/src/NATS.Client.Core/Commands/CommandWriter.cs @@ -334,12 +334,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) { diff --git a/src/NATS.Client.Core/Internal/Telemetry.cs b/src/NATS.Client.Core/Internal/Telemetry.cs index 02944bfba..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; }); } 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/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/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/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/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