jellyfin-plugin-enigma2/Jellyfin.Plugin.Enigma2/LiveStreamProbe.cs

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Globalization;
using System.IO;
using System.Linq;
using System.Net.Http;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using MediaBrowser.Controller.MediaEncoding;
using MediaBrowser.Model.Entities;

namespace Jellyfin.Plugin.Enigma2;

/// <summary>One bounded metadata probe per source, shared by concurrent playback requests.</summary>
public sealed class LiveStreamProbe
{
    private readonly Func<string, CancellationToken, Task<string>> runner;
    private readonly Dictionary<string, (Task<List<MediaStream>> Task, DateTime Until)> cache = new();
    private readonly object gate = new();
    private readonly SemaphoreSlim slots = new(8);
    private readonly TimeSpan timeout;
    public string LastError { get; private set; }
    public LiveStreamProbe(IMediaEncoder encoder) : this((url, token) => Run(encoder.ProbePath, url, token)) { }
    public LiveStreamProbe(Func<string, CancellationToken, Task<string>> runner, TimeSpan? timeout = null) { this.runner = runner; this.timeout = timeout ?? TimeSpan.FromSeconds(15); }

    public Task<List<MediaStream>> Get(string url, CancellationToken caller, int? programme = null)
    {
        if (programme != null) url += "?programme=" + programme.Value.ToString(CultureInfo.InvariantCulture);
        Task<List<MediaStream>> task;
        lock (gate)
        {
            if (cache.TryGetValue(url, out var existing) && existing.Until > DateTime.UtcNow) task = existing.Task;
            else
            {
                foreach (var expired in cache.Where(x => x.Value.Until <= DateTime.UtcNow).Select(x => x.Key).ToArray()) cache.Remove(expired);
                var completion = new TaskCompletionSource<List<MediaStream>>(TaskCreationOptions.RunContinuationsAsynchronously);
                task = completion.Task;
                cache[url] = (task, DateTime.MaxValue);
                _ = Produce(url, completion);
            }
        }
        return task.WaitAsync(caller);
    }

    private async Task Produce(string url, TaskCompletionSource<List<MediaStream>> completion)
    {
        var success = false;
        using var deadline = new CancellationTokenSource(timeout);
        try
        {
            await slots.WaitAsync(deadline.Token);
            try { completion.TrySetResult(Parse(await runner(url, deadline.Token))); LastError = null; success = true; }
            finally { slots.Release(); }
        }
        catch (Exception error) { LastError = error.Message; completion.TrySetException(new InvalidOperationException("Decoder returned no usable live stream within the probe deadline.", error)); }
        finally { lock (gate) cache[url] = (completion.Task, DateTime.UtcNow.AddSeconds(success ? 30 : 3)); }
    }

    private static async Task<string> Run(string executable, string url, CancellationToken token)
    {
        // ffprobe may wait for an EPG/data PID forever on live MPEG-TS. Give it a finite
        // sample from the shared relay, preserving the same PID/index ordering.
        var sample = Path.Combine(Path.GetTempPath(), "enigma-probe-" + Guid.NewGuid().ToString("N") + ".ts");
        try
        {
            using var client = new HttpClient(new SocketsHttpHandler { ConnectTimeout = TimeSpan.FromSeconds(3) }) { Timeout = Timeout.InfiniteTimeSpan };
            using var response = await client.GetAsync(url, HttpCompletionOption.ResponseHeadersRead, token);
            response.EnsureSuccessStatusCode();
            using var input = await response.Content.ReadAsStreamAsync(token);
            await using (var file = new FileStream(sample, FileMode.CreateNew, FileAccess.Write, FileShare.None, 65536, useAsync: true))
            {
                using var capture = CancellationTokenSource.CreateLinkedTokenSource(token);
                capture.CancelAfter(TimeSpan.FromSeconds(4));
                var buffer = new byte[188 * 256];
                try
                {
                    while (file.Length < 4 * 1024 * 1024)
                    {
                        var count = await input.ReadAsync(buffer, capture.Token);
                        if (count == 0) break;
                        await file.WriteAsync(buffer.AsMemory(0, count), token);
                    }
                }
                catch (OperationCanceledException) when (!token.IsCancellationRequested && file.Length > 0) { }
                if (file.Length == 0) throw new InvalidOperationException("Decoder returned an empty probe sample.");
            }
            var query = new Uri(url).Query;
            var programme = query.StartsWith("?programme=", StringComparison.Ordinal) && int.TryParse(query.Substring(11), out var value) ? value : (int?)null;
            return await RunFile(executable, sample, programme, token);
        }
        finally { if (File.Exists(sample)) File.Delete(sample); }
    }

    private static async Task<string> RunFile(string executable, string sample, int? programme, CancellationToken token)
    {
        using var process = new Process { StartInfo = new ProcessStartInfo(executable) { RedirectStandardOutput = true, RedirectStandardError = true, UseShellExecute = false, CreateNoWindow = true } };
        foreach (var argument in new[] { "-v", "error", "-f", "mpegts", "-analyzeduration", "3000000", "-probesize", "4194304", "-i", sample, "-show_streams", "-of", "json" }) process.StartInfo.ArgumentList.Add(argument);
        if (programme != null) { process.StartInfo.ArgumentList.Add("-select_streams"); process.StartInfo.ArgumentList.Add("p:" + programme.Value.ToString(CultureInfo.InvariantCulture)); }
        process.Start();
        var output = process.StandardOutput.ReadToEndAsync();
        var errors = process.StandardError.ReadToEndAsync();
        try
        {
            await process.WaitForExitAsync(token);
            var diagnostics = await errors;
            if (process.ExitCode != 0) throw new InvalidOperationException("Decoder stream probe failed: " + System.Text.RegularExpressions.Regex.Replace(diagnostics, @"https?://[^\s]+", "<decoder-stream>"));
            return await output;
        }
        finally
        {
            if (!process.HasExited) { process.Kill(entireProcessTree: true); await process.WaitForExitAsync(); }
            await Task.WhenAll(output, errors);
        }
    }

    public static List<MediaStream> Parse(string json)
    {
        using var document = JsonDocument.Parse(json);
        var result = new List<MediaStream>();
        foreach (var item in document.RootElement.GetProperty("streams").EnumerateArray())
        {
            string Text(string field) => item.TryGetProperty(field, out var value) ? value.ToString() : null;
            int? Number(string field) => int.TryParse(Text(field), out var value) ? value : null;
            var type = Text("codec_type");
            if (type != "video" && type != "audio") continue;
            var codec = Text("codec_name"); var index = Number("index");
            if (string.IsNullOrEmpty(codec) || index == null) continue;
            float? Rate(string field)
            {
                var values = (Text(field) ?? "").Split('/');
                if (!float.TryParse(values[0], NumberStyles.Float, CultureInfo.InvariantCulture, out var numerator)) return null;
                if (values.Length == 1) return numerator;
                return float.TryParse(values[1], NumberStyles.Float, CultureInfo.InvariantCulture, out var denominator) && denominator != 0 ? numerator / denominator : null;
            }
            var stream = new MediaStream { Type = type == "video" ? MediaStreamType.Video : MediaStreamType.Audio, Index = index.Value, Codec = codec, Width = Number("width"), Height = Number("height"), Channels = Number("channels"), SampleRate = Number("sample_rate"), BitRate = Number("bit_rate"), Profile = Text("profile"), Level = Number("level"), PixelFormat = Text("pix_fmt"), ColorRange = Text("color_range"), ColorSpace = Text("color_space"), ColorTransfer = Text("color_transfer"), ColorPrimaries = Text("color_primaries"), AverageFrameRate = Rate("avg_frame_rate"), RealFrameRate = Rate("r_frame_rate"), IsInterlaced = Text("field_order") is "tt" or "bb" or "tb" or "bt", ChannelLayout = Text("channel_layout") };
            if (item.TryGetProperty("tags", out var tags) && tags.TryGetProperty("language", out var language)) stream.Language = language.GetString();
            result.Add(stream);
        }
        if (!result.Any(s => s.Type == MediaStreamType.Video && s.Width > 0 && s.Height > 0)) throw new InvalidOperationException("Decoder stream has no identified video.");
        return result;
    }

    public static MediaStream ParseLiveMetadata(MediaStream s) => new MediaStream
    {
        Index = -1, Type = s.Type, Codec = s.Codec, Width = s.Width, Height = s.Height,
        Channels = s.Channels, SampleRate = s.SampleRate, BitRate = s.BitRate,
        Profile = s.Profile, Level = s.Level, PixelFormat = s.PixelFormat,
        ColorRange = s.ColorRange, ColorSpace = s.ColorSpace, ColorTransfer = s.ColorTransfer,
        ColorPrimaries = s.ColorPrimaries, AverageFrameRate = s.AverageFrameRate,
        RealFrameRate = s.RealFrameRate, IsInterlaced = s.IsInterlaced,
        ChannelLayout = s.ChannelLayout, Language = s.Language
    };
}