using System.IO.Pipes; using System.Text; using RioJoy.Core.Feedback; using Xunit; namespace RioJoy.Core.Tests.Feedback; public class FeedbackPipeServerTests { private static string UniqueName() => $"riojoy-fb-test-{Guid.NewGuid():N}"; private static byte[] Latin1(string s) => Encoding.GetEncoding(28591).GetBytes(s); /// Collector whose Count/Snapshot are safe against the reader threads. private sealed class Lines { private readonly List _lines = new(); public void Add(string line) { lock (_lines) _lines.Add(line); } public int Count { get { lock (_lines) return _lines.Count; } } public string[] Snapshot() { lock (_lines) return _lines.ToArray(); } } // The accept loop arms asynchronously after Start; retry until it listens. private static NamedPipeClientStream Connect(string name, int timeoutMs = 5000) { var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs); while (true) { var client = new NamedPipeClientStream(".", name, PipeDirection.Out); try { client.Connect(200); return client; } catch (Exception ex) when (ex is IOException or TimeoutException) { client.Dispose(); Assert.True(DateTime.UtcNow < deadline, $"could not connect to {name}: {ex.Message}"); Thread.Sleep(20); } } } private static void Send(NamedPipeClientStream client, string text) { byte[] data = Latin1(text); client.Write(data, 0, data.Length); client.Flush(); } [Fact] public async Task Lines_AreDeliveredInOrder() { string name = UniqueName(); var lines = new Lines(); using var server = new FeedbackPipeServer(name, lines.Add); server.Start(); using NamedPipeClientStream client = Connect(name); Send(client, "lamp 1 dim\r\nplasma clear\n"); await FeedbackWait.For(() => lines.Count >= 2); Assert.Equal(new[] { "lamp 1 dim", "plasma clear" }, lines.Snapshot()); } [Fact] public async Task Line_SplitAcrossWrites_Reassembles() { string name = UniqueName(); var lines = new Lines(); using var server = new FeedbackPipeServer(name, lines.Add); server.Start(); using NamedPipeClientStream client = Connect(name); Send(client, "lamp 0x12 fa"); Send(client, "st bright\n"); await FeedbackWait.For(() => lines.Count >= 1); Assert.Equal("lamp 0x12 fast bright", Assert.Single(lines.Snapshot())); } [Fact] public async Task TwoConcurrentClients_BothDeliver() { string name = UniqueName(); var lines = new Lines(); using var server = new FeedbackPipeServer(name, lines.Add); server.Start(); using NamedPipeClientStream a = Connect(name); using NamedPipeClientStream b = Connect(name); // second instance while A stays connected Send(a, "lamp 1 dim\n"); Send(b, "lamp 2 off\n"); await FeedbackWait.For(() => lines.Count >= 2); Assert.Equal(new[] { "lamp 1 dim", "lamp 2 off" }, lines.Snapshot().OrderBy(l => l)); } [Fact] public async Task ClientDisconnect_ThenReconnect_Works() { string name = UniqueName(); var lines = new Lines(); using var server = new FeedbackPipeServer(name, lines.Add); server.Start(); using (NamedPipeClientStream first = Connect(name)) Send(first, "lamp 1 dim\n"); await FeedbackWait.For(() => lines.Count >= 1); using NamedPipeClientStream second = Connect(name); // server re-arms after the EOF Send(second, "lamp 2 off\n"); await FeedbackWait.For(() => lines.Count >= 2); } [Fact] public async Task ThrowingLineHandler_DoesNotKillTheConnection() { string name = UniqueName(); var lines = new Lines(); using var server = new FeedbackPipeServer(name, line => { if (line.Contains("boom")) throw new InvalidOperationException("handler bug"); lines.Add(line); }); server.Start(); using NamedPipeClientStream client = Connect(name); Send(client, "boom\nlamp 1 dim\n"); await FeedbackWait.For(() => lines.Count >= 1); // the good line still lands Assert.Equal("lamp 1 dim", Assert.Single(lines.Snapshot())); } [Fact] public void Dispose_UnblocksThePendingAccept() { var server = new FeedbackPipeServer(UniqueName(), _ => { }); server.Start(); Thread.Sleep(100); // let the accept loop park in WaitForConnection server.Dispose(); // must not hang on the pending accept (poke-connect) } }