using System.IO.Pipelines;
using System.IO.Pipes;
using AyCode.Core.Serializers.Binaries;
using AyCode.Core.Tests.TestModels;
using static AyCode.Core.Tests.TestModels.AcSerializerModels;
namespace AyCode.Core.Tests.Serialization;
///
/// Cross-platform NamedPipe IPC roundtrip tests for AcBinarySerializer's transport-agnostic
/// streaming helpers (Step 4 of ADR-0003, ACCORE-BIN-T-A3T8).
///
/// The serializer/deserializer surface intentionally has NO NamedPipe-specific helpers —
/// the tests own the /
/// lifecycle directly and call the generic
/// +
/// primitives. This proves
/// the streaming framework works on arbitrary PipeWriter/PipeReader sources
/// (NamedPipe, FileStream, NetworkStream, custom transports) without per-transport adapters in
/// the framework.
///
/// With BufferWriterChunkSize = 256, even small test payloads cross multiple chunk
/// boundaries on the wire — exercises the real chunking + sliding-window cycling behavior.
///
[TestClass]
public class AcBinarySerializerNamedPipeTests
{
[TestMethod]
public async Task RoundTrip_SmallChunkSize_PayloadEquals()
{
var pipeName = $"AcBinaryTest-{Guid.NewGuid():N}";
// 256-byte chunk size = Kestrel slab default; small enough to force multi-chunk framing
// for our 50-item payload, exercises the AsyncSegment chunked wire format end-to-end.
var opts = new AcBinarySerializerOptions { BufferWriterChunkSize = 256 };
var original = CreatePayload(50);
var result = await RunNamedPipeRoundTripAsync(pipeName, original, opts);
Assert.IsNotNull(result);
AssertPayloadEquals(original, result);
}
[TestMethod]
public async Task RoundTrip_LargeScalePayload_ChunkSize256_StructuralEquality()
{
// Production-scale payload via TestDataFactory: 100 root items × 3 pallets × 3 measurements × 4 points
// = ~3700 deeply-nested objects with shared references (50 tags, 20 users, metadata, 10 categories).
// Serialized size ~few hundred KB → many chunks at chunkSize=256 → real backpressure-driven streaming
// (sequential per-chunk flush on StreamPipeWriter, bytes flow incrementally as consumer drains).
#if DEBUG
// Capture BOTH receiver and sender state to diagnose StreamPipeWriter interaction if needed.
var diagLogs = new List();
AsyncPipeReaderInput.DiagnosticLog = msg => diagLogs.Add($"[R] {msg}");
AsyncPipeWriterOutput.DiagnosticLog = msg => diagLogs.Add($"[S] {msg}");
#endif
try
{
var pipeName = $"AcBinaryTest-{Guid.NewGuid():N}";
var opts = new AcBinarySerializerOptions { BufferWriterChunkSize = 256 };
var original = TestDataFactory.CreateLargeScaleBenchmarkOrder(rootItemCount: 100);
var result = await RunNamedPipeRoundTripAsync(pipeName, original, opts);
Assert.IsNotNull(result);
Assert.AreEqual(original.Id, result.Id);
Assert.AreEqual(original.OrderNumber, result.OrderNumber);
Assert.AreEqual(original.Status, result.Status);
Assert.AreEqual(original.TotalAmount, result.TotalAmount);
// Deep structure: count items + pallets + measurements + points must match exactly
var origCounts = CountTestOrderHierarchy(original);
var resultCounts = CountTestOrderHierarchy(result);
Assert.AreEqual(origCounts.items, resultCounts.items, "Items count mismatch");
Assert.AreEqual(origCounts.pallets, resultCounts.pallets, "Pallets count mismatch");
Assert.AreEqual(origCounts.measurements, resultCounts.measurements, "Measurements count mismatch");
Assert.AreEqual(origCounts.points, resultCounts.points, "Points count mismatch");
}
finally
{
#if DEBUG
AsyncPipeReaderInput.DiagnosticLog = null;
AsyncPipeWriterOutput.DiagnosticLog = null;
if (diagLogs.Count > 0)
{
Console.WriteLine($"=== Sender [S] + Receiver [R] DiagnosticLog trail ({diagLogs.Count} entries) ===");
// Print last 60 entries (most relevant to failure point)
var startIdx = Math.Max(0, diagLogs.Count - 60);
if (startIdx > 0) Console.WriteLine($" ... ({startIdx} earlier entries elided)");
for (var i = startIdx; i < diagLogs.Count; i++) Console.WriteLine($" [{i}] {diagLogs[i]}");
Console.WriteLine($"=== End DiagnosticLog ===");
}
#endif
}
}
///
/// Owns the full NamedPipe lifecycle: binds server, accepts connect, drives the generic
/// on
/// the client side and
/// on the server side. The framework helpers know nothing about NamedPipe — only PipeWriter /
/// PipeReader.
///
private static async Task RunNamedPipeRoundTripAsync(string pipeName, T original, AcBinarySerializerOptions opts)
{
// Server-side bind is synchronous (NamedPipeServerStream ctor registers the pipe with
// the OS), so the client can immediately attempt connect once we hand off to async.
await using var pipeServer = new NamedPipeServerStream(pipeName, PipeDirection.In, 1, PipeTransmissionMode.Message, System.IO.Pipes.PipeOptions.Asynchronous);
var receiveTask = Task.Run(async () =>
{
await pipeServer.WaitForConnectionAsync().ConfigureAwait(false);
var pipeReader = PipeReader.Create(pipeServer);
return await AcBinaryDeserializer.DeserializeFromPipeReaderAsync(pipeReader, opts).ConfigureAwait(false);
});
await using var pipeClient = new NamedPipeClientStream(".", pipeName, PipeDirection.Out, System.IO.Pipes.PipeOptions.Asynchronous);
await pipeClient.ConnectAsync().ConfigureAwait(false);
var pipeWriter = PipeWriter.Create(pipeClient);
try
{
// Public PipeWriter overload (raw chunked stream — no per-chunk frame headers,
// bit-compatible with Serialize(v, opts) byte[] output). Auto-selects sequential
// flush strategy because PipeWriter.Create(stream) returns StreamPipeWriter
// (race-incompatible with parallel send).
AcBinarySerializer.SerializeChunked(original, pipeWriter, opts);
}
finally
{
await pipeWriter.CompleteAsync().ConfigureAwait(false);
}
return await receiveTask.ConfigureAwait(false);
}
private static (int items, int pallets, int measurements, int points) CountTestOrderHierarchy(TestOrder order)
{
var items = order.Items.Count;
int pallets = 0, measurements = 0, points = 0;
foreach (var item in order.Items)
{
pallets += item.Pallets.Count;
foreach (var p in item.Pallets)
{
measurements += p.Measurements.Count;
points += p.Measurements.Sum(m => m.Points.Count);
}
}
return (items, pallets, measurements, points);
}
// Note: a "default chunk size" test was deliberately omitted. The default
// AcBinarySerializerOptions.BufferWriterChunkSize used to be 65536, which exceeded the
// UINT16 max (256). Fixed in this work to 256. Tests above explicitly set chunk size
// for reproducibility regardless of default.
private static TestParentWithDateTimeItemCollection CreatePayload(int itemCount)
{
var now = DateTime.UtcNow;
var items = new List(itemCount);
for (var i = 0; i < itemCount; i++)
{
items.Add(new TestEntityWithDateTimeAndInt
{
Id = i + 1,
IntValue = i * 3,
Created = now.AddMinutes(-i),
Modified = now.AddMinutes(i),
StatusCode = i % 4,
Name = $"item-{i}"
});
}
return new TestParentWithDateTimeItemCollection
{
Id = 11,
Name = "named-pipe-roundtrip",
Created = now,
Items = items
};
}
private static void AssertPayloadEquals(TestParentWithDateTimeItemCollection expected, TestParentWithDateTimeItemCollection actual)
{
Assert.AreEqual(expected.Id, actual.Id);
Assert.AreEqual(expected.Name, actual.Name);
Assert.AreEqual(expected.Created, actual.Created);
Assert.IsNotNull(expected.Items);
Assert.IsNotNull(actual.Items);
Assert.AreEqual(expected.Items.Count, actual.Items.Count);
for (var i = 0; i < expected.Items.Count; i++)
{
var e = expected.Items[i];
var a = actual.Items[i];
Assert.AreEqual(e.Id, a.Id);
Assert.AreEqual(e.IntValue, a.IntValue);
Assert.AreEqual(e.Created, a.Created);
Assert.AreEqual(e.Modified, a.Modified);
Assert.AreEqual(e.StatusCode, a.StatusCode);
Assert.AreEqual(e.Name, a.Name);
}
}
}