Use required members to improve the code clarity

sakno · sakno · commit 9f7e4f6f881c · 2025-08-22T23:01:37.000+03:00
diff --git a/src/cluster/DotNext.Net.Cluster/Net/Multiplexing/InputMultiplexer.cs b/src/cluster/DotNext.Net.Cluster/Net/Multiplexing/InputMultiplexer.cs
@@ -1,5 +1,4 @@
 using System.Collections.Concurrent;
-using System.Diagnostics;
 using System.Net.Sockets;
 using System.Runtime.CompilerServices;
 
@@ -8,50 +7,61 @@ namespace DotNext.Net.Multiplexing;
 using Buffers;
 using Threading;
 
-internal sealed class InputMultiplexer<T>(
-    ConcurrentDictionary<uint, MultiplexedStream> streams,
-    AsyncAutoResetEvent writeSignal,
-    BufferWriter<byte> framingBuffer,
-    int flushThreshold,
-    in TagList measurementTags,
-    TimeSpan timeout,
-    TimeSpan heartbeatTimeout,
-    CancellationToken token) : Multiplexer<T>(streams, new ConcurrentQueue<ProtocolCommand>(), measurementTags, token)
+internal sealed class InputMultiplexer<T>() : Multiplexer<T>(new(), new ConcurrentQueue<ProtocolCommand>())
     where T : IStreamMetrics
 {
-    
-    public TimeSpan Timeout => timeout;
+    public required TimeSpan Timeout { get; init; }
+
+    public required TimeSpan HeartbeatTimeout { private get; init; }
+
+    public required int FlushThreshold { private get; init; }
+
+    public required BufferWriter<byte> FramingBuffer { private get; init; }
+
+    public required AsyncAutoResetEvent TransportSignal { private get; init; }
 
     public bool TryAddStream(uint streamId, MultiplexedStream stream)
     {
-        var result = streams.TryAdd(streamId, stream);
+        var result = Streams.TryAdd(streamId, stream);
         ChangeStreamCount(Unsafe.BitCast<bool, byte>(result));
         return result;
     }
 
     public bool TryRemoveStream(uint streamId, MultiplexedStream stream)
     {
-        var removed = streams.TryRemove(new(streamId, stream));
+        var removed = Streams.TryRemove(new(streamId, stream));
         ChangeStreamCount(-Unsafe.BitCast<bool, byte>(removed));
         return removed;
     }
 
-    public OutputMultiplexer<T> CreateOutput(Memory<byte> framingBuffer, TimeSpan receiveTimeout)
-        => new(streams, writeSignal, commands, framingBuffer, measurementTags, receiveTimeout, RootToken);
+    public OutputMultiplexer<T> CreateOutput(Memory<byte> framingBuffer, TimeSpan receiveTimeout) => new(Streams, Commands)
+    {
+        MeasurementTags = MeasurementTags,
+        RootToken = RootToken,
+        FramingBuffer = framingBuffer,
+        Timeout = receiveTimeout,
+        TransportSignal = TransportSignal,
+    };
 
     public OutputMultiplexer<T> CreateOutput(Memory<byte> framingBuffer, TimeSpan receiveTimeout, MultiplexedStreamFactory handlerFactory,
-        CancellationToken token)
-        => new(streams, writeSignal, commands, framingBuffer, measurementTags, receiveTimeout, token)
-            { Factory = handlerFactory };
+        CancellationToken token) => new(Streams, Commands)
+    {
+        MeasurementTags = MeasurementTags,
+        RootToken = token,
+        FramingBuffer = framingBuffer,
+        Timeout = receiveTimeout,
+        TransportSignal = TransportSignal,
+        Factory = handlerFactory,
+    };
 
     public async Task ProcessAsync(Func<bool> condition, Socket socket)
     {
-        using var enumerator = streams.GetEnumerator();
+        using var enumerator = Streams.GetEnumerator();
         for (var requiresHeartbeat = false;
              condition();
-             requiresHeartbeat = !await writeSignal.WaitAsync(heartbeatTimeout, RootToken).ConfigureAwait(false))
+             requiresHeartbeat = !await TransportSignal.WaitAsync(HeartbeatTimeout, RootToken).ConfigureAwait(false))
         {
-            framingBuffer.Clear(reuseBuffer: true);
+            FramingBuffer.Clear(reuseBuffer: true);
 
             // combine streams
             while (enumerator.MoveNext())
@@ -60,33 +70,33 @@ public async Task ProcessAsync(Func<bool> condition, Socket socket)
 
                 if (stream.IsCompleted && TryRemoveStream(streamId, stream))
                 {
-                    Protocol.WriteStreamClosed(framingBuffer, streamId);
+                    Protocol.WriteStreamClosed(FramingBuffer, streamId);
                 }
                 else
                 {
-                    await stream.WriteFrameAsync(framingBuffer, streamId).ConfigureAwait(false);
+                    await stream.WriteFrameAsync(FramingBuffer, streamId).ConfigureAwait(false);
                 }
 
                 // write the buffer on overflow
-                if (framingBuffer.WrittenCount >= flushThreshold)
+                if (FramingBuffer.WrittenCount >= FlushThreshold)
                 {
-                    await SendAsync(framingBuffer.WrittenMemory, socket).ConfigureAwait(false);
-                    framingBuffer.Clear(reuseBuffer: true);
+                    await SendAsync(FramingBuffer.WrittenMemory, socket).ConfigureAwait(false);
+                    FramingBuffer.Clear(reuseBuffer: true);
                 }
             }
 
             // process protocol commands
-            commands.Serialize(framingBuffer);
+            Commands.Serialize(FramingBuffer);
 
-            switch (framingBuffer.WrittenCount)
+            switch (FramingBuffer.WrittenCount)
             {
                 case 0 when requiresHeartbeat:
-                    Protocol.WriteHeartbeat(framingBuffer);
+                    Protocol.WriteHeartbeat(FramingBuffer);
                     goto default;
                 case 0:
                     break;
                 default:
-                    await SendAsync(framingBuffer.WrittenMemory, socket).ConfigureAwait(false);
+                    await SendAsync(FramingBuffer.WrittenMemory, socket).ConfigureAwait(false);
                     break;
             }
 
@@ -99,7 +109,7 @@ private async ValueTask SendAsync(ReadOnlyMemory<byte> buffer, Socket socket)
     {
         for (int bytesWritten; !buffer.IsEmpty; buffer = buffer.Slice(bytesWritten))
         {
-            StartOperation(timeout);
+            StartOperation(Timeout);
             try
             {
                 bytesWritten = await socket.SendAsync(buffer, TimeBoundedToken).ConfigureAwait(false);
@@ -121,9 +131,9 @@ private async ValueTask SendAsync(ReadOnlyMemory<byte> buffer, Socket socket)
 
     public async ValueTask CompleteAllAsync(Exception e)
     {
-        foreach (var id in streams.Keys)
+        foreach (var id in Streams.Keys)
         {
-            if (streams.TryRemove(id, out var stream))
+            if (Streams.TryRemove(id, out var stream))
             {
                 await stream.CompleteTransportOutputAsync(e).ConfigureAwait(false);
                 await stream.CompleteTransportInputAsync(e).ConfigureAwait(false);
diff --git a/src/cluster/DotNext.Net.Cluster/Net/Multiplexing/MultiplexedClient.cs b/src/cluster/DotNext.Net.Cluster/Net/Multiplexing/MultiplexedClient.cs
@@ -34,14 +34,17 @@ protected MultiplexedClient(Options configuration)
         var flushThreshold = configuration.BufferCapacity;
         framingBuffer = new PoolingBufferWriter<byte>(configuration.ToAllocator()) { Capacity = flushThreshold };
 
-        input = new(new(),
-            writeSignal,
-            framingBuffer,
-            flushThreshold,
-            configuration.MeasurementTags,
-            configuration.Timeout,
-            configuration.HeartbeatTimeout,
-            lifetimeToken);
+        input = new()
+        {
+            TransportSignal = writeSignal,
+            FramingBuffer = framingBuffer,
+            FlushThreshold = flushThreshold,
+            MeasurementTags = configuration.MeasurementTags,
+            Timeout = configuration.Timeout,
+            HeartbeatTimeout = configuration.HeartbeatTimeout,
+            RootToken = lifetimeToken,
+        };
+        
         output = input.CreateOutput(GC.AllocateArray<byte>(configuration.BufferCapacity, pinned: true), configuration.Timeout);
     }
 
diff --git a/src/cluster/DotNext.Net.Cluster/Net/Multiplexing/MultiplexedListener.Dispatcher.cs b/src/cluster/DotNext.Net.Cluster/Net/Multiplexing/MultiplexedListener.Dispatcher.cs
@@ -30,15 +30,16 @@ private async Task DispatchAsync(Socket socket)
         var writeSignal = new AsyncAutoResetEvent(initialState: false);
         var receiveBuffer = allocator(flushThreshold);
         var framingBuffer = new PoolingBufferWriter<byte>(allocator) { Capacity = flushThreshold };
-        var input = new InputMultiplexer<MultiplexedListener>(
-            new(),
-            writeSignal,
-            framingBuffer,
-            flushThreshold,
-            CreateMeasurementTags(socket),
-            timeout,
-            heartbeatTimeout,
-            lifetimeToken);
+        var input = new InputMultiplexer<MultiplexedListener>
+        {
+            TransportSignal = writeSignal,
+            FramingBuffer = framingBuffer,
+            MeasurementTags = CreateMeasurementTags(socket),
+            FlushThreshold = flushThreshold,
+            Timeout = timeout,
+            HeartbeatTimeout = heartbeatTimeout,
+            RootToken = lifetimeToken,
+        };
         var receiveTokenSource = CancellationTokenSource.CreateLinkedTokenSource(lifetimeToken);
         var output = input.CreateOutput(
             receiveBuffer.Memory,
diff --git a/src/cluster/DotNext.Net.Cluster/Net/Multiplexing/Multiplexer.TimeoutControl.cs b/src/cluster/DotNext.Net.Cluster/Net/Multiplexing/Multiplexer.TimeoutControl.cs
@@ -1,14 +1,21 @@
+using System.Diagnostics.CodeAnalysis;
+
 namespace DotNext.Net.Multiplexing;
 
 using Threading;
 
-partial class Multiplexer<T>
+partial class Multiplexer
 {
-    private TimeoutSource source = new(TimeProvider.System, token);
+    private TimeoutSource source;
 
-    protected void StartOperation(TimeSpan timeout) => source.TryStart(timeout);
+    public required CancellationToken RootToken
+    {
+        get => source.RootToken;
 
-    public CancellationToken RootToken => source.RootToken;
+        [MemberNotNull(nameof(source))] init => source = new(TimeProvider.System, value);
+    }
+
+    protected void StartOperation(TimeSpan timeout) => source.TryStart(timeout);
 
     protected CancellationToken TimeBoundedToken => source.Token;
 
diff --git a/src/cluster/DotNext.Net.Cluster/Net/Multiplexing/Multiplexer.cs b/src/cluster/DotNext.Net.Cluster/Net/Multiplexing/Multiplexer.cs
@@ -1,33 +1,36 @@
 using System.Collections.Concurrent;
 using System.Diagnostics;
-using System.Diagnostics.Metrics;
 
 namespace DotNext.Net.Multiplexing;
 
-internal abstract partial class Multiplexer<T>(
+internal abstract partial class Multiplexer(
     ConcurrentDictionary<uint, MultiplexedStream> streams,
-    IProducerConsumerCollection<ProtocolCommand> commands,
-    in TagList measurementTags,
-    CancellationToken token) : Disposable
-    where T : IStreamMetrics
+    IProducerConsumerCollection<ProtocolCommand> commands) : Disposable
 {
-    protected readonly TagList measurementTags = measurementTags;
-    protected readonly ConcurrentDictionary<uint, MultiplexedStream> streams = streams;
-    protected readonly IProducerConsumerCollection<ProtocolCommand> commands = commands;
-
-    protected void ChangeStreamCount(long delta = 1) => T.ChangeStreamCount(delta, measurementTags);
+    protected readonly IProducerConsumerCollection<ProtocolCommand> Commands = commands;
+    protected readonly ConcurrentDictionary<uint, MultiplexedStream> Streams = streams;
 
     protected override void Dispose(bool disposing)
     {
         if (disposing)
         {
             source.Dispose();
         }
-        
+
         base.Dispose(disposing);
     }
 
     protected override ValueTask DisposeAsyncCore() => source.DisposeAsync();
 
     public new ValueTask DisposeAsync() => base.DisposeAsync();
+}
+
+internal abstract class Multiplexer<T>(
+    ConcurrentDictionary<uint, MultiplexedStream> streams,
+    IProducerConsumerCollection<ProtocolCommand> commands) : Multiplexer(streams, commands)
+    where T : IStreamMetrics
+{
+    public required TagList MeasurementTags;
+
+    protected void ChangeStreamCount(long delta = 1) => T.ChangeStreamCount(delta, MeasurementTags);
 }
diff --git a/src/cluster/DotNext.Net.Cluster/Net/Multiplexing/OutputMultiplexer.cs b/src/cluster/DotNext.Net.Cluster/Net/Multiplexing/OutputMultiplexer.cs
@@ -1,5 +1,4 @@
 using System.Collections.Concurrent;
-using System.Diagnostics;
 using System.Net.Sockets;
 
 namespace DotNext.Net.Multiplexing;
@@ -8,14 +7,20 @@ namespace DotNext.Net.Multiplexing;
 
 internal sealed class OutputMultiplexer<T>(
     ConcurrentDictionary<uint, MultiplexedStream> streams,
-    AsyncAutoResetEvent writeSignal,
-    IProducerConsumerCollection<ProtocolCommand> commands,
-    Memory<byte> framingBuffer,
-    in TagList measurementTags,
-    TimeSpan timeout,
-    CancellationToken token) : Multiplexer<T>(streams, commands, measurementTags, token)
+    IProducerConsumerCollection<ProtocolCommand> commands): Multiplexer<T>(streams, commands)
     where T : IStreamMetrics
 {
+    private readonly Memory<byte> framingBuffer;
+    
+    public required AsyncAutoResetEvent TransportSignal { private get; init; }
+
+    public required Memory<byte> FramingBuffer
+    {
+        init => framingBuffer = value;
+    }
+
+    public required TimeSpan Timeout { private get; init; }
+
     public MultiplexedStreamFactory? Factory { get; init; }
 
     public Task ProcessAsync(Socket socket)
@@ -24,7 +29,7 @@ public Task ProcessAsync(Socket socket)
         
         // if output multiplexer is completed due to exception, we need to trigger
         // the input multiplexer to handle the error
-        task.ConfigureAwait(false).GetAwaiter().UnsafeOnCompleted(writeSignal.SetNoResult);
+        task.ConfigureAwait(false).GetAwaiter().UnsafeOnCompleted(TransportSignal.SetNoResult);
         return task;
     }
 
@@ -33,7 +38,7 @@ private async Task ProcessCoreAsync(Socket socket)
         FrameHeader header;
         for (var bufferedBytes = 0;; AdjustFramingBuffer(ref bufferedBytes, header, framingBuffer.Span))
         {
-            StartOperation(timeout); // resumed by heartbeat
+            StartOperation(Timeout); // resumed by heartbeat
             try
             {
                 // read at least header
@@ -66,21 +71,21 @@ private async Task ProcessCoreAsync(Socket socket)
             if (header.Control is FrameControl.Heartbeat)
                 continue;
 
-            if (!streams.TryGetValue(header.Id, out var stream))
+            if (!Streams.TryGetValue(header.Id, out var stream))
             {
                 if (Factory is null || header.CanBeIgnored)
                 {
                     continue;
                 }
 
-                if ((stream = Factory(writeSignal, measurementTags)) is null)
+                if ((stream = Factory(TransportSignal, MeasurementTags)) is null)
                 {
-                    commands.TryAdd(new StreamRejectedCommand(header.Id));
-                    writeSignal.Set();
+                    Commands.TryAdd(new StreamRejectedCommand(header.Id));
+                    TransportSignal.Set();
                     continue;
                 }
 
-                streams[header.Id] = stream;
+                Streams[header.Id] = stream;
                 ChangeStreamCount();
             }
             else if (stream.IsTransportOutputCompleted)
