Share decoder live sources and bound web playback startup

2026-10-12 20:59:30 +02:00
parent 92052d588f
commit cecd5b5625
13 changed files with 579 additions and 37 deletions
@@ -2,8 +2,8 @@

  <PropertyGroup>
    <TargetFrameworks>net10.0</TargetFrameworks>
-    <AssemblyVersion>7.0.1.0</AssemblyVersion>
-    <FileVersion>7.0.1.0</FileVersion>
+    <AssemblyVersion>7.0.4.0</AssemblyVersion>
+    <FileVersion>7.0.4.0</FileVersion>
  </PropertyGroup>

  <PropertyGroup Condition="'$(Configuration)|$(TargetFramework)|$(Platform)'=='Release|net6.0|AnyCPU'">
@@ -0,0 +1,105 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Globalization;
+using System.Linq;
+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)
+    {
+        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))); 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)
+    {
+        using var process = new Process { StartInfo = new ProcessStartInfo(executable) { RedirectStandardOutput = true, RedirectStandardError = true, UseShellExecute = false, CreateNoWindow = true } };
+        foreach (var argument in new[] { "-v", "error", "-rw_timeout", "8000000", "-seekable", "0", "-skip_estimate_duration_from_pts", "1", "-f", "mpegts", "-analyzeduration", "1500000", "-probesize", "1048576", "-fpsprobesize", "10", "-i", url, "-show_streams", "-of", "json" }) process.StartInfo.ArgumentList.Add(argument);
+        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;
+    }
+}
@@ -0,0 +1,197 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Net;
+using System.Net.Http;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using MediaBrowser.Controller;
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.AspNetCore.Mvc;
+
+namespace Jellyfin.Plugin.Enigma2;
+
+/// <summary>Shares one receiver HTTP connection between probing and all consumers of a channel.</summary>
+public sealed class LiveStreamRelay : IDisposable
+{
+    private readonly HttpClient client;
+    private readonly int port;
+    private readonly Dictionary<string, Source> sources = new();
+    private readonly Dictionary<string, Source> tokens = new();
+    private readonly object gate = new();
+    private int opened;
+    private readonly TimeSpan idleGrace;
+    public LiveStreamRelay(IServerApplicationHost host) : this(new HttpClient(new SocketsHttpHandler { ConnectTimeout = TimeSpan.FromSeconds(3), AllowAutoRedirect = false }), host.HttpPort) { }
+    public LiveStreamRelay(HttpClient client, int port, TimeSpan? idleGrace = null) { this.client = client; client.Timeout = Timeout.InfiniteTimeSpan; this.port = port; this.idleGrace = idleGrace ?? TimeSpan.FromSeconds(15); }
+    public string Register(string url)
+    {
+        lock (gate)
+        {
+            if (!sources.TryGetValue(url, out var source))
+            {
+                source = new Source(this, url);
+                sources.Add(url, source); tokens.Add(source.Id, source);
+            }
+            return $"http://127.0.0.1:{port}/DecoderIntegration/Live/{source.Id}";
+        }
+    }
+    public Task<Subscription> Subscribe(string id, CancellationToken token)
+    {
+        Source source;
+        lock (gate) if (!tokens.TryGetValue(id, out source)) throw new FileNotFoundException();
+        return source.Subscribe(token);
+    }
+    public object Status()
+    {
+        lock (gate) return new { UpstreamConnectionsOpened = opened, ActiveSources = sources.Values.Count(s => s.Active), Consumers = sources.Values.Sum(s => s.Consumers), LastErrors = sources.Values.Select(s => s.LastError).Where(s => s != null).ToArray(), Channels = sources.Values.Select(s => new { s.Id, s.ServiceReference, s.Active, s.Consumers }).ToArray() };
+    }
+    public void Dispose() { lock (gate) foreach (var s in sources.Values) s.Stop(); client.Dispose(); }
+
+    public sealed class Subscription : IAsyncDisposable
+    {
+        private readonly Source source;
+        internal readonly Channel<byte[]> Buffer = Channel.CreateBounded<byte[]>(new BoundedChannelOptions(128) { FullMode = BoundedChannelFullMode.Wait, SingleReader = true });
+        internal Subscription(Source source) { this.source = source; }
+        public ChannelReader<byte[]> Reader => Buffer.Reader;
+        public ValueTask DisposeAsync() { source.Remove(this); return ValueTask.CompletedTask; }
+    }
+
+    internal sealed class Source
+    {
+        private readonly LiveStreamRelay owner;
+        private readonly string url;
+        private readonly object gate = new();
+        private readonly HashSet<Subscription> subscribers = new();
+        private readonly Queue<byte[]> history = new();
+        private int historyBytes;
+        private CancellationTokenSource lifetime;
+        private CancellationTokenSource idle;
+        private Task pump;
+        private TaskCompletionSource ready;
+        private DateTime failedUntil;
+        public string Id { get; } = Guid.NewGuid().ToString("N");
+        public string ServiceReference => new Uri(url).AbsolutePath;
+        public string LastError { get; private set; }
+        public bool Active { get { lock (gate) return pump != null && !pump.IsCompleted; } }
+        public int Consumers { get { lock (gate) return subscribers.Count; } }
+        public Source(LiveStreamRelay owner, string url) { this.owner = owner; this.url = url; }
+        public async Task<Subscription> Subscribe(CancellationToken token)
+        {
+            Subscription subscription; Task initial;
+            lock (gate)
+            {
+                idle?.Cancel(); idle?.Dispose(); idle = null;
+                if (subscribers.Count >= 16) throw new IOException("Channel consumer limit reached.");
+                if (pump == null || pump.IsCompleted)
+                {
+                    if (DateTime.UtcNow < failedUntil) throw new IOException("Decoder source is unavailable.");
+                    lifetime?.Dispose(); lifetime = new();
+                    history.Clear(); historyBytes = 0;
+                    ready = new(TaskCreationOptions.RunContinuationsAsynchronously);
+                    pump = Task.Run(() => Pump(lifetime.Token));
+                }
+                subscription = new(this); subscribers.Add(subscription);
+                foreach (var chunk in history) subscription.Buffer.Writer.TryWrite(chunk);
+                initial = ready.Task;
+            }
+            try { await initial.WaitAsync(TimeSpan.FromSeconds(12), token); return subscription; }
+            catch { await subscription.DisposeAsync(); throw; }
+        }
+        public void Remove(Subscription subscription)
+        {
+            lock (gate)
+            {
+                if (!subscribers.Remove(subscription)) return;
+                subscription.Buffer.Writer.TryComplete();
+                if (subscribers.Count == 0 && pump != null && !pump.IsCompleted)
+                {
+                    idle?.Cancel(); idle?.Dispose(); idle = new();
+                    var timer = idle.Token;
+                    var current = lifetime;
+                    _ = Task.Run(async () => {
+                        try { await Task.Delay(owner.idleGrace, timer); lock (gate) if (subscribers.Count == 0 && lifetime == current) current.Cancel(); }
+                        catch (OperationCanceledException) { }
+                    });
+                }
+            }
+        }
+        public void Stop() { lock (gate) { idle?.Cancel(); lifetime?.Cancel(); } }
+        private async Task Pump(CancellationToken token)
+        {
+            Exception failure = null;
+            try
+            {
+                using var opening = CancellationTokenSource.CreateLinkedTokenSource(token); opening.CancelAfter(TimeSpan.FromSeconds(5));
+                Interlocked.Increment(ref owner.opened);
+                using var response = await owner.client.GetAsync(url, HttpCompletionOption.ResponseHeadersRead, opening.Token);
+                response.EnsureSuccessStatusCode();
+                using var stream = await response.Content.ReadAsStreamAsync(token);
+                var buffer = new byte[188 * 256 + 188]; var carry = 0;
+                while (true)
+                {
+                    using var reading = CancellationTokenSource.CreateLinkedTokenSource(token); reading.CancelAfter(TimeSpan.FromSeconds(8));
+                    var count = await stream.ReadAtLeastAsync(buffer.AsMemory(carry, buffer.Length - carry), 188 * 256 - carry, throwOnEndOfStream: false, cancellationToken: reading.Token);
+                    if (count == 0) throw new IOException("Decoder closed the live stream.");
+                    count += carry; var bytes = count / 188 * 188;
+                    if (bytes > 0)
+                    {
+                        if (buffer[0] != 0x47 || (bytes >= 376 && buffer[188] != 0x47)) throw new IOException("Decoder did not return MPEG-TS packets.");
+                        var chunk = buffer.AsSpan(0, bytes).ToArray();
+                        lock (gate)
+                        {
+                            history.Enqueue(chunk); historyBytes += bytes;
+                            while (historyBytes > 2 * 1024 * 1024) historyBytes -= history.Dequeue().Length;
+                            foreach (var subscriber in subscribers.ToArray())
+                                if (!subscriber.Buffer.Writer.TryWrite(chunk)) { subscriber.Buffer.Writer.TryComplete(new IOException("Slow live stream consumer.")); Remove(subscriber); }
+                            ready.TrySetResult();
+                        }
+                    }
+                    carry = count - bytes;
+                    if (carry > 0) System.Buffer.BlockCopy(buffer, bytes, buffer, 0, carry);
+                }
+            }
+            catch (Exception error) { failure = error; }
+            finally
+            {
+                lock (gate)
+                {
+                    LastError = token.IsCancellationRequested ? null : failure?.Message;
+                    if (!token.IsCancellationRequested) failedUntil = DateTime.UtcNow.AddSeconds(3);
+                    ready.TrySetException(new IOException("Decoder stream unavailable.", failure));
+                    foreach (var subscriber in subscribers) subscriber.Buffer.Writer.TryComplete(failure);
+                    subscribers.Clear(); history.Clear(); historyBytes = 0;
+                }
+            }
+        }
+    }
+}
+
+[ApiController]
+[Route("DecoderIntegration/Live")]
+public sealed class LiveRelayController : ControllerBase
+{
+    private readonly LiveStreamRelay relay;
+    public LiveRelayController(LiveStreamRelay relay) { this.relay = relay; }
+    [HttpGet("{id}")]
+    [AllowAnonymous]
+    public async Task Stream(string id, CancellationToken token)
+    {
+        var remote = HttpContext.Connection.RemoteIpAddress;
+        if (remote == null || !(IPAddress.IsLoopback(remote) || (remote.IsIPv4MappedToIPv6 && IPAddress.IsLoopback(remote.MapToIPv4())))) { Response.StatusCode = 403; return; }
+        try
+        {
+            await using var subscription = await relay.Subscribe(id, token);
+            Response.ContentType = "video/mp2t"; Response.Headers.CacheControl = "no-store";
+            await foreach (var chunk in subscription.Reader.ReadAllAsync(token))
+            { await Response.Body.WriteAsync(chunk, token); await Response.Body.FlushAsync(token); }
+        }
+        catch (OperationCanceledException) when (token.IsCancellationRequested) { }
+        catch (Exception)
+        {
+            if (Response.HasStarted) HttpContext.Abort();
+            else Response.StatusCode = 502;
+        }
+    }
+}
@@ -30,6 +30,8 @@ namespace Jellyfin.Plugin.Enigma2
    {
        private readonly ILogger<LiveTvService> _logger;
        private int _liveStreams;
+        private readonly LiveStreamProbe _streamProbe;
+        private readonly LiveStreamRelay _streamRelay;

        private string tvBouquetSRef;
        private List<ChannelInfo> tvChannelInfos = new List<ChannelInfo>();
@@ -40,10 +42,12 @@ namespace Jellyfin.Plugin.Enigma2

        public static LiveTvService Instance { get; private set; }

-        public LiveTvService(ILogger<LiveTvService> logger, IHttpClientFactory httpClientFactory, InternetEpg internetEpg)
+        public LiveTvService(ILogger<LiveTvService> logger, IHttpClientFactory httpClientFactory, InternetEpg internetEpg, LiveStreamProbe streamProbe, LiveStreamRelay streamRelay)
        {
            _logger = logger;
            _httpClientFactory = httpClientFactory;
+            _streamProbe = streamProbe;
+            _streamRelay = streamRelay;
            Instance = this;
        }

@@ -1167,34 +1171,20 @@ namespace Jellyfin.Plugin.Enigma2
                trancodingUrl = "?bitrate=1000000?width=1280?height=720?vcodec=h264?aspectratio=2?interlaced=0.mp4";
            }

-            _liveStreams++;
            var streamUrl = string.Format("{0}/{1}{2}", baseUrl, channelOid, trancodingUrl);
-            UtilsHelper.DebugInformation(_logger, string.Format("[Enigma2] GetChannelStream url: {0}", streamUrl));
+            streamUrl = _streamRelay.Register(streamUrl);
+            var streams = await _streamProbe.Get(streamUrl, cancellationToken).ConfigureAwait(false);

            return new MediaSourceInfo
            {
-                Id = _liveStreams.ToString(CultureInfo.InvariantCulture),
+                Id = Interlocked.Increment(ref _liveStreams).ToString(CultureInfo.InvariantCulture),
                Path = streamUrl,
                Protocol = MediaProtocol.Http,
-                MediaStreams = new List<MediaStream>
-                        {
-                            new MediaStream
-                            {
-                                Type = MediaStreamType.Video,
-                                // Set the index to -1 because we don't know the exact index of the video stream within the container
-                                Index = -1,
-
-                                // Set to true if unknown to enable deinterlacing
-                                IsInterlaced = true
-
-                            },
-                            new MediaStream
-                            {
-                                Type = MediaStreamType.Audio,
-                                // Set the index to -1 because we don't know the exact index of the audio stream within the container
-                                Index = -1
-                            }
-                        }
+                Container = Plugin.Instance.Configuration.TranscodedStream ? "mpegts" : "ts",
+                IsInfiniteStream = true,
+                SupportsProbing = false,
+                SupportsDirectPlay = false,
+                MediaStreams = streams
            };
            throw new ResourceNotFoundException(string.Format("Could not stream channel {0}", channelOid));
        }
@@ -50,7 +50,9 @@ public sealed class PreferencesController : ControllerBase
 {
    private readonly PreferencesStore store;
    private readonly InternetEpg epg;
-    public PreferencesController(PreferencesStore store,InternetEpg epg) { this.store=store; this.epg=epg; }
+    private readonly LiveStreamRelay relay;
+    private readonly LiveStreamProbe probe;
+    public PreferencesController(PreferencesStore store,InternetEpg epg,LiveStreamRelay relay,LiveStreamProbe probe) { this.store=store; this.epg=epg; this.relay=relay; this.probe=probe; }
    private Guid UserId => Guid.TryParse(User.FindFirst("Jellyfin-UserId")?.Value,out var id) ? id : Guid.Empty;
    [HttpGet("Channels")]
    [AllowAnonymous]
@@ -66,5 +68,5 @@ public sealed class PreferencesController : ControllerBase
    { if (UserId == Guid.Empty) return Unauthorized(); await store.Save(UserId,value,token); return NoContent(); }
    [HttpGet("Status")]
    [Authorize(Policy="RequiresElevation")]
-    public IActionResult Status() => Ok(new { epg.LastSuccess,epg.MatchedImages,epg.LastError, Source="https://epg.polishteam.one/guide.xml" });
+    public IActionResult Status() => Ok(new { epg.LastSuccess,epg.MatchedImages,epg.LastError, Source="https://epg.polishteam.one/guide.xml", Streams=relay.Status(), ProbeError=probe.LastError });
 }
@@ -19,6 +19,8 @@ namespace Jellyfin.Plugin.Enigma2
        public void RegisterServices(IServiceCollection serviceCollection, IServerApplicationHost applicationHost)
        {
            serviceCollection.AddSingleton<InternetEpg>();
+            serviceCollection.AddSingleton<LiveStreamProbe>();
+            serviceCollection.AddSingleton<LiveStreamRelay>();
            serviceCollection.AddSingleton<PreferencesStore>();
            serviceCollection.AddTransient<Microsoft.AspNetCore.Hosting.IStartupFilter, NativeChannelFilter>();
            serviceCollection.AddSingleton<ILiveTvService, LiveTvService>();