More optimisation based on review

Author: thompson-tomo

src/NATS.Client.Core/Internal/InboxSub.cs

@@ -23,11 +23,11 @@ public InboxSub(
     }
 
     // Avoid base class error handling since inboxed subscribers will be responsible for that.
-    public override ValueTask ReceiveAsync(string subject, string? replyTo, ReadOnlySequence<byte>? headersBuffer, ReadOnlySequence<byte> payloadBuffer) =>
-        _inbox.ReceivedAsync(subject, replyTo, headersBuffer, payloadBuffer, _connection);
+    public override ValueTask ReceiveAsync(NatsProcessProps props, ReadOnlySequence<byte>? headersBuffer, ReadOnlySequence<byte> payloadBuffer) =>
+        _inbox.ReceivedAsync(props, headersBuffer, payloadBuffer, _connection);
 
     // Not used. Dummy implementation to keep base happy.
-    protected override ValueTask ReceiveInternalAsync(string subject, string? replyTo, ReadOnlySequence<byte>? headersBuffer, ReadOnlySequence<byte> payloadBuffer)
+    protected override ValueTask ReceiveInternalAsync(NatsProcessProps props, ReadOnlySequence<byte>? headersBuffer, ReadOnlySequence<byte> payloadBuffer)
         => default;
 
     protected override void TryComplete()
@@ -119,11 +119,11 @@ public ValueTask RegisterAsync(NatsSubBase sub)
         return sub.ReadyAsync();
     }
 
-    public async ValueTask ReceivedAsync(string subject, string? replyTo, ReadOnlySequence<byte>? headersBuffer, ReadOnlySequence<byte> payloadBuffer, NatsConnection connection)
+    public async ValueTask ReceivedAsync(NatsProcessProps props, ReadOnlySequence<byte>? headersBuffer, ReadOnlySequence<byte> payloadBuffer, NatsConnection connection)
     {
-        if (!_bySubject.TryGetValue(subject, out var subTable))
+        if (!_bySubject.TryGetValue(props.Subject, out var subTable))
         {
-            _logger.LogWarning(NatsLogEvents.InboxSubscription, "Unregistered message inbox received for {Subject}", subject);
+            _logger.LogWarning(NatsLogEvents.InboxSubscription, "Unregistered message inbox received for {Subject}", props.Subject);
             return;
         }
 
@@ -132,13 +132,13 @@ public async ValueTask ReceivedAsync(string subject, string? replyTo, ReadOnlySe
         {
             if (weakReference.TryGetTarget(out var sub))
             {
-                await sub.ReceiveAsync(subject, replyTo, headersBuffer, payloadBuffer).ConfigureAwait(false);
+                await sub.ReceiveAsync(props, headersBuffer, payloadBuffer).ConfigureAwait(false);
             }
         }
 #else
         foreach (var (sub, _) in subTable)
         {
-            await sub.ReceiveAsync(subject, replyTo, headersBuffer, payloadBuffer).ConfigureAwait(false);
+            await sub.ReceiveAsync(props, headersBuffer, payloadBuffer).ConfigureAwait(false);
         }
 #endif
     }

src/NATS.Client.Core/Internal/NatsReadProtocolProcessor.cs

@@ -252,8 +252,6 @@ private async Task ReadLoopAsync()
 
                         var props = ParseHMessageHeader(msgHeader);
 
-                        props.InboxPrefix = _connection.InboxPrefix;
-
                         if (_trace)
                         {
                             _logger.LogTrace(NatsLogEvents.Protocol, "HMSG trace parsed: {Subject} {Sid} {ReplyTo} {HeadersLength} {TotalLength}", props.Subject, props.SubscriptionId, props.ReplyTo, props.HeaderLength, props.TotalMessageLength);
@@ -478,15 +476,15 @@ private async ValueTask<ReadOnlySequence<byte>> DispatchCommandAsync(int code, R
     // MSG <subject> <sid> [reply-to] <#bytes>\r\n[payload]
     private NatsProcessProps ParseMessageHeader(ReadOnlySpan<byte> msgHeader)
     {
-        var metaLength = msgHeader.Length;
+        var framingLength = msgHeader.Length;
         msgHeader = msgHeader.Slice(4);
         msgHeader.Split(out var subjectBytes, out msgHeader);
         msgHeader.Split(out var sidBytes, out msgHeader);
         msgHeader.Split(out var replyToOrSizeBytes, out msgHeader);
 
-        var props = new NatsProcessProps(Encoding.ASCII.GetString(subjectBytes), GetInt32(sidBytes))
+        var props = new NatsProcessProps(Encoding.ASCII.GetString(subjectBytes), GetInt32(sidBytes), _connection.InboxPrefix)
         {
-            MetaLength = metaLength,
+            FramingLength = framingLength,
             HeaderLength = 0,
         };
 
@@ -531,17 +529,17 @@ private NatsProcessProps ParseMessageHeader(in ReadOnlySequence<byte> msgHeader)
     private NatsProcessProps ParseHMessageHeader(ReadOnlySpan<byte> msgHeader)
     {
         // 'HMSG' literal
-        var metaLength = msgHeader.Length;
+        var framingLength = msgHeader.Length;
         msgHeader.Split(out _, out msgHeader);
 
         msgHeader.Split(out var subjectBytes, out msgHeader);
         msgHeader.Split(out var sidBytes, out msgHeader);
         msgHeader.Split(out var replyToOrHeaderLenBytes, out msgHeader);
         msgHeader.Split(out var headerLenOrTotalLenBytes, out msgHeader);
 
-        var props = new NatsProcessProps(Encoding.ASCII.GetString(subjectBytes), GetInt32(sidBytes))
+        var props = new NatsProcessProps(Encoding.ASCII.GetString(subjectBytes), GetInt32(sidBytes), _connection.InboxPrefix)
         {
-            MetaLength = metaLength,
+            FramingLength = framingLength,
         };
 
         // We don't have the optional reply-to field

src/NATS.Client.Core/Internal/ReplyTask.cs

@@ -1,35 +1,30 @@
 using System.Buffers;
-using System.Buffers.Text;
 using System.Collections.Concurrent;
-using System.Text;
 
 namespace NATS.Client.Core.Internal;
 
 internal sealed class ReplyTask<T> : ReplyTaskBase, IDisposable
 {
     private readonly object _gate;
     private readonly ReplyTaskFactory _factory;
-    private readonly long _id;
+    private readonly long _subjectNum;
     private readonly NatsConnection _connection;
     private readonly INatsDeserialize<T> _deserializer;
     private readonly TimeSpan _requestTimeout;
     private readonly TaskCompletionSource _tcs;
     private NatsMsg<T> _msg;
 
-    public ReplyTask(ReplyTaskFactory factory, long id, string subject, NatsConnection connection, INatsDeserialize<T> deserializer, TimeSpan requestTimeout)
+    public ReplyTask(ReplyTaskFactory factory, long subjectNum, NatsConnection connection, INatsDeserialize<T> deserializer, TimeSpan requestTimeout)
     {
         _factory = factory;
-        _id = id;
-        Subject = subject;
+        _subjectNum = subjectNum;
         _connection = connection;
         _deserializer = deserializer;
         _requestTimeout = requestTimeout;
         _tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
         _gate = new object();
     }
 
-    public string Subject { get; }
-
     public async ValueTask<NatsMsg<T>> GetResultAsync(CancellationToken cancellationToken)
     {
         try
@@ -53,13 +48,13 @@ public override void SetResult(NatsProcessProps props, ReadOnlySequence<byte> pa
     {
         lock (_gate)
         {
-            _msg = NatsMsg<T>.Build(Subject, props.ReplyTo, headersBuffer, payload, _connection, _connection.HeaderParser, _deserializer);
+            _msg = NatsMsg<T>.Build(props.Subject, props.ReplyTo, headersBuffer, payload, _connection, _connection.HeaderParser, _deserializer);
         }
 
         _tcs.TrySetResult();
     }
 
-    public void Dispose() => _factory.Return(_id);
+    public void Dispose() => _factory.Return(_subjectNum);
 }
 
 internal abstract class ReplyTaskBase
@@ -69,23 +64,15 @@ internal abstract class ReplyTaskBase
 
 internal sealed class ReplyTaskFactory
 {
-    private readonly byte[] _inboxPrefix;
-    private readonly string _inboxPrefixString;
     private readonly NatsConnection _connection;
     private readonly ConcurrentDictionary<long, ReplyTaskBase> _replies;
     private readonly INatsSerializerRegistry _serializerRegistry;
     private readonly TimeSpan _requestTimeout;
-    private readonly int _subjectMaxLength;
-    private readonly bool _allocSubject;
     private long _nextId;
 
     public ReplyTaskFactory(NatsConnection connection)
     {
         _connection = connection;
-        _inboxPrefixString = _connection.InboxPrefix + ".";
-        _inboxPrefix = Encoding.UTF8.GetBytes(_inboxPrefixString);
-        _subjectMaxLength = _inboxPrefix.Length + 20; // 20 digits for long
-        _allocSubject = _subjectMaxLength < 128;
         _serializerRegistry = _connection.Opts.SerializerRegistry;
         _requestTimeout = _connection.Opts.RequestTimeout;
         _replies = new ConcurrentDictionary<long, ReplyTaskBase>();
@@ -95,29 +82,7 @@ public ReplyTask<TReply> CreateReplyTask<TReply>(INatsDeserialize<TReply>? deser
     {
         deserializer ??= _serializerRegistry.GetDeserializer<TReply>();
         var id = Interlocked.Increment(ref _nextId);
-
-        string subject;
-        if (_allocSubject)
-        {
-            Span<byte> buffer = stackalloc byte[_subjectMaxLength];
-            _inboxPrefix.CopyTo(buffer);
-            var idSpan = buffer.Slice(_inboxPrefix.Length);
-            if (Utf8Formatter.TryFormat(id, idSpan, out var written))
-            {
-                var subjectSpan = buffer.Slice(0, written + _inboxPrefix.Length);
-                subject = Encoding.UTF8.GetString(subjectSpan);
-            }
-            else
-            {
-                subject = _inboxPrefixString + id;
-            }
-        }
-        else
-        {
-            subject = _inboxPrefixString + id;
-        }
-
-        var rt = new ReplyTask<TReply>(this, id, subject, _connection, deserializer, requestTimeout ?? _requestTimeout);
+        var rt = new ReplyTask<TReply>(this, id, _connection, deserializer, requestTimeout ?? _requestTimeout);
         _replies.TryAdd(id, rt);
         return rt;
     }
@@ -126,7 +91,7 @@ public ReplyTask<TReply> CreateReplyTask<TReply>(INatsDeserialize<TReply>? deser
 
     public bool TrySetResult(NatsProcessProps props, in ReadOnlySequence<byte> payloadBuffer, in ReadOnlySequence<byte>? headersBuffer)
     {
-        if (_replies.TryGetValue(props.InboxId, out var rt))
+        if (_replies.TryGetValue(props.SubjectNumber, out var rt))
         {
             rt.SetResult(props, payloadBuffer, headersBuffer);
             return true;

src/NATS.Client.Core/Internal/SocketReader.cs

@@ -64,7 +64,7 @@ public async ValueTask<ReadOnlySequence<byte>> ReadAtLeastAsync(int minimumSize)
             }
 
             totalRead += read;
-            Telemetry.RecordReceivedBytes(read, Activity.Current.TagObjects.ToArray());
+            Telemetry.RecordReceivedBytes(read, Activity.Current.TagObjects.ToList());
             _seqeunceBuilder.Append(_availableMemory.Slice(0, read));
             _availableMemory = _availableMemory.Slice(read);
         }
@@ -110,7 +110,7 @@ public async ValueTask<ReadOnlySequence<byte>> ReadUntilReceiveNewLineAsync()
                 throw ex;
             }
 
-            Telemetry.RecordReceivedBytes(read, Activity.Current.TagObjects.ToArray());
+            Telemetry.RecordReceivedBytes(read, Activity.Current.TagObjects.ToList());
             var appendMemory = _availableMemory.Slice(0, read);
             _seqeunceBuilder.Append(appendMemory);
             _availableMemory = _availableMemory.Slice(read);

src/NATS.Client.Core/Internal/SubscriptionManager.cs

@@ -1,7 +1,6 @@
 using System.Buffers;
 using System.Collections.Concurrent;
 using System.Runtime.CompilerServices;
-using System.Threading;
 using Microsoft.Extensions.Logging;
 using NATS.Client.Core.Commands;
 
@@ -42,7 +41,7 @@ public SubscriptionManager(NatsConnection connection, string inboxPrefix)
         _cleanupInterval = _connection.Opts.SubscriptionCleanUpInterval;
         _timer = Task.Run(CleanupAsync);
         InboxSubBuilder = new InboxSubBuilder(connection.Opts.LoggerFactory.CreateLogger<InboxSubBuilder>());
-        _inboxSubSentinel = new InboxSub(InboxSubBuilder, new NatsSubscriptionProps(nameof(_inboxSubSentinel)), default, connection, this);
+        _inboxSubSentinel = new InboxSub(InboxSubBuilder, new NatsSubscriptionProps(nameof(_inboxSubSentinel), _connection.InboxPrefix), default, connection, this);
         _inboxSub = _inboxSubSentinel;
     }
 
@@ -62,7 +61,7 @@ public ValueTask PublishToClientHandlersAsync(NatsProcessProps props, in ReadOnl
             _logger.LogTrace(NatsLogEvents.Subscription, "Received subscription data for {Subject}/{Sid}", props.Subject, props.SubscriptionId);
         }
 
-        int? orphanSid = null;
+        var orphan = false;
         lock (_gate)
         {
             if (_bySid.TryGetValue(props.SubscriptionId, out var sidMetadata))
@@ -75,12 +74,12 @@ public ValueTask PublishToClientHandlersAsync(NatsProcessProps props, in ReadOnl
                         _logger.LogTrace(NatsLogEvents.Subscription, "Found subscription handler for {Subject}/{Sid}", props.Subject, props.SubscriptionId);
                     }
 
-                    return sub.ReceiveAsync(props.Subject, props.ReplyTo, headersBuffer, payloadBuffer);
+                    return sub.ReceiveAsync(props, headersBuffer, payloadBuffer);
                 }
                 else
                 {
                     _logger.LogWarning(NatsLogEvents.Subscription, "Subscription GCd but was never disposed {Subject}/{Sid}", props.Subject, props.SubscriptionId);
-                    orphanSid = props.SubscriptionId;
+                    orphan = true;
                 }
             }
             else
@@ -89,11 +88,11 @@ public ValueTask PublishToClientHandlersAsync(NatsProcessProps props, in ReadOnl
             }
         }
 
-        if (orphanSid != null)
+        if (orphan)
         {
             try
             {
-                return _connection.UnsubscribeAsync(new NatsSubscriptionProps(orphanSid.Value));
+                return _connection.UnsubscribeAsync(new NatsSubscriptionProps(props.SubscriptionId, props.InboxPrefix));
             }
             catch (Exception e)
             {
@@ -148,7 +147,7 @@ public ValueTask RemoveAsync(NatsSubBase sub)
             _logger.LogDebug(NatsLogEvents.Subscription, "Removing subscription {Subject}/{Sid}", sub.Subject, subMetadata.Sid);
         }
 
-        return _connection.UnsubscribeAsync(sub.Props);
+        return _connection.UnsubscribeAsync(sub.SubscriptionProps(_connection.InboxPrefix));
     }
 
     /// <summary>
@@ -183,7 +182,7 @@ internal async ValueTask WriteReconnectCommandsAsync(CommandWriter commandWriter
 
         foreach (var (sub, sid) in subs)
         {
-            await sub.WriteReconnectCommandsAsync(commandWriter, new NatsSubscriptionProps(sid)).ConfigureAwait(false);
+            await sub.WriteReconnectCommandsAsync(commandWriter, new NatsSubscriptionProps(sid, _connection.InboxPrefix)).ConfigureAwait(false);
 
             if (_debug)
             {
@@ -208,18 +207,14 @@ internal async Task InitializeInboxSubscriptionAsync(CancellationToken cancellat
             {
                 if (Interlocked.CompareExchange(ref _inboxSub, _inboxSubSentinel, _inboxSubSentinel) == _inboxSubSentinel)
                 {
-                    var inboxSubject = new NatsSubscriptionProps($"{_inboxPrefix}.*");
+                    var inboxSubject = new NatsSubscriptionProps($"{_inboxPrefix}.*", _connection.InboxPrefix);
 
                     // We need to subscribe to the real inbox subject before we can register the internal subject.
                     // We use 'default' options here since options provided by the user are for the internal subscription.
                     // For example if the user provides a timeout, we don't want to timeout the real inbox subscription
                     // since it must live duration of the connection.
                     _inboxSub = InboxSubBuilder.Build(inboxSubject, opts: default, _connection, manager: this);
-                    var props = new NatsSubscriptionProps(_inboxSub.Subject)
-                    {
-                        InboxPrefix = _connection.InboxPrefix,
-                        QueueGroup = _inboxSub.QueueGroup,
-                    };
+                    var props = new NatsSubscriptionProps(_inboxSub.Subject, _connection.InboxPrefix, _inboxSub.QueueGroup);
                     await SubscribeQueueAsync(props, _inboxSub, cancellationToken).ConfigureAwait(false);
                 }
             }
@@ -233,11 +228,7 @@ internal async Task InitializeInboxSubscriptionAsync(CancellationToken cancellat
     private ValueTask SubscribeInternalAsync(NatsSubBase sub, CancellationToken cancellationToken)
     {
         var start = DateTimeOffset.UtcNow;
-        var props = new NatsSubscriptionProps(sub.Subject)
-        {
-            InboxPrefix = _connection.InboxPrefix,
-            QueueGroup = sub.QueueGroup,
-        };
+        var props = new NatsSubscriptionProps(sub.Subject, _connection.InboxPrefix, sub.QueueGroup);
         var tags = Telemetry.GetTags(_connection.ServerInfo, props);
         using var activity = Telemetry.StartActivity(start, props, _connection.ServerInfo, Telemetry.Constants.SubscribeActivityName, tags);
         ValueTask task;
@@ -364,7 +355,7 @@ private async ValueTask UnsubscribeSidsAsync(List<int> sids)
             try
             {
                 _logger.LogWarning(NatsLogEvents.Subscription, "Unsubscribing orphan subscription {Sid}", sid);
-                await _connection.UnsubscribeAsync(new NatsSubscriptionProps(sid)).ConfigureAwait(false);
+                await _connection.UnsubscribeAsync(new NatsSubscriptionProps(sid, _connection.InboxPrefix)).ConfigureAwait(false);
             }
             catch (Exception e)
             {

src/NATS.Client.Core/NatsConnection.Publish.cs

@@ -14,10 +14,7 @@ public ValueTask PublishAsync(string subject, NatsHeaders? headers = default, st
     public ValueTask PublishAsync<T>(string subject, T? data, NatsHeaders? headers = default, string? replyTo = default, INatsSerialize<T>? serializer = default, NatsPubOpts? opts = default, CancellationToken cancellationToken = default)
     {
         serializer ??= Opts.SerializerRegistry.GetSerializer<T>();
-        var props = new NatsPublishProps(subject)
-        {
-            InboxPrefix = InboxPrefix,
-        };
+        var props = new NatsPublishProps(subject, InboxPrefix);
         props.SetReplyTo(replyTo);
         return PublishInternalAsync(props, serializer, data, headers, default, cancellationToken);
     }

src/NATS.Client.Core/NatsConnection.RequestReply.cs

@@ -74,8 +74,7 @@ private async ValueTask<NatsMsg<TReply>> RequestInternalAsync<TRequest, TReply>(
         NatsSubOpts? replyOpts = default,
         CancellationToken cancellationToken = default)
     {
-        props ??= new NatsPublishProps(subject);
-        props.InboxPrefix ??= InboxPrefix;
+        props ??= new NatsPublishProps(subject, InboxPrefix);
 
         var start = DateTimeOffset.UtcNow;
         var tags = Telemetry.GetTags(ServerInfo, props);