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; /// One bounded metadata probe per source, shared by concurrent playback requests. public sealed class LiveStreamProbe { private readonly Func> runner; private readonly Dictionary> 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> runner, TimeSpan? timeout = null) { this.runner = runner; this.timeout = timeout ?? TimeSpan.FromSeconds(15); } public Task> Get(string url, CancellationToken caller, int? programme = null) { if (programme != null) url += "?programme=" + programme.Value.ToString(CultureInfo.InvariantCulture); Task> 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>(TaskCreationOptions.RunContinuationsAsynchronously); task = completion.Task; cache[url] = (task, DateTime.MaxValue); _ = Produce(url, completion); } } return task.WaitAsync(caller); } private async Task Produce(string url, TaskCompletionSource> 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 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 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]+", "")); return await output; } finally { if (!process.HasExited) { process.Kill(entireProcessTree: true); await process.WaitForExitAsync(); } await Task.WhenAll(output, errors); } } public static List Parse(string json) { using var document = JsonDocument.Parse(json); var result = new List(); 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 }; }