Skip to content

Commit 267ca7b

Browse files
tgiachitgiachi
committed
merge: per-connection transport codec in SquidStd.Network
Add ITransportCodec (in-place, length-preserving transport transform) with an atomic mid-connection SwapCodec, a ConnectionPipeline descriptor, and a per-connection connectionPipelineFactory on SquidTcpServer. Codec is applied innermost (decode on receive before middleware, encode on send under the send lock). Fully additive and backward compatible; receive history stays in sync with decoded output. 97/97 Network tests green.
2 parents f6e5912 + d4a5187 commit 267ca7b

9 files changed

Lines changed: 437 additions & 11 deletions

File tree

‎src/SquidStd.Network/Client/SquidStdTcpClient.cs‎

Lines changed: 45 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
using SquidStd.Network.Buffers;
66
using SquidStd.Network.Data.Events;
77
using SquidStd.Network.Interfaces.Client;
8+
using SquidStd.Network.Interfaces.Codecs;
89
using SquidStd.Network.Interfaces.Framing;
910
using SquidStd.Network.Interfaces.Middleware;
1011
using SquidStd.Network.Pipeline;
@@ -32,6 +33,7 @@ public sealed class SquidStdTcpClient : INetworkConnection, IAsyncDisposable, ID
3233
private readonly Stream _stream;
3334
private static long _sessionIdSequence;
3435
private int _closed;
36+
private ITransportCodec? _codec;
3537

3638
private CancellationTokenRegistration _externalCancellationTokenRegistration;
3739
private byte[]? _pendingBuffer;
@@ -153,13 +155,15 @@ public SquidStdTcpClient(
153155
Socket socket,
154156
IEnumerable<INetMiddleware>? middlewares = null,
155157
INetFramer? framer = null,
158+
ITransportCodec? codec = null,
156159
int receiveBufferSize = DefaultReceiveBufferSize,
157160
int historyBufferCapacity = DefaultHistoryBufferCapacity
158161
) : this(
159162
socket,
160163
new NetworkStream(socket, false),
161164
middlewares,
162165
framer,
166+
codec,
163167
receiveBufferSize,
164168
historyBufferCapacity
165169
) { }
@@ -172,6 +176,7 @@ public SquidStdTcpClient(
172176
Stream stream,
173177
IEnumerable<INetMiddleware>? middlewares = null,
174178
INetFramer? framer = null,
179+
ITransportCodec? codec = null,
175180
int receiveBufferSize = DefaultReceiveBufferSize,
176181
int historyBufferCapacity = DefaultHistoryBufferCapacity
177182
)
@@ -183,6 +188,7 @@ public SquidStdTcpClient(
183188
_stream = stream;
184189
_middlewarePipeline = new(middlewares);
185190
_framer = framer;
191+
_codec = codec;
186192
_receiveBuffer = new(historyBufferCapacity);
187193
ReceiveBufferSize = receiveBufferSize;
188194
SessionId = Interlocked.Increment(ref _sessionIdSequence);
@@ -246,13 +252,14 @@ public static async Task<SquidStdTcpClient> ConnectAsync(
246252
IPEndPoint endPoint,
247253
IEnumerable<INetMiddleware>? middlewares = null,
248254
INetFramer? framer = null,
255+
ITransportCodec? codec = null,
249256
CancellationToken cancellationToken = default
250257
)
251258
{
252259
var socket = new Socket(endPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
253260
await socket.ConnectAsync(endPoint, cancellationToken);
254261

255-
var client = new SquidStdTcpClient(socket, middlewares, framer);
262+
var client = new SquidStdTcpClient(socket, middlewares, framer, codec);
256263
await client.StartAsync(cancellationToken);
257264

258265
return client;
@@ -346,6 +353,14 @@ public byte[] PeekData(int count = 0)
346353
}
347354
}
348355

356+
/// <summary>
357+
/// Atomically swaps the transport codec for this connection. The new codec takes effect from the next
358+
/// socket read; the caller must trigger the swap at a read boundary (no old-regime bytes still pending).
359+
/// </summary>
360+
/// <param name="codec">The new codec, or null to remove transport transformation.</param>
361+
public void SwapCodec(ITransportCodec? codec)
362+
=> Volatile.Write(ref _codec, codec);
363+
349364
/// <summary>
350365
/// Removes all middleware components of the specified type from this client pipeline.
351366
/// </summary>
@@ -374,7 +389,28 @@ public async Task SendAsync(ReadOnlyMemory<byte> payload, CancellationToken canc
374389

375390
try
376391
{
377-
await _stream.WriteAsync(processedPayload, cancellationToken);
392+
var codec = Volatile.Read(ref _codec);
393+
394+
if (codec is null)
395+
{
396+
await _stream.WriteAsync(processedPayload, cancellationToken);
397+
}
398+
else
399+
{
400+
var sendBuffer = ArrayPool<byte>.Shared.Rent(processedPayload.Length);
401+
402+
try
403+
{
404+
processedPayload.Span.CopyTo(sendBuffer);
405+
codec.Encode(sendBuffer.AsSpan(0, processedPayload.Length));
406+
await _stream.WriteAsync(sendBuffer.AsMemory(0, processedPayload.Length), cancellationToken);
407+
}
408+
finally
409+
{
410+
ArrayPool<byte>.Shared.Return(sendBuffer);
411+
}
412+
}
413+
378414
await _stream.FlushAsync(cancellationToken);
379415
}
380416
catch (Exception ex)
@@ -532,17 +568,19 @@ private async Task ReceiveLoopAsync()
532568
break;
533569
}
534570

535-
lock (_receiveBufferSync)
536-
{
537-
_receiveBuffer.PushBackRange(buffer.AsSpan(0, received));
538-
}
539-
540571
var chunk = ArrayPool<byte>.Shared.Rent(received);
541572

542573
try
543574
{
544575
buffer.AsSpan(0, received).CopyTo(chunk);
545576

577+
Volatile.Read(ref _codec)?.Decode(chunk.AsSpan(0, received));
578+
579+
lock (_receiveBufferSync)
580+
{
581+
_receiveBuffer.PushBackRange(chunk.AsSpan(0, received));
582+
}
583+
546584
var chunkMemory = new ReadOnlyMemory<byte>(chunk, 0, received);
547585
var processed = await _middlewarePipeline.ExecuteAsync(
548586
this,

‎src/SquidStd.Network/Data/ConnectionPipeline.cs‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
using SquidStd.Network.Interfaces.Codecs;
2+
using SquidStd.Network.Interfaces.Framing;
3+
using SquidStd.Network.Interfaces.Middleware;
4+
5+
namespace SquidStd.Network.Data;
6+
7+
/// <summary>
8+
/// Per-connection transport configuration produced by a server factory on each accepted connection.
9+
/// Any member left null falls back to the server's shared configuration.
10+
/// </summary>
11+
public sealed record ConnectionPipeline(
12+
ITransportCodec? Codec = null,
13+
IReadOnlyList<INetMiddleware>? Middlewares = null,
14+
INetFramer? Framer = null
15+
);

‎src/SquidStd.Network/Interfaces/Codecs/ITransportCodec.cs‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
namespace SquidStd.Network.Interfaces.Codecs;
2+
3+
/// <summary>
4+
/// An in-place, length-preserving transport transform applied at the wire — for example a stream cipher.
5+
/// </summary>
6+
/// <remarks>
7+
/// Implementations MUST preserve the buffer length and transform in place. <see cref="Decode" /> is invoked
8+
/// only from a connection's receive loop (serial), and <see cref="Encode" /> only from its send path (serial
9+
/// under the send lock). The two directions may run concurrently (one each), so an implementation MUST NOT
10+
/// share mutable state between decode and encode (real ciphers keep separate receive/send positions).
11+
/// </remarks>
12+
public interface ITransportCodec
13+
{
14+
/// <summary>Transforms inbound bytes in place, before the middleware pipeline.</summary>
15+
/// <param name="buffer">The bytes to transform in place.</param>
16+
void Decode(Span<byte> buffer);
17+
18+
/// <summary>Transforms outbound bytes in place, after the middleware pipeline.</summary>
19+
/// <param name="buffer">The bytes to transform in place.</param>
20+
void Encode(Span<byte> buffer);
21+
}

‎src/SquidStd.Network/Server/SquidTcpServer.cs‎

Lines changed: 17 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
using System.Net.Sockets;
55
using Serilog;
66
using SquidStd.Network.Client;
7+
using SquidStd.Network.Data;
78
using SquidStd.Network.Data.Events;
89
using SquidStd.Network.Data.Options;
910
using SquidStd.Network.Interfaces.Framing;
@@ -23,6 +24,7 @@ public sealed class SquidTcpServer : INetworkServer, IAsyncDisposable, IDisposab
2324
private readonly ConcurrentDictionary<long, SquidStdTcpClient> _clients = new();
2425
private readonly IPEndPoint _endPoint;
2526
private readonly INetFramer? _framer;
27+
private readonly Func<ConnectionPipeline>? _connectionPipelineFactory;
2628
private readonly int _historyBufferCapacity;
2729
private readonly SquidStdTcpServerTlsOptions? _tlsOptions;
2830

@@ -81,19 +83,26 @@ public sealed class SquidTcpServer : INetworkServer, IAsyncDisposable, IDisposab
8183
/// </param>
8284
/// <param name="receiveBufferSize">Per-client receive chunk size.</param>
8385
/// <param name="historyBufferCapacity">Per-client history buffer capacity.</param>
86+
/// <param name="connectionPipelineFactory">
87+
/// Optional factory invoked once per accepted connection to produce its transport configuration.
88+
/// It MUST return fresh per-connection state — in particular a new <c>ITransportCodec</c> instance per
89+
/// call — because codecs are stateful and must not be shared across connections.
90+
/// </param>
8491
public SquidTcpServer(
8592
IPEndPoint endPoint,
8693
INetFramer? framer = null,
8794
int receiveBufferSize = 8192,
8895
int historyBufferCapacity = 65536,
89-
SquidStdTcpServerTlsOptions? tlsOptions = null
96+
SquidStdTcpServerTlsOptions? tlsOptions = null,
97+
Func<ConnectionPipeline>? connectionPipelineFactory = null
9098
)
9199
{
92100
_endPoint = endPoint;
93101
_framer = framer;
94102
_receiveBufferSize = receiveBufferSize;
95103
_historyBufferCapacity = historyBufferCapacity;
96104
_tlsOptions = tlsOptions;
105+
_connectionPipelineFactory = connectionPipelineFactory;
97106
}
98107

99108
/// <summary>
@@ -212,12 +221,16 @@ private async Task AcceptLoopAsync()
212221
var clientSocket = await serverSocket.AcceptAsync(cts.Token);
213222
var clientStream = await CreateClientStreamAsync(clientSocket, cts.Token).ConfigureAwait(false);
214223

215-
var middlewareSnapshot = _middlewares;
224+
var pipeline = _connectionPipelineFactory?.Invoke();
225+
var middlewares = pipeline?.Middlewares ?? _middlewares;
226+
var framer = pipeline?.Framer ?? _framer;
227+
var codec = pipeline?.Codec;
216228
var client = new SquidStdTcpClient(
217229
clientSocket,
218230
clientStream,
219-
middlewareSnapshot,
220-
_framer,
231+
middlewares,
232+
framer,
233+
codec,
221234
_receiveBufferSize,
222235
_historyBufferCapacity
223236
);

‎tests/SquidStd.Tests/Network/ConnectionPipelineTests.cs‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
using SquidStd.Network.Data;
2+
using SquidStd.Tests.Support;
3+
4+
namespace SquidStd.Tests.Network;
5+
6+
public class ConnectionPipelineTests
7+
{
8+
[Fact]
9+
public void Defaults_AreNull()
10+
{
11+
var pipeline = new ConnectionPipeline();
12+
13+
Assert.Null(pipeline.Codec);
14+
Assert.Null(pipeline.Middlewares);
15+
Assert.Null(pipeline.Framer);
16+
}
17+
18+
[Fact]
19+
public void Positional_SetsCodec()
20+
{
21+
var codec = new CountingXorCodec(1);
22+
23+
var pipeline = new ConnectionPipeline(codec);
24+
25+
Assert.Same(codec, pipeline.Codec);
26+
}
27+
}

‎tests/SquidStd.Tests/Network/TransportCodecFakeTests.cs‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
using SquidStd.Tests.Support;
2+
3+
namespace SquidStd.Tests.Network;
4+
5+
public class TransportCodecFakeTests
6+
{
7+
[Fact]
8+
public void EncodeThenDecode_WithMatchingCodecs_RestoresOriginal()
9+
{
10+
var original = new byte[] { 10, 20, 30, 40, 50 };
11+
var buffer = (byte[])original.Clone();
12+
13+
new CountingXorCodec(7).Encode(buffer);
14+
Assert.NotEqual(original, buffer);
15+
16+
new CountingXorCodec(7).Decode(buffer);
17+
Assert.Equal(original, buffer);
18+
}
19+
20+
[Fact]
21+
public void Decode_AdvancesIndependentlyFromEncode()
22+
{
23+
var codec = new CountingXorCodec(0);
24+
var encodeProbe = new byte[] { 0, 0, 0 };
25+
var decodeProbe = new byte[] { 0, 0, 0 };
26+
27+
codec.Encode(encodeProbe);
28+
codec.Decode(decodeProbe);
29+
30+
// Both directions started at position 0, so they produce the same keystream.
31+
Assert.Equal(encodeProbe, decodeProbe);
32+
}
33+
}

0 commit comments

Comments
 (0)