diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationClient.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationClient.cs
new file mode 100644
index 00000000..f5c732ad
--- /dev/null
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationClient.cs
@@ -0,0 +1,289 @@
+using System.Net.Sockets;
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip;
+
+///
+/// One EtherNet/IP TCP session: connects, registers, and sends SendRRData (unconnected) and
+/// SendUnitData (connected) requests with at most one request in flight.
+///
+///
+/// Single flight. Every exchange (write one frame, read until its reply) runs under a
+/// (1,1). The connected sequence count is advanced inside that gate, so
+/// concurrent callers can never put sequences on the wire out of order.
+/// Caller cancellation never faults the connection, and never misattributes a reply
+/// (rulings R5-1/R5-2). The caller gets at once, but the
+/// abandoned exchange keeps reading its own reply in the background while still holding the
+/// gate, bounded by the original timeout. The reply is discarded and the connection stays healthy;
+/// if the timeout elapses first the connection faults exactly as a normal timeout does. Because the
+/// next request is not written until the previous reply has been consumed, a late reply can never be
+/// taken for the next request's — which matters because ab_server zeroes the reply sender context,
+/// so context matching alone cannot tell two unconnected replies apart.
+/// Faults. A timeout, an I/O error or an encapsulation-level protocol error
+/// () closes the socket; turns false and
+/// every later call throws ().
+/// The owner reconnects with a fresh client.
+/// No half-written frames. A frame write never observes a cancellation token.
+///
+public sealed class EncapsulationClient : IAsyncDisposable
+{
+ private static readonly TimeSpan UnregisterBudget = TimeSpan.FromMilliseconds(500);
+
+ private readonly TcpClient _tcp;
+ private readonly NetworkStream _stream;
+
+ // Never disposed: an abandoned exchange may release it after DisposeAsync returns, and
+ // SemaphoreSlim.Release on a disposed instance throws. It holds no unmanaged resource unless
+ // AvailableWaitHandle is touched, which this class never does.
+ private readonly SemaphoreSlim _gate = new(1, 1);
+ private long _ctx;
+ private ushort _seq; // connected sequence count; only touched under _gate
+ private volatile bool _faulted;
+ private int _disposed;
+
+ /// The session handle the target assigned in its RegisterSession reply.
+ public uint SessionHandle { get; private set; }
+
+ /// False once the connection has faulted or been disposed.
+ public bool IsConnected => !_faulted && Volatile.Read(ref _disposed) == 0 && _tcp.Connected;
+
+ private EncapsulationClient(TcpClient tcp)
+ {
+ _tcp = tcp;
+ _stream = tcp.GetStream();
+ }
+
+ /// Open the TCP connection and register an encapsulation session.
+ /// bounds the TCP connect and the RegisterSession exchange separately.
+ public static async Task ConnectAsync(string host, int port, TimeSpan timeout, CancellationToken ct)
+ {
+ var tcp = new TcpClient { NoDelay = true };
+ try
+ {
+ using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
+ cts.CancelAfter(timeout);
+ await tcp.ConnectAsync(host, port, cts.Token).ConfigureAwait(false);
+ }
+ catch (OperationCanceledException) when (!ct.IsCancellationRequested)
+ {
+ tcp.Dispose();
+ throw new AbTransportException(AbStatus.Timeout, $"TCP connect to {host}:{port} timed out.");
+ }
+ catch (SocketException ex)
+ {
+ tcp.Dispose();
+ throw new AbTransportException(AbStatus.BadConnection, $"TCP connect to {host}:{port} failed: {ex.SocketErrorCode}", ex);
+ }
+ catch
+ {
+ tcp.Dispose();
+ throw;
+ }
+
+ var client = new EncapsulationClient(tcp);
+ try
+ {
+ var (header, _) = await client.ExchangeAsync(connected: false,
+ (ctx, _) => EncapsulationCodec.RegisterSession(ctx), timeout, ct).ConfigureAwait(false);
+ client.SessionHandle = header.SessionHandle;
+ return client;
+ }
+ catch
+ {
+ client.Abort();
+ throw;
+ }
+ }
+
+ /// Send one unconnected (UCMM) CIP request and return the CIP reply bytes.
+ public async Task> SendRRDataAsync(ReadOnlyMemory cip, TimeSpan timeout, CancellationToken ct)
+ {
+ var secs = (ushort)Math.Clamp((int)Math.Ceiling(timeout.TotalSeconds), 1, ushort.MaxValue);
+ var (_, reply) = await ExchangeAsync(connected: false,
+ (ctx, _) => EncapsulationCodec.SendRRData(SessionHandle, ctx, cip.Span, secs), timeout, ct).ConfigureAwait(false);
+ return reply;
+ }
+
+ /// Send one connected (class 3) CIP request over and return
+ /// the CIP reply bytes. The client owns the connected sequence count (review B1).
+ public async Task> SendUnitDataAsync(uint otConnectionId, ReadOnlyMemory cip,
+ TimeSpan timeout, CancellationToken ct)
+ {
+ var (_, reply) = await ExchangeAsync(connected: true,
+ (ctx, seq) => EncapsulationCodec.SendUnitData(SessionHandle, ctx, otConnectionId, seq, cip.Span), timeout, ct)
+ .ConfigureAwait(false);
+ return reply;
+ }
+
+ /// The connected sequence count after : increments, wrapping
+ /// 65535 → 1 (0 is never put on the wire).
+ internal static ushort NextSequence(ushort current) => current == ushort.MaxValue ? (ushort)1 : (ushort)(current + 1);
+
+ /// Acquire the gate, build the request (sequence assigned under the gate), then hand the
+ /// gate to , which owns its release from that point on.
+ private async Task<(EncapsulationHeader Header, ReadOnlyMemory Cip)> ExchangeAsync(
+ bool connected, Func build, TimeSpan timeout, CancellationToken ct)
+ {
+ ThrowIfUnusable();
+ await _gate.WaitAsync(ct).ConfigureAwait(false);
+
+ Task<(EncapsulationHeader, ReadOnlyMemory)> exchange;
+ try
+ {
+ ThrowIfUnusable(); // a drain we queued behind may have faulted
+ ct.ThrowIfCancellationRequested(); // nothing written yet: cancel cleanly
+ var ctx = (ulong)Interlocked.Increment(ref _ctx);
+ ushort seq = 0;
+ if (connected) seq = _seq = NextSequence(_seq);
+ var request = build(ctx, seq);
+ exchange = RunExchangeAsync(request, connected, ctx, seq, timeout);
+ }
+ catch
+ {
+ _gate.Release(); // RunExchangeAsync never started: release here
+ throw;
+ }
+
+ try
+ {
+ return await exchange.WaitAsync(ct).ConfigureAwait(false);
+ }
+ catch (OperationCanceledException) when (ct.IsCancellationRequested)
+ {
+ // Caller abandoned the wait (R5-1). The exchange keeps draining its own reply under the
+ // gate and faults on its own timeout; observe its outcome so a fault is never unobserved.
+ _ = exchange.ContinueWith(static t => _ = t.Exception, CancellationToken.None,
+ TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
+ throw;
+ }
+ }
+
+ /// Write one frame, then read frames until the matching reply arrives. Releases the gate
+ /// exactly once, in its finally. Never throws : every
+ /// failure surfaces as and faults the connection first.
+ private async Task<(EncapsulationHeader, ReadOnlyMemory)> RunExchangeAsync(
+ byte[] request, bool connected, ulong ctx, ushort seq, TimeSpan timeout)
+ {
+ var cmd = (ushort)(request[0] | (request[1] << 8));
+ try
+ {
+ using var timeoutCts = new CancellationTokenSource(timeout);
+ try
+ {
+ await _stream.WriteAsync(request, CancellationToken.None).ConfigureAwait(false);
+ while (true)
+ {
+ var frame = await ReadFrameAsync(timeoutCts.Token).ConfigureAwait(false);
+ if (TryMatch(frame, cmd, connected, ctx, seq, out var reply)) return reply;
+ }
+ }
+ catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested)
+ {
+ Fault();
+ throw new AbTransportException(AbStatus.Timeout,
+ $"No reply to command 0x{cmd:X4} within {timeout.TotalMilliseconds:F0} ms.");
+ }
+ catch (AbTransportException)
+ {
+ Fault();
+ throw;
+ }
+ catch (Exception ex) when (ex is IOException or SocketException or ObjectDisposedException
+ or EndOfStreamException or InvalidOperationException)
+ {
+ Fault();
+ throw new AbTransportException(AbStatus.BadConnection, $"Connection lost: {ex.Message}", ex);
+ }
+ }
+ finally
+ {
+ _gate.Release();
+ }
+ }
+
+ /// Does answer the request in flight? Unconnected replies match on
+ /// command and sender context, tolerating a zeroed context (ab_server, R5-2); connected replies match
+ /// on the echoed sequence count. Anything else is skipped. A matching frame is fully validated, so an
+ /// encapsulation error throws BadReply.
+ private static bool TryMatch(byte[] frame, ushort cmd, bool connected, ulong ctx, ushort seq,
+ out (EncapsulationHeader, ReadOnlyMemory) reply)
+ {
+ reply = default;
+ var h = EncapsulationCodec.ReadHeader(frame);
+ if (h.Command != cmd) return false;
+
+ if (!connected)
+ {
+ if (h.SenderContext != ctx && h.SenderContext != 0) return false;
+ var p = EncapsulationCodec.ParseReply(frame, expectConnected: false);
+ reply = (p.Header, p.Cip);
+ return true;
+ }
+
+ // A connected error reply (status != 0) carries no readable sequence; with one request in flight
+ // and abandoned exchanges drained, it can only be ours — ParseReply throws BadReply for it.
+ var c = EncapsulationCodec.ParseReply(frame, expectConnected: true);
+ if (c.Sequence != seq) return false;
+ reply = (c.Header, c.Cip);
+ return true;
+ }
+
+ private async Task ReadFrameAsync(CancellationToken ct)
+ {
+ var head = new byte[EncapsulationHeader.Size];
+ await _stream.ReadExactlyAsync(head, ct).ConfigureAwait(false);
+ var len = head[2] | (head[3] << 8);
+ var frame = new byte[EncapsulationHeader.Size + len];
+ head.CopyTo(frame, 0);
+ await _stream.ReadExactlyAsync(frame.AsMemory(EncapsulationHeader.Size), ct).ConfigureAwait(false);
+ return frame;
+ }
+
+ private void ThrowIfUnusable()
+ {
+ if (Volatile.Read(ref _disposed) != 0)
+ throw new AbTransportException(AbStatus.BadConnection, "Session is disposed.");
+ if (_faulted)
+ throw new AbTransportException(AbStatus.BadConnection, "Session is faulted; reconnect.");
+ }
+
+ private void Fault()
+ {
+ _faulted = true;
+ try { _tcp.Close(); } catch { /* already closed */ }
+ }
+
+ /// Close without UnRegisterSession (registration never completed).
+ private void Abort()
+ {
+ Interlocked.Exchange(ref _disposed, 1);
+ Fault();
+ }
+
+ /// Best-effort UnRegisterSession, then close. Waits briefly for an in-flight exchange so the
+ /// UnRegisterSession frame is never interleaved with another frame; if the gate is not free in time
+ /// the socket is closed without it.
+ public async ValueTask DisposeAsync()
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) != 0) return;
+ var acquired = await _gate.WaitAsync(UnregisterBudget).ConfigureAwait(false);
+ try
+ {
+ if (acquired && !_faulted && _tcp.Connected)
+ {
+ try
+ {
+ var bytes = EncapsulationCodec.UnRegisterSession(SessionHandle, (ulong)Interlocked.Increment(ref _ctx));
+ using var cts = new CancellationTokenSource(UnregisterBudget);
+ await _stream.WriteAsync(bytes, cts.Token).ConfigureAwait(false);
+ await _stream.FlushAsync(cts.Token).ConfigureAwait(false);
+ }
+ catch { /* best effort: the socket is closed next either way */ }
+ }
+ }
+ finally
+ {
+ Fault();
+ _tcp.Dispose();
+ if (acquired) _gate.Release();
+ }
+ }
+}
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationCodec.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationCodec.cs
new file mode 100644
index 00000000..60234a05
--- /dev/null
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationCodec.cs
@@ -0,0 +1,111 @@
+using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Wire;
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip;
+
+/// Builds and parses EtherNet/IP encapsulation frames, including the Common Packet Format
+/// (CPF) items SendRRData / SendUnitData carry.
+public static class EncapsulationCodec
+{
+ /// CPF item: Null Address (unconnected messaging).
+ public const ushort ItemNullAddress = 0x0000;
+
+ /// CPF item: Connected Address (carries the O→T connection id).
+ public const ushort ItemConnectedAddress = 0x00A1;
+
+ /// CPF item: Connected Data (sequence count + CIP).
+ public const ushort ItemConnectedData = 0x00B1;
+
+ /// CPF item: Unconnected Data (CIP).
+ public const ushort ItemUnconnectedData = 0x00B2;
+
+ /// Largest CIP payload one frame can carry: the encapsulation length field (u16) also
+ /// covers the CPF overhead, so anything near 64 KiB cannot be framed.
+ public const int MaxCipPayload = ushort.MaxValue - 64;
+
+ private static CipWriter Header(ushort cmd, uint session, ulong ctx) =>
+ new CipWriter().U16(cmd).U16(0).U32(session).U32(0).U64(ctx).U32(0);
+
+ private static byte[] Finish(CipWriter w)
+ {
+ w.PatchU16(2, (ushort)(w.Length - EncapsulationHeader.Size));
+ return w.ToArray();
+ }
+
+ /// RegisterSession: protocol version 1, option flags 0.
+ public static byte[] RegisterSession(ulong ctx) =>
+ Finish(Header(EncapCommand.RegisterSession, 0, ctx).U16(1).U16(0));
+
+ /// UnRegisterSession: header only, no data.
+ public static byte[] UnRegisterSession(uint session, ulong ctx) =>
+ Finish(Header(EncapCommand.UnRegisterSession, session, ctx));
+
+ /// SendRRData (unconnected): Null Address item + Unconnected Data item carrying .
+ public static byte[] SendRRData(uint session, ulong ctx, ReadOnlySpan cip, ushort timeoutSeconds) =>
+ Finish(Header(EncapCommand.SendRRData, session, ctx)
+ .U32(0).U16(timeoutSeconds).U16(2)
+ .U16(ItemNullAddress).U16(0)
+ .U16(ItemUnconnectedData).U16(CheckedLength(cip.Length)).Bytes(cip));
+
+ /// SendUnitData (connected): Connected Address item + Connected Data item carrying the
+ /// sequence count followed by .
+ public static byte[] SendUnitData(uint session, ulong ctx, uint connectionId, ushort sequence, ReadOnlySpan cip) =>
+ Finish(Header(EncapCommand.SendUnitData, session, ctx)
+ .U32(0).U16(0).U16(2)
+ .U16(ItemConnectedAddress).U16(4).U32(connectionId)
+ .U16(ItemConnectedData).U16((ushort)(CheckedLength(cip.Length) + 2)).U16(sequence).Bytes(cip));
+
+ private static ushort CheckedLength(int length)
+ {
+ if (length > MaxCipPayload)
+ throw new AbTransportException(AbStatus.TooLarge, $"CIP payload of {length} bytes cannot be framed (max {MaxCipPayload}).");
+ return (ushort)length;
+ }
+
+ /// Read the 24-byte header at the start of .
+ public static EncapsulationHeader ReadHeader(ReadOnlySpan b)
+ {
+ var r = new CipReader(b);
+ return new EncapsulationHeader(r.U16(), r.U16(), r.U32(), r.U32(), r.U64(), r.U32());
+ }
+
+ /// Parse a reply frame. Returns the header plus the CIP payload of the data item
+ /// (0x00B2 unconnected / 0x00B1 connected, sequence count stripped) and, for a connected reply,
+ /// the echoed sequence count. RegisterSession / UnRegisterSession replies return an empty payload.
+ /// Throws () on a non-zero
+ /// encapsulation status, a length mismatch, a truncated item list or a missing data item.
+ public static (EncapsulationHeader Header, ReadOnlyMemory Cip, ushort Sequence) ParseReply(
+ ReadOnlyMemory frame, bool expectConnected)
+ {
+ var h = ReadHeader(frame.Span);
+ if (h.Status != 0)
+ throw new AbTransportException(AbStatus.BadReply, $"Encapsulation status 0x{h.Status:X8} on command 0x{h.Command:X4}.");
+ if (h.Length != frame.Length - EncapsulationHeader.Size)
+ throw new AbTransportException(AbStatus.BadReply,
+ $"Encapsulation length mismatch: header says {h.Length}, frame carries {frame.Length - EncapsulationHeader.Size}.");
+ if (h.Command is EncapCommand.RegisterSession or EncapCommand.UnRegisterSession)
+ return (h, ReadOnlyMemory.Empty, 0);
+
+ var r = new CipReader(frame.Span[EncapsulationHeader.Size..]);
+ r.U32(); r.U16(); // interface handle, timeout
+ var items = r.U16();
+ var offset = EncapsulationHeader.Size + 8;
+ var wanted = expectConnected ? ItemConnectedData : ItemUnconnectedData;
+ for (var i = 0; i < items; i++)
+ {
+ var type = r.U16();
+ var len = r.U16();
+ offset += 4;
+ r.Bytes(len); // bounds-checks the item against the frame
+ if (type == wanted)
+ {
+ if (!expectConnected) return (h, frame.Slice(offset, len), 0);
+ if (len < 2)
+ throw new AbTransportException(AbStatus.BadReply, "Connected data item shorter than its sequence count.");
+ var seq = (ushort)(frame.Span[offset] | (frame.Span[offset + 1] << 8));
+ return (h, frame.Slice(offset + 2, len - 2), seq);
+ }
+ offset += len;
+ }
+ throw new AbTransportException(AbStatus.BadReply,
+ $"Reply to command 0x{h.Command:X4} carried no 0x{wanted:X4} data item.");
+ }
+}
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationHeader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationHeader.cs
new file mode 100644
index 00000000..f55d1619
--- /dev/null
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationHeader.cs
@@ -0,0 +1,15 @@
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip;
+
+/// EtherNet/IP encapsulation command codes (CIP Vol. 2, Table 2-3.2).
+public static class EncapCommand
+{
+ public const ushort ListIdentity = 0x0063, RegisterSession = 0x0065, UnRegisterSession = 0x0066,
+ SendRRData = 0x006F, SendUnitData = 0x0070;
+}
+
+/// The fixed 24-byte EtherNet/IP encapsulation header (all fields little-endian).
+public readonly record struct EncapsulationHeader(
+ ushort Command, ushort Length, uint SessionHandle, uint Status, ulong SenderContext, uint Options)
+{
+ public const int Size = 24;
+}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationClientTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationClientTests.cs
new file mode 100644
index 00000000..0d69e38d
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationClientTests.cs
@@ -0,0 +1,316 @@
+using System.Buffers.Binary;
+using System.Net;
+using System.Net.Sockets;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip;
+using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes;
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Eip;
+
+///
+/// against . Late-reply scenarios are
+/// driven by a TaskCompletionSource the test releases, not by sleeps, so ordering is deterministic.
+///
+public sealed class EncapsulationClientTests
+{
+ private static readonly TimeSpan Generous = TimeSpan.FromSeconds(5);
+ private static readonly TimeSpan Guard = TimeSpan.FromSeconds(10);
+ private static CancellationToken TestCt => TestContext.Current.CancellationToken;
+
+ private static Task Connect(FakeEipServer server) =>
+ EncapsulationClient.ConnectAsync("127.0.0.1", server.Port, Generous, TestCt);
+
+ private static bool IsRR(byte[] f) => FakeEipServer.Command(f) == EncapCommand.SendRRData;
+
+ [Fact]
+ public async Task Registers_and_stores_session_handle()
+ {
+ await using var server = new FakeEipServer();
+ await using var client = await Connect(server);
+ client.SessionHandle.ShouldBe(FakeEipServer.Session);
+ client.IsConnected.ShouldBeTrue();
+ FakeEipServer.Command(server.Received[0]).ShouldBe(EncapCommand.RegisterSession);
+ }
+
+ [Fact]
+ public async Task SendRRData_returns_cip_payload()
+ {
+ await using var server = new FakeEipServer();
+ byte[] cipReply = [0xCC, 0, 0, 0, 0xC4, 0, 1, 0, 0, 0];
+ server.Handler = req => FakeEipServer.RRReply(req, cipReply);
+ await using var client = await Connect(server);
+
+ var result = await client.SendRRDataAsync(new byte[] { 0x4C, 0x02 }, Generous, TestCt);
+
+ result.ToArray().ShouldBe(cipReply);
+ var sent = server.Received.Single(IsRR);
+ BinaryPrimitives.ReadUInt32LittleEndian(sent.AsSpan(4)).ShouldBe(FakeEipServer.Session);
+ FakeEipServer.RequestCip(sent).ShouldBe(new byte[] { 0x4C, 0x02 });
+ }
+
+ [Fact]
+ public async Task Works_against_a_server_that_zeroes_the_reply_sender_context()
+ {
+ // ab_server behaviour (Task 4 captures): every reply carries context 0 (R5-2, R5-3).
+ await using var server = new FakeEipServer { ZeroReplyContext = true };
+ server.Handler = req => FakeEipServer.RRReply(req, [0xCC, 0, 0, 0]);
+ await using var client = await Connect(server);
+
+ client.SessionHandle.ShouldBe(FakeEipServer.Session);
+ (await client.SendRRDataAsync(new byte[] { 0x4C }, Generous, TestCt)).ToArray().ShouldBe(new byte[] { 0xCC, 0, 0, 0 });
+ (await client.SendRRDataAsync(new byte[] { 0x4C }, Generous, TestCt)).ToArray().ShouldBe(new byte[] { 0xCC, 0, 0, 0 });
+ }
+
+ [Fact]
+ public async Task Concurrent_sends_are_serialized()
+ {
+ await using var server = new FakeEipServer();
+ server.HandlerAsync = async req =>
+ {
+ await Task.Delay(20, TestCt);
+ return FakeEipServer.RRReply(req, FakeEipServer.RequestCip(req)); // echo
+ };
+ await using var client = await Connect(server);
+
+ var calls = Enumerable.Range(1, 5)
+ .Select(i => client.SendRRDataAsync(new[] { (byte)i }, Generous, TestCt))
+ .ToArray();
+ var results = await Task.WhenAll(calls).WaitAsync(Guard, TestCt);
+
+ // The fake records frames the instant they arrive, independently of its responder, so a client
+ // that pipelined a second request before the first reply would push this to 2+.
+ server.MaxOutstanding.ShouldBe(1);
+ for (var i = 0; i < 5; i++) results[i].ToArray().ShouldBe(new[] { (byte)(i + 1) });
+ server.Received.Count(IsRR).ShouldBe(5);
+ }
+
+ [Fact]
+ public async Task Stalled_reply_times_out_and_closes_the_socket()
+ {
+ await using var server = new FakeEipServer { Stall = true };
+ await using var client = await Connect(server);
+
+ var ex = await Should.ThrowAsync(
+ client.SendRRDataAsync(new byte[] { 0x4C }, TimeSpan.FromMilliseconds(200), TestCt).WaitAsync(Guard, TestCt));
+
+ ex.Status.ShouldBe(AbStatus.Timeout);
+ client.IsConnected.ShouldBeFalse();
+ (await Should.ThrowAsync(client.SendRRDataAsync(new byte[] { 0x4C }, Generous, TestCt)))
+ .Status.ShouldBe(AbStatus.BadConnection);
+ }
+
+ [Theory]
+ [InlineData(false)]
+ [InlineData(true)] // R5-3: the late reply carries a ZEROED context, so context matching cannot reject it
+ public async Task Caller_cancellation_does_not_fault_the_connection_and_the_late_reply_is_never_misattributed(bool zeroContext)
+ {
+ await using var server = new FakeEipServer { ZeroReplyContext = zeroContext };
+ var releaseFirst = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ byte[] stale = [0xCC, 0, 0, 0, 0xAA];
+ byte[] fresh = [0xCC, 0, 0, 0, 0xBB];
+ server.ScriptAsync(
+ async req => { await releaseFirst.Task; return FakeEipServer.RRReply(req, stale); },
+ req => Task.FromResult(FakeEipServer.RRReply(req, fresh)));
+ await using var client = await Connect(server);
+
+ using var callerCts = new CancellationTokenSource();
+ var first = client.SendRRDataAsync(new byte[] { 0x01 }, Generous, callerCts.Token);
+ await server.WaitForFrameAsync(IsRR, Guard); // request 1 is on the wire
+ await callerCts.CancelAsync();
+
+ await Should.ThrowAsync(first.WaitAsync(Guard, TestCt));
+ client.IsConnected.ShouldBeTrue();
+
+ // Request 2 is issued while request 1's reply is still outstanding; only then does the server
+ // answer request 1 (late, after the cancel, before the timeout).
+ var second = client.SendRRDataAsync(new byte[] { 0x02 }, Generous, TestCt);
+ releaseFirst.SetResult();
+
+ (await second.WaitAsync(Guard, TestCt)).ToArray().ShouldBe(fresh);
+ client.IsConnected.ShouldBeTrue();
+ server.MaxOutstanding.ShouldBe(1); // request 2 waited for reply 1 (R5-1)
+ server.Received.Count(IsRR).ShouldBe(2);
+ server.HandlerErrors.ShouldBeEmpty();
+ }
+
+ [Fact]
+ public async Task Caller_cancellation_then_no_reply_faults_at_the_timeout_and_the_next_call_does_not_hang()
+ {
+ await using var server = new FakeEipServer { Stall = true };
+ await using var client = await Connect(server);
+
+ using var callerCts = new CancellationTokenSource();
+ var first = client.SendRRDataAsync(new byte[] { 0x01 }, TimeSpan.FromMilliseconds(300), callerCts.Token);
+ await server.WaitForFrameAsync(IsRR, Guard);
+ await callerCts.CancelAsync();
+ await Should.ThrowAsync(first.WaitAsync(Guard, TestCt));
+ client.IsConnected.ShouldBeTrue(); // cancel alone never faults
+
+ // Queued behind the draining exchange: must fail fast once that exchange times out, not hang.
+ var next = client.SendRRDataAsync(new byte[] { 0x02 }, Generous, TestCt);
+ var ex = await Should.ThrowAsync(next.WaitAsync(Guard, TestCt));
+
+ ex.Status.ShouldBe(AbStatus.BadConnection);
+ client.IsConnected.ShouldBeFalse();
+ server.Received.Count(IsRR).ShouldBe(1); // the queued request was never written
+ (await Should.ThrowAsync(client.SendRRDataAsync(new byte[] { 0x03 }, Generous, TestCt)
+ .WaitAsync(Guard, TestCt)))
+ .Status.ShouldBe(AbStatus.BadConnection);
+ }
+
+ [Fact]
+ public async Task An_abandoned_exchange_that_times_out_leaves_no_unobserved_task_exception()
+ {
+ var unobserved = new List();
+ EventHandler onUnobserved = (_, e) =>
+ {
+ if (e.Exception.Flatten().InnerExceptions.Any(x => x.Message.Contains("within 317 ms")))
+ lock (unobserved) unobserved.Add(e.Exception);
+ };
+ TaskScheduler.UnobservedTaskException += onUnobserved;
+ try
+ {
+ await AbandonAnExchangeThatTimesOut();
+ for (var i = 0; i < 5; i++)
+ {
+ GC.Collect();
+ GC.WaitForPendingFinalizers();
+ await Task.Delay(20, TestCt);
+ }
+ lock (unobserved) unobserved.ShouldBeEmpty();
+ }
+ finally
+ {
+ TaskScheduler.UnobservedTaskException -= onUnobserved;
+ }
+ }
+
+ // Kept out of the test body so no local in the test's own state machine roots the abandoned task.
+ [System.Runtime.CompilerServices.MethodImpl(System.Runtime.CompilerServices.MethodImplOptions.NoInlining)]
+ private static async Task AbandonAnExchangeThatTimesOut()
+ {
+ await using var server = new FakeEipServer { Stall = true };
+ await using var client = await Connect(server);
+ using (var callerCts = new CancellationTokenSource())
+ {
+ var first = client.SendRRDataAsync(new byte[] { 0x01 }, TimeSpan.FromMilliseconds(317), callerCts.Token);
+ await server.WaitForFrameAsync(IsRR, Guard);
+ await callerCts.CancelAsync();
+ await Should.ThrowAsync(first.WaitAsync(Guard, TestCt));
+ }
+ var deadline = DateTime.UtcNow + Guard;
+ while (client.IsConnected && DateTime.UtcNow < deadline) await Task.Delay(10, TestCt); // drain times out
+ client.IsConnected.ShouldBeFalse();
+ }
+
+ [Fact]
+ public async Task Connected_sequence_is_assigned_inside_the_gate()
+ {
+ await using var server = new FakeEipServer();
+ server.Handler = req => FakeEipServer.UnitReply(req, FakeEipServer.RequestCip(req)); // echo seq + cip
+ await using var client = await Connect(server);
+
+ var calls = Enumerable.Range(1, 20)
+ .Select(i => client.SendUnitDataAsync(0xAABBCCDD, new[] { (byte)i }, Generous, TestCt))
+ .ToArray();
+ var results = await Task.WhenAll(calls).WaitAsync(Guard, TestCt);
+
+ var units = server.Received.Where(f => FakeEipServer.Command(f) == EncapCommand.SendUnitData).ToList();
+ var sequences = units.Select(FakeEipServer.RequestSequence).ToList();
+ sequences.ShouldBe(Enumerable.Range(1, 20).Select(i => (ushort)i).ToList()); // strictly increasing, no dupes
+ units.Select(f => BitConverter.ToUInt32(f, 36)).ShouldAllBe(id => id == 0xAABBCCDD); // A1 item connection id
+ for (var i = 0; i < 20; i++) results[i].ToArray().ShouldBe(new[] { (byte)(i + 1) }); // each got its own reply
+ server.MaxOutstanding.ShouldBe(1);
+ }
+
+ [Theory]
+ [InlineData(0, 1)]
+ [InlineData(1, 2)]
+ [InlineData(65534, 65535)]
+ [InlineData(65535, 1)] // wraps past 0, which is never put on the wire
+ public void Connected_sequence_wraps_65535_to_1(int current, int next) =>
+ EncapsulationClient.NextSequence((ushort)current).ShouldBe((ushort)next);
+
+ [Fact]
+ public async Task Reply_with_foreign_context_is_skipped()
+ {
+ await using var server = new FakeEipServer();
+ byte[] good = [0xCC, 0, 0, 0, 0x01];
+ server.Handler = req =>
+ [.. FakeEipServer.RRReplyWithContext(req, 0xDEADBEEF, [0xCC, 0, 0, 0, 0xEE]), .. FakeEipServer.RRReply(req, good)];
+ await using var client = await Connect(server);
+
+ (await client.SendRRDataAsync(new byte[] { 0x4C }, Generous, TestCt)).ToArray().ShouldBe(good);
+ client.IsConnected.ShouldBeTrue();
+ }
+
+ [Fact]
+ public async Task Connected_reply_with_foreign_sequence_is_skipped()
+ {
+ await using var server = new FakeEipServer();
+ byte[] good = [0xCC, 0, 0, 0, 0x01];
+ server.Handler = req =>
+ [.. FakeEipServer.UnitReplyWithSequence(req, 999, [0xCC, 0, 0, 0, 0xEE]), .. FakeEipServer.UnitReply(req, good)];
+ await using var client = await Connect(server);
+
+ (await client.SendUnitDataAsync(1, new byte[] { 0x4C }, Generous, TestCt)).ToArray().ShouldBe(good);
+ client.IsConnected.ShouldBeTrue();
+ }
+
+ [Fact]
+ public async Task Dropped_connection_is_BadConnection_and_faults()
+ {
+ await using var server = new FakeEipServer { DropOnNextRequest = true };
+ await using var client = await Connect(server);
+
+ (await Should.ThrowAsync(
+ client.SendRRDataAsync(new byte[] { 0x4C }, Generous, TestCt).WaitAsync(Guard, TestCt)))
+ .Status.ShouldBe(AbStatus.BadConnection);
+ client.IsConnected.ShouldBeFalse();
+ }
+
+ [Fact]
+ public async Task Nonzero_encapsulation_status_in_the_reply_is_BadReply_and_faults()
+ {
+ await using var server = new FakeEipServer();
+ server.Handler = req =>
+ {
+ var r = FakeEipServer.RRReply(req, [0xCC, 0, 0, 0]);
+ r[8] = 0x64; // status 0x64: invalid session handle
+ return r;
+ };
+ await using var client = await Connect(server);
+
+ (await Should.ThrowAsync(client.SendRRDataAsync(new byte[] { 0x4C }, Generous, TestCt)))
+ .Status.ShouldBe(AbStatus.BadReply);
+ client.IsConnected.ShouldBeFalse();
+ }
+
+ [Fact]
+ public async Task Dispose_sends_UnRegisterSession()
+ {
+ await using var server = new FakeEipServer();
+ var client = await Connect(server);
+ await client.DisposeAsync();
+
+ var frame = await server.WaitForFrameAsync(f => FakeEipServer.Command(f) == EncapCommand.UnRegisterSession, Guard);
+ BinaryPrimitives.ReadUInt32LittleEndian(frame.AsSpan(4)).ShouldBe(FakeEipServer.Session);
+ client.IsConnected.ShouldBeFalse();
+ (await Should.ThrowAsync(client.SendRRDataAsync(new byte[] { 0x4C }, Generous, TestCt)))
+ .Status.ShouldBe(AbStatus.BadConnection);
+ await client.DisposeAsync(); // idempotent
+ }
+
+ [Fact]
+ public async Task Connect_to_a_closed_port_is_BadConnection()
+ {
+ var probe = new TcpListener(IPAddress.Loopback, 0);
+ probe.Start();
+ var port = ((IPEndPoint)probe.LocalEndpoint).Port;
+ probe.Stop();
+
+ (await Should.ThrowAsync(
+ EncapsulationClient.ConnectAsync("127.0.0.1", port, Generous, TestCt)))
+ .Status.ShouldBe(AbStatus.BadConnection);
+ }
+}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationCodecTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationCodecTests.cs
new file mode 100644
index 00000000..a9c3888b
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationCodecTests.cs
@@ -0,0 +1,151 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip;
+using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Captures;
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Eip;
+
+public sealed class EncapsulationCodecTests
+{
+ private const ulong Ctx = 0x0807060504030201;
+
+ [Fact]
+ public void RegisterSession_frame_is_byte_exact() =>
+ Convert.ToHexString(EncapsulationCodec.RegisterSession(Ctx)).ShouldBe(
+ "6500" + "0400" + "00000000" + "00000000" + "0102030405060708" + "00000000" + "0100" + "0000");
+
+ [Fact]
+ public void UnRegisterSession_frame_is_header_only() =>
+ Convert.ToHexString(EncapsulationCodec.UnRegisterSession(0x11223344, Ctx)).ShouldBe(
+ "6600" + "0000" + "44332211" + "00000000" + "0102030405060708" + "00000000");
+
+ [Fact]
+ public void SendRRData_wraps_cip_in_null_address_and_unconnected_data_items()
+ {
+ var frame = EncapsulationCodec.SendRRData(0x11223344, Ctx, new byte[] { 0x4C, 0x02 }, timeoutSeconds: 10);
+ Convert.ToHexString(frame).ShouldBe(
+ "6F00" + "1200" + "44332211" + "00000000" + "0102030405060708" + "00000000" +
+ "00000000" + "0A00" + "0200" + "00000000" + "B2000200" + "4C02");
+ }
+
+ [Fact]
+ public void SendUnitData_wraps_cip_with_connection_id_and_sequence()
+ {
+ var frame = EncapsulationCodec.SendUnitData(0x11223344, Ctx, connectionId: 0xAABBCCDD, sequence: 7, new byte[] { 0x4C, 0x02 });
+ Convert.ToHexString(frame).ShouldBe(
+ "7000" + "1800" + "44332211" + "00000000" + "0102030405060708" + "00000000" +
+ "00000000" + "0000" + "0200" + "A1000400" + "DDCCBBAA" + "B1000400" + "0700" + "4C02");
+ }
+
+ [Fact]
+ public void Oversized_cip_payload_is_refused_not_truncated() =>
+ Should.Throw(() =>
+ EncapsulationCodec.SendRRData(1, Ctx, new byte[EncapsulationCodec.MaxCipPayload + 1], 1))
+ .Status.ShouldBe(AbStatus.TooLarge);
+
+ [Fact]
+ public void RegisterSession_matches_libplctag_capture_except_context()
+ {
+ var captured = CaptureFile.Load("logix-register-session.hex").First(f => f.FromClient).Bytes;
+ var ours = EncapsulationCodec.RegisterSession(Ctx);
+ ours.Length.ShouldBe(captured.Length);
+ ours.AsSpan(0, 12).ToArray().ShouldBe(captured.AsSpan(0, 12).ToArray()); // cmd,len,session,status
+ ours.AsSpan(20).ToArray().ShouldBe(captured.AsSpan(20).ToArray()); // options + data
+ }
+
+ [Fact]
+ public void SendRRData_matches_libplctag_forward_open_capture()
+ {
+ // libplctag put a 1-second timeout in the SendRRData timeout field; the session handle and
+ // context are copied from the capture so the whole frame must match byte for byte.
+ var captured = CaptureFile.Load("logix-forward-open.hex").First(f => f.FromClient).Bytes;
+ var h = EncapsulationCodec.ReadHeader(captured);
+ var ours = EncapsulationCodec.SendRRData(h.SessionHandle, h.SenderContext,
+ CaptureFile.CipPayload(new CapturedFrame(true, captured)), timeoutSeconds: 1);
+ Convert.ToHexString(ours).ShouldBe(Convert.ToHexString(captured));
+ }
+
+ [Fact]
+ public void SendUnitData_matches_libplctag_read_capture_except_timeout_field()
+ {
+ // libplctag writes 1 into the SendUnitData interface-timeout field; the CIP spec says the field
+ // is unused for SendUnitData and the brief pins ours to 0 — so bytes 28-29 are excluded and
+ // asserted separately. Everything else (connection id, sequence, CIP) must match exactly.
+ var captured = CaptureFile.Load("logix-read-dint.hex").First(f => f.FromClient && f.Bytes[0] == 0x70).Bytes;
+ var h = EncapsulationCodec.ReadHeader(captured);
+ var parsed = EncapsulationCodec.ParseReply(captured, expectConnected: true);
+ var connId = BitConverter.ToUInt32(captured, 24 + 8 + 4); // header, iface+timeout+count, A1 type+len
+ var ours = EncapsulationCodec.SendUnitData(h.SessionHandle, h.SenderContext, connId, parsed.Sequence, parsed.Cip.Span);
+
+ ours.Length.ShouldBe(captured.Length);
+ ours.AsSpan(0, 28).ToArray().ShouldBe(captured.AsSpan(0, 28).ToArray());
+ ours.AsSpan(30).ToArray().ShouldBe(captured.AsSpan(30).ToArray());
+ ours.AsSpan(28, 2).ToArray().ShouldBe(new byte[] { 0, 0 });
+ connId.ShouldBe(0x2CDF633Fu);
+ parsed.Sequence.ShouldBe((ushort)1);
+ }
+
+ [Fact]
+ public void ParseReply_of_captured_connected_read_strips_the_sequence_count()
+ {
+ var reply = CaptureFile.Load("logix-read-dint.hex").First(f => !f.FromClient && f.Bytes[0] == 0x70).Bytes;
+ var (h, cip, seq) = EncapsulationCodec.ParseReply(reply, expectConnected: true);
+ h.Command.ShouldBe(EncapCommand.SendUnitData);
+ h.SessionHandle.ShouldBe(0x1A08A5F3u);
+ seq.ShouldBe((ushort)1);
+ Convert.ToHexString(cip.ToArray()).ShouldBe("D2000000C40040E20100"); // Read Tag reply, DINT 123456
+ }
+
+ [Fact]
+ public void ParseReply_of_captured_unconnected_forward_open_returns_the_cip_reply()
+ {
+ var reply = CaptureFile.Load("logix-forward-open.hex").First(f => !f.FromClient).Bytes;
+ var (h, cip, seq) = EncapsulationCodec.ParseReply(reply, expectConnected: false);
+ h.Command.ShouldBe(EncapCommand.SendRRData);
+ seq.ShouldBe((ushort)0);
+ cip.Length.ShouldBe(0x1E);
+ cip.Span[0].ShouldBe((byte)0xDB); // Large Forward Open reply
+ }
+
+ [Fact]
+ public void ParseReply_of_register_session_returns_header_and_empty_payload()
+ {
+ var reply = CaptureFile.Load("logix-register-session.hex").First(f => !f.FromClient).Bytes;
+ var (h, cip, _) = EncapsulationCodec.ParseReply(reply, expectConnected: false);
+ h.SessionHandle.ShouldBe(0x1A08A5F3u);
+ cip.IsEmpty.ShouldBeTrue();
+ }
+
+ [Fact]
+ public void Nonzero_encapsulation_status_is_BadReply()
+ {
+ var reply = Convert.FromHexString("6F00" + "0000" + "44332211" + "01000000" + "0102030405060708" + "00000000");
+ Should.Throw(() => EncapsulationCodec.ParseReply(reply, expectConnected: false))
+ .Status.ShouldBe(AbStatus.BadReply);
+ }
+
+ [Fact]
+ public void Length_mismatch_is_BadReply()
+ {
+ var frame = EncapsulationCodec.SendRRData(1, Ctx, new byte[] { 0xCC, 0 }, 0);
+ Should.Throw(() => EncapsulationCodec.ParseReply(frame.AsMemory(0, frame.Length - 1), false))
+ .Status.ShouldBe(AbStatus.BadReply);
+ }
+
+ [Fact]
+ public void Missing_expected_data_item_is_BadReply()
+ {
+ // An unconnected (0xB2) reply parsed as connected has no 0xB1 item.
+ var frame = EncapsulationCodec.SendRRData(1, Ctx, new byte[] { 0xCC, 0 }, 0);
+ Should.Throw(() => EncapsulationCodec.ParseReply(frame, expectConnected: true))
+ .Status.ShouldBe(AbStatus.BadReply);
+ }
+
+ [Fact]
+ public void Item_length_running_past_the_frame_is_BadReply()
+ {
+ var frame = EncapsulationCodec.SendRRData(1, Ctx, new byte[] { 0xCC, 0 }, 0);
+ frame[24 + 8 + 4 + 2] = 0x40; // B2 item length 2 -> 64
+ Should.Throw(() => EncapsulationCodec.ParseReply(frame, expectConnected: false))
+ .Status.ShouldBe(AbStatus.BadReply);
+ }
+}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeEipServer.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeEipServer.cs
new file mode 100644
index 00000000..79bf33bf
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeEipServer.cs
@@ -0,0 +1,251 @@
+using System.Buffers.Binary;
+using System.Collections.Concurrent;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading.Channels;
+using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip;
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes;
+
+///
+/// Scriptable loopback EtherNet/IP endpoint. RegisterSession is answered automatically (session
+/// ) and UnRegisterSession is recorded but never answered. Every other frame is
+/// answered by the next queued step, else by ; a
+/// non-null result is written back verbatim (it may hold several concatenated frames).
+///
+///
+/// Each connection runs a reader loop and a responder loop, so frames are recorded the moment they
+/// arrive — even while a reply is still pending. therefore really does
+/// see a client that pipelines a second request before the first reply.
+///
+internal sealed class FakeEipServer : IAsyncDisposable
+{
+ public const uint Session = 0x11223344;
+
+ private readonly TcpListener _listener = new(IPAddress.Loopback, 0);
+ private readonly CancellationTokenSource _cts = new();
+ private readonly ConcurrentQueue>> _script = new();
+ private readonly List _received = [];
+ private readonly List<(Predicate Match, TaskCompletionSource Tcs)> _waiters = [];
+ private readonly ConcurrentBag _clients = [];
+ private readonly ConcurrentBag _handlerErrors = [];
+ private int _connections;
+ private int _outstanding;
+ private int _maxOutstanding;
+
+ public FakeEipServer()
+ {
+ _listener.Start();
+ _ = AcceptLoop();
+ }
+
+ public int Port => ((IPEndPoint)_listener.LocalEndpoint).Port;
+
+ /// Every frame received, in arrival order (snapshot).
+ public IReadOnlyList Received { get { lock (_received) return [.. _received]; } }
+
+ public int Connections => Volatile.Read(ref _connections);
+
+ /// Largest number of non-session requests ever received but not yet answered.
+ public int MaxOutstanding => Volatile.Read(ref _maxOutstanding);
+
+ /// Exceptions thrown by handlers / script steps (a test assertion inside a handler lands here).
+ public IReadOnlyCollection HandlerErrors => _handlerErrors;
+
+ /// Default responder, used once the script is exhausted.
+ public Func> HandlerAsync { get; set; } = _ => Task.FromResult(null);
+
+ /// Synchronous convenience setter for .
+ public Func Handler { set => HandlerAsync = req => Task.FromResult(value(req)); }
+
+ /// Never answer non-session requests.
+ public bool Stall { get; set; }
+
+ /// Zero the sender context (header bytes 12-19) of every reply frame, as ab_server does.
+ public bool ZeroReplyContext { get; set; }
+
+ /// Close the connection instead of answering the next non-session request.
+ public bool DropOnNextRequest { get; set; }
+
+ /// Queue one-shot responders, consumed one per non-session request.
+ public void Script(params Func[] steps)
+ {
+ foreach (var s in steps) _script.Enqueue(req => Task.FromResult(s(req)));
+ }
+
+ /// Queue one-shot asynchronous responders (e.g. one that waits for the test to release it).
+ public void ScriptAsync(params Func>[] steps)
+ {
+ foreach (var s in steps) _script.Enqueue(s);
+ }
+
+ /// Completes with the first received frame (past or future) matching .
+ public Task WaitForFrameAsync(Predicate match, TimeSpan timeout)
+ {
+ var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ lock (_received)
+ {
+ var hit = _received.FirstOrDefault(f => match(f));
+ if (hit is not null) return Task.FromResult(hit);
+ _waiters.Add((match, tcs));
+ }
+ return tcs.Task.WaitAsync(timeout);
+ }
+
+ /// Completes once frames with encapsulation command have arrived.
+ public async Task WaitForCommandCountAsync(ushort command, int count, TimeSpan timeout)
+ {
+ var seen = 0;
+ await WaitForFrameAsync(f => Command(f) == command && ++seen >= count, timeout);
+ }
+
+ public static ushort Command(byte[] frame) => BinaryPrimitives.ReadUInt16LittleEndian(frame);
+
+ private async Task AcceptLoop()
+ {
+ while (!_cts.IsCancellationRequested)
+ {
+ TcpClient c;
+ try { c = await _listener.AcceptTcpClientAsync(_cts.Token); }
+ catch { return; }
+ Interlocked.Increment(ref _connections);
+ _clients.Add(c);
+ _ = Serve(c);
+ }
+ }
+
+ private async Task Serve(TcpClient c)
+ {
+ using var _ = c;
+ var s = c.GetStream();
+ var inbox = Channel.CreateUnbounded();
+ var responder = Respond(c, s, inbox.Reader);
+ var head = new byte[EncapsulationHeader.Size];
+ while (!_cts.IsCancellationRequested)
+ {
+ if (!await ReadExact(s, head)) break;
+ var len = head[2] | (head[3] << 8);
+ var frame = new byte[EncapsulationHeader.Size + len];
+ head.CopyTo(frame, 0);
+ if (!await ReadExact(s, frame.AsMemory(EncapsulationHeader.Size))) break;
+ if (Command(frame) is not (EncapCommand.RegisterSession or EncapCommand.UnRegisterSession))
+ {
+ var now = Interlocked.Increment(ref _outstanding);
+ int max;
+ while (now > (max = Volatile.Read(ref _maxOutstanding)) &&
+ Interlocked.CompareExchange(ref _maxOutstanding, now, max) != max) { }
+ }
+ Record(frame);
+ inbox.Writer.TryWrite(frame);
+ }
+ inbox.Writer.TryComplete();
+ try { await responder; } catch { /* connection gone */ }
+ }
+
+ private async Task Respond(TcpClient c, NetworkStream s, ChannelReader inbox)
+ {
+ await foreach (var frame in inbox.ReadAllAsync(_cts.Token))
+ {
+ byte[]? reply;
+ switch (Command(frame))
+ {
+ case EncapCommand.RegisterSession:
+ reply = (byte[])frame.Clone();
+ BinaryPrimitives.WriteUInt32LittleEndian(reply.AsSpan(4), Session);
+ break;
+ case EncapCommand.UnRegisterSession:
+ reply = null;
+ break;
+ default:
+ if (DropOnNextRequest)
+ {
+ DropOnNextRequest = false;
+ c.Close();
+ return;
+ }
+ if (Stall) continue; // stays outstanding forever
+ var step = _script.TryDequeue(out var scripted) ? scripted : HandlerAsync;
+ try { reply = await step(frame); }
+ catch (Exception ex) { _handlerErrors.Add(ex); reply = null; }
+ Interlocked.Decrement(ref _outstanding); // before the write: the client cannot send
+ break; // its next request until it has this reply
+ }
+ if (reply is null) continue;
+ if (ZeroReplyContext) ZeroContexts(reply);
+ await s.WriteAsync(reply, _cts.Token);
+ }
+ }
+
+ private void Record(byte[] frame)
+ {
+ List> fire = [];
+ lock (_received)
+ {
+ _received.Add(frame);
+ for (var i = _waiters.Count - 1; i >= 0; i--)
+ {
+ if (!_waiters[i].Match(frame)) continue;
+ fire.Add(_waiters[i].Tcs);
+ _waiters.RemoveAt(i);
+ }
+ }
+ foreach (var t in fire) t.TrySetResult(frame);
+ }
+
+ private static void ZeroContexts(byte[] buffer)
+ {
+ for (var off = 0; off + EncapsulationHeader.Size <= buffer.Length;
+ off += EncapsulationHeader.Size + BinaryPrimitives.ReadUInt16LittleEndian(buffer.AsSpan(off + 2)))
+ buffer.AsSpan(off + 12, 8).Clear();
+ }
+
+ private async Task ReadExact(NetworkStream s, Memory buf)
+ {
+ var got = 0;
+ while (got < buf.Length)
+ {
+ int n;
+ try { n = await s.ReadAsync(buf[got..], _cts.Token); }
+ catch { return false; }
+ if (n == 0) return false;
+ got += n;
+ }
+ return true;
+ }
+
+ // ---- reply builders -------------------------------------------------------------------------
+
+ /// The CIP payload of a SendRRData / SendUnitData request.
+ public static byte[] RequestCip(byte[] request) =>
+ EncapsulationCodec.ParseReply(request, expectConnected: Command(request) == EncapCommand.SendUnitData).Cip.ToArray();
+
+ /// The connected sequence count of a SendUnitData request.
+ public static ushort RequestSequence(byte[] request) =>
+ EncapsulationCodec.ParseReply(request, expectConnected: true).Sequence;
+
+ /// A SendRRData reply echoing the request's session + context, carrying .
+ public static byte[] RRReply(byte[] request, byte[] cipReply) =>
+ RRReplyWithContext(request, BinaryPrimitives.ReadUInt64LittleEndian(request.AsSpan(12)), cipReply);
+
+ /// A SendRRData reply with an explicit sender context.
+ public static byte[] RRReplyWithContext(byte[] request, ulong ctx, byte[] cipReply) =>
+ EncapsulationCodec.SendRRData(BinaryPrimitives.ReadUInt32LittleEndian(request.AsSpan(4)), ctx, cipReply, 0);
+
+ /// A SendUnitData reply echoing the request's sequence count, carrying .
+ public static byte[] UnitReply(byte[] request, byte[] cipReply) =>
+ UnitReplyWithSequence(request, RequestSequence(request), cipReply);
+
+ /// A SendUnitData reply with an explicit sequence count (T→O connection id 0x5A5A5A5A).
+ public static byte[] UnitReplyWithSequence(byte[] request, ushort sequence, byte[] cipReply) =>
+ EncapsulationCodec.SendUnitData(BinaryPrimitives.ReadUInt32LittleEndian(request.AsSpan(4)),
+ BinaryPrimitives.ReadUInt64LittleEndian(request.AsSpan(12)), 0x5A5A5A5A, sequence, cipReply);
+
+ public async ValueTask DisposeAsync()
+ {
+ await _cts.CancelAsync();
+ _listener.Stop();
+ foreach (var c in _clients)
+ {
+ try { c.Dispose(); } catch { /* already gone */ }
+ }
+ }
+}