using System.IO.Pipes;
using MessagePack;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Wonderware.Client.Ipc;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Wonderware.Client.Tests;
///
/// In-process fake of the Wonderware historian sidecar. Reuses the client-side framing
/// code (which is byte-identical to the real sidecar) so the wire bytes round-trip
/// correctly without requiring the .NET 4.8 sidecar binary at test time.
///
internal sealed class FakeSidecarServer : IAsyncDisposable
{
private readonly string _pipeName;
private readonly string _expectedSecret;
private readonly CancellationTokenSource _cts = new();
private Task? _loop;
public Func OnReadRaw { get; set; } = _ => new ReadRawReply { Success = true };
public Func OnReadProcessed { get; set; } = _ => new ReadProcessedReply { Success = true };
public Func OnReadAtTime { get; set; } = _ => new ReadAtTimeReply { Success = true };
public Func OnReadEvents { get; set; } = _ => new ReadEventsReply { Success = true };
public Func OnWriteAlarmEvents { get; set; } = req
=> new WriteAlarmEventsReply { Success = true, PerEventOk = Enumerable.Repeat(true, req.Events.Length).ToArray() };
/// Force-disconnect the next accepted client mid-call to exercise reconnect.
public bool DisconnectAfterHandshake { get; set; }
public FakeSidecarServer(string pipeName, string expectedSecret)
{
_pipeName = pipeName;
_expectedSecret = expectedSecret;
}
public string PipeName => _pipeName;
public Task StartAsync()
{
_loop = Task.Run(() => RunAsync(_cts.Token));
// Give the listener a moment to start so client connect doesn't race.
return Task.Delay(50);
}
private async Task RunAsync(CancellationToken ct)
{
while (!ct.IsCancellationRequested)
{
await using var pipe = new NamedPipeServerStream(
_pipeName, PipeDirection.InOut, maxNumberOfServerInstances: 1,
PipeTransmissionMode.Byte, PipeOptions.Asynchronous,
inBufferSize: 64 * 1024, outBufferSize: 64 * 1024);
try { await pipe.WaitForConnectionAsync(ct).ConfigureAwait(false); }
catch (OperationCanceledException) { break; }
try
{
using var reader = new FrameReader(pipe, leaveOpen: true);
using var writer = new FrameWriter(pipe, leaveOpen: true);
// Hello handshake.
var first = await reader.ReadFrameAsync(ct).ConfigureAwait(false);
if (first is null || first.Value.Kind != MessageKind.Hello) continue;
var hello = MessagePackSerializer.Deserialize(first.Value.Body);
if (!string.Equals(hello.SharedSecret, _expectedSecret, StringComparison.Ordinal))
{
await writer.WriteAsync(MessageKind.HelloAck, new HelloAck { Accepted = false, RejectReason = "shared-secret-mismatch" }, ct);
continue;
}
await writer.WriteAsync(MessageKind.HelloAck, new HelloAck { Accepted = true, HostName = "fake-sidecar" }, ct);
if (DisconnectAfterHandshake)
{
DisconnectAfterHandshake = false; // arm once
pipe.Disconnect();
continue;
}
while (!ct.IsCancellationRequested)
{
var frame = await reader.ReadFrameAsync(ct).ConfigureAwait(false);
if (frame is null) break;
switch (frame.Value.Kind)
{
case MessageKind.ReadRawRequest:
{
var req = MessagePackSerializer.Deserialize(frame.Value.Body);
var reply = OnReadRaw(req);
reply.CorrelationId = req.CorrelationId;
await writer.WriteAsync(MessageKind.ReadRawReply, reply, ct);
break;
}
case MessageKind.ReadProcessedRequest:
{
var req = MessagePackSerializer.Deserialize(frame.Value.Body);
var reply = OnReadProcessed(req);
reply.CorrelationId = req.CorrelationId;
await writer.WriteAsync(MessageKind.ReadProcessedReply, reply, ct);
break;
}
case MessageKind.ReadAtTimeRequest:
{
var req = MessagePackSerializer.Deserialize(frame.Value.Body);
var reply = OnReadAtTime(req);
reply.CorrelationId = req.CorrelationId;
await writer.WriteAsync(MessageKind.ReadAtTimeReply, reply, ct);
break;
}
case MessageKind.ReadEventsRequest:
{
var req = MessagePackSerializer.Deserialize(frame.Value.Body);
var reply = OnReadEvents(req);
reply.CorrelationId = req.CorrelationId;
await writer.WriteAsync(MessageKind.ReadEventsReply, reply, ct);
break;
}
case MessageKind.WriteAlarmEventsRequest:
{
var req = MessagePackSerializer.Deserialize(frame.Value.Body);
var reply = OnWriteAlarmEvents(req);
reply.CorrelationId = req.CorrelationId;
await writer.WriteAsync(MessageKind.WriteAlarmEventsReply, reply, ct);
break;
}
}
}
}
catch (OperationCanceledException) { break; }
catch (IOException) { /* peer dropped — accept next */ }
}
}
public async ValueTask DisposeAsync()
{
_cts.Cancel();
if (_loop is not null)
{
try { await _loop.ConfigureAwait(false); } catch { /* ignore shutdown errors */ }
}
_cts.Dispose();
}
}