using System; using System.Diagnostics; using System.IO; using System.Net; using System.Net.Http; using System.Text.Json; using System.Threading; using System.Threading.Tasks; using Jellyfin.Plugin.Enigma2; public static class RelayStartupTests { public static async Task Run(Action check) { var delayed = new StartupPacketStream(TimeSpan.FromMilliseconds(120)); using (var handler = new StartupHttpHandler(delayed, TimeSpan.FromMilliseconds(70))) using (var relay = new LiveStreamRelay(new HttpClient(handler), 8096, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(300), TimeSpan.FromMilliseconds(50))) { var id = relay.Register("http://receiver/delayed").Split('/')[^1]; var subscriptions = await Task.WhenAll(relay.Subscribe(id, CancellationToken.None), relay.Subscribe(id, CancellationToken.None)); await using var first = subscriptions[0]; await using var second = subscriptions[1]; var one = await first.Reader.ReadAsync(); var two = await second.Reader.ReadAsync(); check(handler.Calls == 1 && one[0] == 0x47 && two[0] == 0x47 && delayed.Delivered > 0, "two subscribers share delayed startup beyond the steady-state read timeout"); } var slowTotal = new StartupPacketStream(TimeSpan.FromMilliseconds(180)); using (var handler = new StartupHttpHandler(slowTotal, TimeSpan.FromMilliseconds(180))) using (var relay = new LiveStreamRelay(new HttpClient(handler), 8096, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(300), TimeSpan.FromMilliseconds(50))) { var id = relay.Register("http://receiver/combined-delay").Split('/')[^1]; var watch = Stopwatch.StartNew(); await ExpectFailure(relay.Subscribe(id, CancellationToken.None)); check(handler.Calls == 1 && handler.HeadersReturned && slowTotal.ReadCancelled && slowTotal.Delivered == 0 && slowTotal.Disposed && watch.Elapsed < TimeSpan.FromSeconds(2), "HTTP headers and first packet use one cumulative startup deadline"); for (var i = 0; i < 20 && !JsonSerializer.Serialize(relay.Status()).Contains("\"ActiveSources\":0"); i++) await Task.Delay(5); check(JsonSerializer.Serialize(relay.Status()).Contains("\"ActiveSources\":0"), "startup timeout closes its source"); } var noHeaders = new StartupPacketStream(TimeSpan.Zero); using (var handler = new StartupHttpHandler(noHeaders, TimeSpan.FromMilliseconds(180))) using (var relay = new LiveStreamRelay(new HttpClient(handler), 8096, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(60), TimeSpan.FromMilliseconds(30))) { var id = relay.Register("http://receiver/silent-headers").Split('/')[^1]; await ExpectFailure(relay.Subscribe(id, CancellationToken.None)); check(handler.Cancelled && !handler.HeadersReturned && noHeaders.Delivered == 0, "silent HTTP headers obey the same bounded startup deadline"); } var stalled = new StartupPacketStream(TimeSpan.Zero, stallAfterFirst: true); using (var handler = new StartupHttpHandler(stalled, TimeSpan.Zero)) using (var relay = new LiveStreamRelay(new HttpClient(handler), 8096, TimeSpan.FromMilliseconds(10), TimeSpan.FromSeconds(2), TimeSpan.FromMilliseconds(50))) { var id = relay.Register("http://receiver/steady-stall").Split('/')[^1]; await using var subscription = await relay.Subscribe(id, CancellationToken.None); check((await subscription.Reader.ReadAsync()).Length > 0, "initial packet makes relay ready"); var watch = Stopwatch.StartNew(); await ExpectFailure(subscription.Reader.ReadAsync().AsTask()); check(stalled.ReadCancelled && stalled.Disposed && stalled.Delivered == 1 && watch.Elapsed < TimeSpan.FromSeconds(1), "a stall after startup retains the shorter steady-state timeout"); } using (var client = new HttpClient(new StartupHttpHandler(new StartupPacketStream(TimeSpan.Zero), TimeSpan.Zero))) { var startupRejected = false; var readRejected = false; try { using var invalid = new LiveStreamRelay(client, 8096, startupTimeout: TimeSpan.Zero); } catch (ArgumentOutOfRangeException) { startupRejected = true; } try { using var invalid = new LiveStreamRelay(client, 8096, readTimeout: TimeSpan.FromMilliseconds(-1)); } catch (ArgumentOutOfRangeException) { readRejected = true; } check(startupRejected && readRejected, "relay rejects non-positive startup and read budgets"); } } private static async Task ExpectFailure(Task task) { try { await task.WaitAsync(TimeSpan.FromSeconds(2)); } catch (IOException) { return; } catch (OperationCanceledException) { return; } throw new Exception("Expected bounded relay failure."); } } public sealed class StartupHttpHandler : HttpMessageHandler { private readonly Stream stream; private readonly TimeSpan headerDelay; public int Calls; public bool HeadersReturned; public bool Cancelled; public StartupHttpHandler(Stream stream, TimeSpan headerDelay) { this.stream = stream; this.headerDelay = headerDelay; } protected override async Task SendAsync(HttpRequestMessage request, CancellationToken token) { Interlocked.Increment(ref Calls); try { await Task.Delay(headerDelay, token); } catch (OperationCanceledException) { Cancelled = true; throw; } HeadersReturned = true; return new HttpResponseMessage(HttpStatusCode.OK) { Content = new StreamContent(stream) }; } protected override void Dispose(bool disposing) { if (disposing) stream.Dispose(); base.Dispose(disposing); } } public sealed class StartupPacketStream : Stream { private readonly TimeSpan firstDelay; private readonly bool stallAfterFirst; public int Delivered; public bool ReadCancelled; public bool Disposed; public StartupPacketStream(TimeSpan firstDelay, bool stallAfterFirst = false) { this.firstDelay = firstDelay; this.stallAfterFirst = stallAfterFirst; } public override async ValueTask ReadAsync(Memory buffer, CancellationToken token = default) { try { await Task.Delay(Delivered == 0 ? firstDelay : stallAfterFirst ? Timeout.InfiniteTimeSpan : TimeSpan.FromMilliseconds(10), token); } catch (OperationCanceledException) { ReadCancelled = true; throw; } var count = Math.Min(buffer.Length, 188 * 256); count -= count % 188; buffer.Span[..count].Clear(); for (var i = 0; i < count; i += 188) buffer.Span[i] = 0x47; Delivered++; return count; } protected override void Dispose(bool disposing) { Disposed = true; base.Dispose(disposing); } public override bool CanRead => true; public override bool CanSeek => false; public override bool CanWrite => false; public override long Length => throw new NotSupportedException(); public override long Position { get => 0; set => throw new NotSupportedException(); } public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); public override void Flush() { } public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); public override void SetLength(long value) => throw new NotSupportedException(); public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); }