dohertj2/lmxopcua
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

FOCAS Tier-C PR C — IPC path end-to-end: Proxy IpcFocasClient + Host FwlibFrameHandler + IFocasBackend abstraction. Third of 5 PRs for #220. Ships the wire path from IFocasClient calls in the .NET 10 driver, over a named-pipe (or in-memory stream) to the .NET 4.8 Host's FwlibFrameHandler, dispatched to an IFocasBackend. Keeps the existing IFocasClient DI seam intact so existing unit tests are unaffected (172/172 still pass). Proxy side adds Ipc/FocasIpcClient (owns one pipe stream + call gate so concurrent callers don't interleave frames, supports both real NamedPipeClientStream and arbitrary Stream for in-memory test loopback) and Ipc/IpcFocasClient (implements IFocasClient by forwarding every call as an IPC frame — Connect sends OpenSessionRequest and caches the SessionId; Read sends ReadRequest and decodes the typed value via FocasDataTypeCode; Write sends WriteRequest for non-bit data or PmcBitWriteRequest when it's a PMC bit so the RMW critical section stays on the Host; Probe sends ProbeRequest; Dispose best-effort sends CloseSessionRequest); plus FocasIpcException surfacing Host-side ErrorResponse frames as typed exceptions. Host side adds Backend/IFocasBackend (the Host's view of one FOCAS session — Open/Close/Read/Write/PmcBitWrite/Probe) with two implementations: FakeFocasBackend (in-memory, per-address value store, honors bit-write RMW semantics against the containing byte — used by tests and as an OTOPCUA_FOCAS_BACKEND=fake operational stub) and UnconfiguredFocasBackend (structured failure pointing at docs/v2/focas-deployment.md — the safe default when OTOPCUA_FOCAS_BACKEND is unset or hardware isn't configured). Ipc/FwlibFrameHandler replaces StubFrameHandler: deserializes each request DTO, delegates to the IFocasBackend, re-serializes into the matching response kind. Catches backend exceptions and surfaces them as ErrorResponse{backend-exception} rather than tearing down the pipe. Program.cs now picks the backend from OTOPCUA_FOCAS_BACKEND env var (fake/unconfigured/fwlib32; fwlib32 still maps to Unconfigured because the real Fwlib32 P/Invoke integration is a hardware-dependent follow-up — #220 captures it). Tests: 7 new IPC round-trip tests on the Proxy side (IpcFocasClient vs. an IpcLoopback fake server: connect happy path, connect rejection, read decode, write round-trip, PMC bit write routes to first-class RMW frame, probe, ErrorResponse surfaces as typed exception) + 6 new Host-side tests on FwlibFrameHandler (OpenSession allocates id, read-without-session fails, full open/write/read round-trip preserves value, PmcBitWrite sets the specified bit, Probe reports healthy with open session, UnconfiguredBackend returns pointed-at-docs error with ErrorCode=NoFwlibBackend). Existing 165 FOCAS unit tests + 24 Shared tests + 3 Host handshake tests all unchanged. Total post-PR: 172+24+9 = 205 FOCAS-family tests green. What's NOT in this PR: the actual Fwlib32.dll P/Invoke integration inside the Host (FwlibHostedBackend) lands as a hardware-dependent follow-up since no CNC is available for validation; supervisor + respawn + crash-loop breaker comes in PR D; MMF + NSSM install scripts in PR E.

Browse Source 
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
2026-04-20 14:10:52 -04:00
parent 3609a5c676
commit 3892555631
11 changed files with 1163 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

200
tests/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Host.Tests/FwlibFrameHandlerTests.cs Normal file
View File

@@ -0,0 +1,200 @@
using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using MessagePack;
using Serilog;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Host.Backend;
using ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Host.Ipc;
using ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Shared;
using ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Shared.Contracts;
namespace ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Host.Tests
{
/// <summary>
/// Validates that <see cref="FwlibFrameHandler"/> correctly dispatches each
/// <see cref="FocasMessageKind"/> to the corresponding <see cref="IFocasBackend"/>
/// method and serializes the response into the expected response kind. Uses
/// <see cref="FakeFocasBackend"/> so no hardware is needed.
/// </summary>
[Trait("Category", "Unit")]
public sealed class FwlibFrameHandlerTests
{
private static async Task RoundTripAsync<TReq, TResp>(
IFrameHandler handler, FocasMessageKind reqKind, TReq req, FocasMessageKind expectedRespKind,
Action<TResp> assertResponse)
{
using var buffer = new MemoryStream();
using var writer = new FrameWriter(buffer, leaveOpen: true);
await handler.HandleAsync(reqKind, MessagePackSerializer.Serialize(req), writer, CancellationToken.None);
buffer.Position = 0;
using var reader = new FrameReader(buffer, leaveOpen: true);
var frame = await reader.ReadFrameAsync(CancellationToken.None);
frame.HasValue.ShouldBeTrue();
frame!.Value.Kind.ShouldBe(expectedRespKind);
assertResponse(MessagePackSerializer.Deserialize<TResp>(frame.Value.Body));
}
private static FwlibFrameHandler BuildHandler() =>
new(new FakeFocasBackend(), new LoggerConfiguration().CreateLogger());
[Fact]
public async Task OpenSession_returns_a_new_session_id()
{
long sessionId = 0;
await RoundTripAsync<OpenSessionRequest, OpenSessionResponse>(
BuildHandler(),
FocasMessageKind.OpenSessionRequest,
new OpenSessionRequest { HostAddress = "h:8193" },
FocasMessageKind.OpenSessionResponse,
resp => { resp.Success.ShouldBeTrue(); resp.SessionId.ShouldBeGreaterThan(0L); sessionId = resp.SessionId; });
sessionId.ShouldBeGreaterThan(0L);
}
[Fact]
public async Task Read_without_open_session_returns_internal_error()
{
await RoundTripAsync<ReadRequest, ReadResponse>(
BuildHandler(),
FocasMessageKind.ReadRequest,
new ReadRequest
{
SessionId = 999,
Address = new FocasAddressDto { Kind = 0, PmcLetter = "R", Number = 100 },
DataType = FocasDataTypeCode.Int32,
},
FocasMessageKind.ReadResponse,
resp => { resp.Success.ShouldBeFalse(); resp.Error.ShouldContain("session-not-open"); });
}
[Fact]
public async Task Full_open_write_read_round_trip_preserves_value()
{
var handler = BuildHandler();
// Open.
using var buffer = new MemoryStream();
using var writer = new FrameWriter(buffer, leaveOpen: true);
await handler.HandleAsync(FocasMessageKind.OpenSessionRequest,
MessagePackSerializer.Serialize(new OpenSessionRequest { HostAddress = "h:8193" }), writer, CancellationToken.None);
buffer.Position = 0;
using var reader = new FrameReader(buffer, leaveOpen: true);
var openFrame = await reader.ReadFrameAsync(CancellationToken.None);
var openResp = MessagePackSerializer.Deserialize<OpenSessionResponse>(openFrame!.Value.Body);
var sessionId = openResp.SessionId;
// Write 42 at MACRO:500 as Int32.
buffer.Position = 0;
buffer.SetLength(0);
await handler.HandleAsync(FocasMessageKind.WriteRequest,
MessagePackSerializer.Serialize(new WriteRequest
{
SessionId = sessionId,
Address = new FocasAddressDto { Kind = 2, Number = 500 },
DataType = FocasDataTypeCode.Int32,
ValueTypeCode = FocasDataTypeCode.Int32,
ValueBytes = MessagePackSerializer.Serialize((int)42),
}), writer, CancellationToken.None);
// Read back.
buffer.Position = 0;
buffer.SetLength(0);
await handler.HandleAsync(FocasMessageKind.ReadRequest,
MessagePackSerializer.Serialize(new ReadRequest
{
SessionId = sessionId,
Address = new FocasAddressDto { Kind = 2, Number = 500 },
DataType = FocasDataTypeCode.Int32,
}), writer, CancellationToken.None);
buffer.Position = 0;
var readFrame = await reader.ReadFrameAsync(CancellationToken.None);
readFrame.HasValue.ShouldBeTrue();
readFrame!.Value.Kind.ShouldBe(FocasMessageKind.ReadResponse);
// With buffer reuse there may be multiple queued frames; we want the last one.
var lastResp = MessagePackSerializer.Deserialize<ReadResponse>(readFrame.Value.Body);
// If the Write frame is first, drain it.
if (lastResp.ValueBytes is null)
{
var next = await reader.ReadFrameAsync(CancellationToken.None);
lastResp = MessagePackSerializer.Deserialize<ReadResponse>(next!.Value.Body);
}
lastResp.Success.ShouldBeTrue();
MessagePackSerializer.Deserialize<int>(lastResp.ValueBytes!).ShouldBe(42);
}
[Fact]
public async Task PmcBitWrite_sets_specified_bit()
{
var handler = BuildHandler();
using var buffer = new MemoryStream();
using var writer = new FrameWriter(buffer, leaveOpen: true);
await handler.HandleAsync(FocasMessageKind.OpenSessionRequest,
MessagePackSerializer.Serialize(new OpenSessionRequest { HostAddress = "h:8193" }), writer, CancellationToken.None);
buffer.Position = 0;
using var reader = new FrameReader(buffer, leaveOpen: true);
var openFrame = await reader.ReadFrameAsync(CancellationToken.None);
var sessionId = MessagePackSerializer.Deserialize<OpenSessionResponse>(openFrame!.Value.Body).SessionId;
buffer.Position = 0; buffer.SetLength(0);
await handler.HandleAsync(FocasMessageKind.PmcBitWriteRequest,
MessagePackSerializer.Serialize(new PmcBitWriteRequest
{
SessionId = sessionId,
Address = new FocasAddressDto { Kind = 0, PmcLetter = "R", Number = 100 },
BitIndex = 3,
Value = true,
}), writer, CancellationToken.None);
buffer.Position = 0;
var resp = MessagePackSerializer.Deserialize<PmcBitWriteResponse>(
(await reader.ReadFrameAsync(CancellationToken.None))!.Value.Body);
resp.Success.ShouldBeTrue();
resp.StatusCode.ShouldBe(0u);
}
[Fact]
public async Task Probe_reports_healthy_when_session_open()
{
var handler = BuildHandler();
using var buffer = new MemoryStream();
using var writer = new FrameWriter(buffer, leaveOpen: true);
await handler.HandleAsync(FocasMessageKind.OpenSessionRequest,
MessagePackSerializer.Serialize(new OpenSessionRequest { HostAddress = "h:8193" }), writer, CancellationToken.None);
buffer.Position = 0;
using var reader = new FrameReader(buffer, leaveOpen: true);
var sessionId = MessagePackSerializer.Deserialize<OpenSessionResponse>(
(await reader.ReadFrameAsync(CancellationToken.None))!.Value.Body).SessionId;
buffer.Position = 0; buffer.SetLength(0);
await handler.HandleAsync(FocasMessageKind.ProbeRequest,
MessagePackSerializer.Serialize(new ProbeRequest { SessionId = sessionId }), writer, CancellationToken.None);
buffer.Position = 0;
var resp = MessagePackSerializer.Deserialize<ProbeResponse>(
(await reader.ReadFrameAsync(CancellationToken.None))!.Value.Body);
resp.Healthy.ShouldBeTrue();
}
[Fact]
public async Task Unconfigured_backend_returns_pointed_error_message()
{
var handler = new FwlibFrameHandler(new UnconfiguredFocasBackend(), new LoggerConfiguration().CreateLogger());
await RoundTripAsync<OpenSessionRequest, OpenSessionResponse>(
handler,
FocasMessageKind.OpenSessionRequest,
new OpenSessionRequest { HostAddress = "h:8193" },
FocasMessageKind.OpenSessionResponse,
resp =>
{
resp.Success.ShouldBeFalse();
resp.Error.ShouldContain("Fwlib32");
resp.ErrorCode.ShouldBe("NoFwlibBackend");
});
}
}
}

265
tests/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/IpcFocasClientTests.cs Normal file
View File

@@ -0,0 +1,265 @@
using MessagePack;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Ipc;
using ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Shared;
using ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Shared.Contracts;
namespace ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests;
/// <summary>
/// End-to-end IPC round-trips over an in-memory loopback: <c>IpcFocasClient</c> talks
/// to a test fake that plays the Host's role by reading frames, dispatching on kind,
/// and responding with canned DTOs. Validates that every <see cref="IFocasClient"/>
/// method translates to the right wire frame + decodes the response correctly.
/// </summary>
[Trait("Category", "Unit")]
public sealed class IpcFocasClientTests
{
private const string Secret = "test-secret";
private static async Task ServerLoopAsync(Stream serverSide, Func<FocasMessageKind, byte[], FrameWriter, Task> dispatch, CancellationToken ct)
{
using var reader = new FrameReader(serverSide, leaveOpen: true);
using var writer = new FrameWriter(serverSide, leaveOpen: true);
// Hello handshake.
var first = await reader.ReadFrameAsync(ct);
if (first is null) return;
var hello = MessagePackSerializer.Deserialize<Hello>(first.Value.Body);
var accepted = hello.SharedSecret == Secret;
await writer.WriteAsync(FocasMessageKind.HelloAck,
new HelloAck { Accepted = accepted, RejectReason = accepted ? null : "wrong-secret" }, ct);
if (!accepted) return;
while (!ct.IsCancellationRequested)
{
var frame = await reader.ReadFrameAsync(ct);
if (frame is null) return;
await dispatch(frame.Value.Kind, frame.Value.Body, writer);
}
}
[Fact]
public async Task Connect_sends_OpenSessionRequest_and_caches_session_id()
{
await using var loop = new IpcLoopback();
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
OpenSessionRequest? received = null;
var server = Task.Run(() => ServerLoopAsync(loop.ServerSide, async (kind, body, writer) =>
{
if (kind == FocasMessageKind.OpenSessionRequest)
{
received = MessagePackSerializer.Deserialize<OpenSessionRequest>(body);
await writer.WriteAsync(FocasMessageKind.OpenSessionResponse,
new OpenSessionResponse { Success = true, SessionId = 42 }, cts.Token);
}
}, cts.Token));
var ipc = await FocasIpcClient.ConnectAsync(loop.ClientSide, Secret, cts.Token);
var client = new IpcFocasClient(ipc, FocasCncSeries.Thirty_i);
await client.ConnectAsync(new FocasHostAddress("192.168.1.50", 8193), TimeSpan.FromSeconds(2), cts.Token);
client.IsConnected.ShouldBeTrue();
received.ShouldNotBeNull();
received!.HostAddress.ShouldBe("192.168.1.50:8193");
received.CncSeries.ShouldBe((int)FocasCncSeries.Thirty_i);
cts.Cancel();
try { await server; } catch { }
}
[Fact]
public async Task Connect_throws_when_host_rejects()
{
await using var loop = new IpcLoopback();
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
var server = Task.Run(() => ServerLoopAsync(loop.ServerSide, async (kind, body, writer) =>
{
if (kind == FocasMessageKind.OpenSessionRequest)
{
await writer.WriteAsync(FocasMessageKind.OpenSessionResponse,
new OpenSessionResponse { Success = false, Error = "unreachable", ErrorCode = "EW_SOCKET" }, cts.Token);
}
}, cts.Token));
var ipc = await FocasIpcClient.ConnectAsync(loop.ClientSide, Secret, cts.Token);
var client = new IpcFocasClient(ipc);
await Should.ThrowAsync<InvalidOperationException>(async () =>
await client.ConnectAsync(new FocasHostAddress("10.0.0.1", 8193), TimeSpan.FromSeconds(1), cts.Token));
cts.Cancel();
try { await server; } catch { }
}
[Fact]
public async Task Read_sends_ReadRequest_and_decodes_response()
{
await using var loop = new IpcLoopback();
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
ReadRequest? received = null;
var server = Task.Run(() => ServerLoopAsync(loop.ServerSide, async (kind, body, writer) =>
{
switch (kind)
{
case FocasMessageKind.OpenSessionRequest:
await writer.WriteAsync(FocasMessageKind.OpenSessionResponse,
new OpenSessionResponse { Success = true, SessionId = 1 }, cts.Token);
break;
case FocasMessageKind.ReadRequest:
received = MessagePackSerializer.Deserialize<ReadRequest>(body);
await writer.WriteAsync(FocasMessageKind.ReadResponse,
new ReadResponse
{
Success = true,
StatusCode = 0,
ValueBytes = MessagePackSerializer.Serialize((int)12345),
ValueTypeCode = FocasDataTypeCode.Int32,
}, cts.Token);
break;
}
}, cts.Token));
var ipc = await FocasIpcClient.ConnectAsync(loop.ClientSide, Secret, cts.Token);
var client = new IpcFocasClient(ipc);
await client.ConnectAsync(new FocasHostAddress("h", 8193), TimeSpan.FromSeconds(1), cts.Token);
var addr = new FocasAddress(FocasAreaKind.Parameter, null, 1815, null);
var (value, status) = await client.ReadAsync(addr, FocasDataType.Int32, cts.Token);
status.ShouldBe(0u);
value.ShouldBe(12345);
received!.Address.Number.ShouldBe(1815);
cts.Cancel();
try { await server; } catch { }
}
[Fact]
public async Task Write_sends_WriteRequest_and_returns_status()
{
await using var loop = new IpcLoopback();
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
var server = Task.Run(() => ServerLoopAsync(loop.ServerSide, async (kind, body, writer) =>
{
switch (kind)
{
case FocasMessageKind.OpenSessionRequest:
await writer.WriteAsync(FocasMessageKind.OpenSessionResponse,
new OpenSessionResponse { Success = true, SessionId = 1 }, cts.Token);
break;
case FocasMessageKind.WriteRequest:
var req = MessagePackSerializer.Deserialize<WriteRequest>(body);
MessagePackSerializer.Deserialize<double>(req.ValueBytes!).ShouldBe(3.14);
await writer.WriteAsync(FocasMessageKind.WriteResponse,
new WriteResponse { Success = true, StatusCode = 0 }, cts.Token);
break;
}
}, cts.Token));
var ipc = await FocasIpcClient.ConnectAsync(loop.ClientSide, Secret, cts.Token);
var client = new IpcFocasClient(ipc);
await client.ConnectAsync(new FocasHostAddress("h", 8193), TimeSpan.FromSeconds(1), cts.Token);
var status = await client.WriteAsync(new FocasAddress(FocasAreaKind.Macro, null, 500, null),
FocasDataType.Float64, 3.14, cts.Token);
status.ShouldBe(0u);
cts.Cancel();
try { await server; } catch { }
}
[Fact]
public async Task Write_pmc_bit_sends_first_class_RMW_frame()
{
await using var loop = new IpcLoopback();
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
PmcBitWriteRequest? received = null;
var server = Task.Run(() => ServerLoopAsync(loop.ServerSide, async (kind, body, writer) =>
{
switch (kind)
{
case FocasMessageKind.OpenSessionRequest:
await writer.WriteAsync(FocasMessageKind.OpenSessionResponse,
new OpenSessionResponse { Success = true, SessionId = 1 }, cts.Token);
break;
case FocasMessageKind.PmcBitWriteRequest:
received = MessagePackSerializer.Deserialize<PmcBitWriteRequest>(body);
await writer.WriteAsync(FocasMessageKind.PmcBitWriteResponse,
new PmcBitWriteResponse { Success = true, StatusCode = 0 }, cts.Token);
break;
}
}, cts.Token));
var ipc = await FocasIpcClient.ConnectAsync(loop.ClientSide, Secret, cts.Token);
var client = new IpcFocasClient(ipc);
await client.ConnectAsync(new FocasHostAddress("h", 8193), TimeSpan.FromSeconds(1), cts.Token);
var addr = new FocasAddress(FocasAreaKind.Pmc, "R", 100, BitIndex: 5);
var status = await client.WriteAsync(addr, FocasDataType.Bit, true, cts.Token);
status.ShouldBe(0u);
received.ShouldNotBeNull();
received!.BitIndex.ShouldBe(5);
received.Value.ShouldBeTrue();
received.Address.PmcLetter.ShouldBe("R");
cts.Cancel();
try { await server; } catch { }
}
[Fact]
public async Task Probe_round_trips_health_from_host()
{
await using var loop = new IpcLoopback();
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
var server = Task.Run(() => ServerLoopAsync(loop.ServerSide, async (kind, body, writer) =>
{
switch (kind)
{
case FocasMessageKind.OpenSessionRequest:
await writer.WriteAsync(FocasMessageKind.OpenSessionResponse,
new OpenSessionResponse { Success = true, SessionId = 1 }, cts.Token);
break;
case FocasMessageKind.ProbeRequest:
await writer.WriteAsync(FocasMessageKind.ProbeResponse,
new ProbeResponse { Healthy = true, ObservedAtUtcUnixMs = 1_700_000_000_000 }, cts.Token);
break;
}
}, cts.Token));
var ipc = await FocasIpcClient.ConnectAsync(loop.ClientSide, Secret, cts.Token);
var client = new IpcFocasClient(ipc);
await client.ConnectAsync(new FocasHostAddress("h", 8193), TimeSpan.FromSeconds(1), cts.Token);
(await client.ProbeAsync(cts.Token)).ShouldBeTrue();
cts.Cancel();
try { await server; } catch { }
}
[Fact]
public async Task Error_response_from_host_surfaces_as_FocasIpcException()
{
await using var loop = new IpcLoopback();
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
var server = Task.Run(() => ServerLoopAsync(loop.ServerSide, async (kind, body, writer) =>
{
await writer.WriteAsync(FocasMessageKind.ErrorResponse,
new ErrorResponse { Code = "backend-exception", Message = "simulated" }, cts.Token);
}, cts.Token));
var ipc = await FocasIpcClient.ConnectAsync(loop.ClientSide, Secret, cts.Token);
var client = new IpcFocasClient(ipc);
var ex = await Should.ThrowAsync<FocasIpcException>(async () =>
await client.ConnectAsync(new FocasHostAddress("h", 8193), TimeSpan.FromSeconds(1), cts.Token));
ex.Code.ShouldBe("backend-exception");
cts.Cancel();
try { await server; } catch { }
}
}

72
tests/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/IpcLoopback.cs Normal file
View File

@@ -0,0 +1,72 @@
using System.IO.Pipelines;
namespace ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests;
/// <summary>
/// Bidirectional in-memory stream pair for IPC tests. Two <c>System.IO.Pipelines.Pipe</c>
/// instances — one per direction — exposed as <see cref="System.IO.Stream"/> endpoints
/// via <c>PipeReader.AsStream</c> / <c>PipeWriter.AsStream</c>. Lets the test set up a
/// <c>FocasIpcClient</c> on one end and a minimal fake server loop on the other without
/// standing up a real named pipe.
/// </summary>
internal sealed class IpcLoopback : IAsyncDisposable
{
public Stream ClientSide { get; }
public Stream ServerSide { get; }
public IpcLoopback()
{
var clientToServer = new Pipe();
var serverToClient = new Pipe();
ClientSide = new DuplexPipeStream(serverToClient.Reader.AsStream(), clientToServer.Writer.AsStream());
ServerSide = new DuplexPipeStream(clientToServer.Reader.AsStream(), serverToClient.Writer.AsStream());
}
public async ValueTask DisposeAsync()
{
await ClientSide.DisposeAsync();
await ServerSide.DisposeAsync();
}
private sealed class DuplexPipeStream(Stream read, Stream write) : Stream
{
public override bool CanRead => true;
public override bool CanWrite => true;
public override bool CanSeek => false;
public override long Length => throw new NotSupportedException();
public override long Position
{
get => throw new NotSupportedException();
set => throw new NotSupportedException();
}
public override int Read(byte[] buffer, int offset, int count) => read.Read(buffer, offset, count);
public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken ct) =>
read.ReadAsync(buffer, offset, count, ct);
public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken ct = default) =>
read.ReadAsync(buffer, ct);
public override void Write(byte[] buffer, int offset, int count) => write.Write(buffer, offset, count);
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken ct) =>
write.WriteAsync(buffer, offset, count, ct);
public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken ct = default) =>
write.WriteAsync(buffer, ct);
public override void Flush() => write.Flush();
public override Task FlushAsync(CancellationToken ct) => write.FlushAsync(ct);
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
public override void SetLength(long value) => throw new NotSupportedException();
protected override void Dispose(bool disposing)
{
if (disposing)
{
read.Dispose();
write.Dispose();
}
base.Dispose(disposing);
}
}
}