jellyfin-plugin-enigma2/Tests/RelayStartupTests.cs

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<bool, string> 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<HttpResponseMessage> 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<int> ReadAsync(Memory<byte> 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();
}