using System; using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; using System.Globalization; using System.IO; using System.Linq; using System.Net; using System.Text.RegularExpressions; using System.Threading; using System.Threading.Tasks; using MediaBrowser.Common.Configuration; using MediaBrowser.Controller; using MediaBrowser.Controller.MediaEncoding; using MediaBrowser.Model.Dto; using MediaBrowser.Model.Entities; using MediaBrowser.Model.MediaInfo; using Microsoft.AspNetCore.Authorization; using Microsoft.AspNetCore.Mvc; using Microsoft.Extensions.Logging; namespace Jellyfin.Plugin.Enigma2; /// Plugin-owned FFmpeg failover. One producer per channel; per-viewer language and lifetime. public sealed class ContinuityStreams : IDisposable { private readonly string root; private Func encoderPath; private readonly int port; private readonly StartupMessages messages; private readonly ILogger logger; private readonly object gate = new(); private readonly Dictionary sources = new(); private readonly Dictionary viewers = new(); private readonly ConcurrentDictionary>>> slates = new(); private readonly CancellationTokenSource lifetime = new(); private readonly Task sweeper; private readonly SemaphoreSlim slateGate = new(1, 1); private bool disposed; private long attempts, liveSegments; public const int Width = 1920, Height = 1080, FrameRate = 50; public ContinuityStreams(IMediaEncoder encoder, IApplicationPaths paths, IServerApplicationHost host, StartupMessages messages, ILogger logger) : this(encoder.EncoderPath, Path.Combine(paths.CachePath, "enigma2-continuity-v1"), host.HttpPort, messages, logger) { encoderPath = () => encoder.EncoderPath; } // Explicit inputs support an isolated FFmpeg/Kestrel test without a receiver or server configuration changes. public ContinuityStreams(string encoder, string root, int port, StartupMessages messages, ILogger logger) { encoderPath = () => encoder; this.root = Path.GetFullPath(root); this.port = port; this.messages = messages; this.logger = logger; Directory.CreateDirectory(this.root); sweeper = Task.Run(Sweep); } private sealed class Viewer { public string Id = Guid.NewGuid().ToString("N"); public Source Source; public Dictionary Slates; public DateTime LastAccess = DateTime.UtcNow; } public async Task Open(string relayUrl, string language, CancellationToken token) { language = StartupMessages.ResolveLanguage(language, null, "en"); var clips = await CachedSlates(language, token); token.ThrowIfCancellationRequested(); Viewer viewer; lock (gate) { ObjectDisposedException.ThrowIf(disposed, this); if (viewers.Count >= 64) throw new InvalidOperationException("Continuity viewer limit reached"); if (!sources.TryGetValue(relayUrl, out var source)) { if (sources.Count >= 4) throw new InvalidOperationException("Continuity channel limit reached"); source = new(this, relayUrl); sources.Add(relayUrl, source); } viewer = new() { Source = source, Slates = clips }; viewers.Add(viewer.Id, viewer); } return new MediaSourceInfo { Id = viewer.Id, Path = $"http://127.0.0.1:{port}/DecoderIntegration/Continuity/{viewer.Id}/index.m3u8", Protocol = MediaProtocol.Http, Container = "hls", IsInfiniteStream = true, SupportsProbing = false, SupportsDirectPlay = false, SupportsDirectStream = true, SupportsTranscoding = true, BufferMs = 1000, MediaStreams = new[] { new MediaStream { Type = MediaStreamType.Video, Index = -1, Codec = "h264", Profile = "Constrained Baseline", Level = 42, Width = Width, Height = Height, AverageFrameRate = FrameRate, RealFrameRate = FrameRate, IsInterlaced = false, BitRate = 12000000, PixelFormat = "yuv420p" }, new MediaStream { Type = MediaStreamType.Audio, Index = -1, Codec = "aac", Channels = 2, SampleRate = 48000, BitRate = 128000, IsDefault = true } } }; } private async Task> CachedSlates(string language, CancellationToken token) { var cached = slates.GetOrAdd(language, lang => new(() => GenerateSlates(lang), LazyThreadSafetyMode.ExecutionAndPublication)); Dictionary clips; try { clips = await cached.Value.WaitAsync(token); } catch { if (cached.IsValueCreated && cached.Value.IsFaulted) slates.TryRemove(new KeyValuePair>>>(language, cached)); throw; } return clips; } public async Task SlatePreview(string language, string phase, CancellationToken token) { language = StartupMessages.ResolveLanguage(language, null, "en"); if (phase != "waiting" && phase != "retrying") throw new ArgumentException("Unknown slate phase"); return TransportStreamClock.Rebase((await CachedSlates(language, token))[phase], 126000); } public void Close(string id) { lock (gate) { if (!viewers.Remove(id, out var viewer)) return; if (!viewers.Values.Any(v => ReferenceEquals(v.Source, viewer.Source))) { sources.Remove(viewer.Source.Url); viewer.Source.Stop(); } } } public string Playlist(string id) { lock (gate) { if (!viewers.TryGetValue(id, out var viewer)) return null; viewer.LastAccess = DateTime.UtcNow; return viewer.Source.Timeline.Playlist(); } } public byte[] Segment(string id, long sequence) { lock (gate) { if (!viewers.TryGetValue(id, out var viewer)) return null; viewer.LastAccess = DateTime.UtcNow; var segment = viewer.Source.Timeline.Find(sequence); if (segment == null) return null; return TransportStreamClock.Rebase(segment.Slate is { } slate ? viewer.Slates[slate] : segment.Video, segment.Start90k); } } public object Status() { lock (gate) return new { Sources = sources.Count, Viewers = viewers.Count, Retrying = sources.Values.Count(s => s.Retrying), ProducerAttempts = Interlocked.Read(ref attempts), LiveSegmentsProduced = Interlocked.Read(ref liveSegments) }; } private async Task Sweep() { try { while (!lifetime.IsCancellationRequested) { await Task.Delay(100, lifetime.Token); lock (gate) { foreach (var id in viewers.Values.Where(v => DateTime.UtcNow - v.LastAccess > TimeSpan.FromSeconds(60)).Select(v => v.Id).ToArray()) Close(id); foreach (var source in sources.Values) source.Timeline.Tick(DateTime.UtcNow, source.Retrying); } } } catch (OperationCanceledException) { } } public void Dispose() { lock (gate) { if (disposed) return; disposed = true; lifetime.Cancel(); foreach (var source in sources.Values) source.Stop(); sources.Clear(); viewers.Clear(); } } private static IEnumerable CodecArguments() => new[] { "-c:v", "libx264", "-threads", "2", "-preset", "ultrafast", "-tune", "zerolatency", "-crf", "20", "-maxrate", "12M", "-bufsize", "24M", "-profile:v", "baseline", "-level:v", "4.2", "-pix_fmt", "yuv420p", "-r", FrameRate.ToString(CultureInfo.InvariantCulture), "-g", "112", "-keyint_min", "112", "-sc_threshold", "0", "-c:a", "aac", "-ar", "48000", "-ac", "2", "-b:a", "128k" }; private Process Start(IEnumerable arguments, string folder) { var info = new ProcessStartInfo(encoderPath()) { UseShellExecute = false, CreateNoWindow = true, RedirectStandardError = true, RedirectStandardOutput = true, WorkingDirectory = folder }; foreach (var arg in new[] { "-hide_banner", "-loglevel", "error", "-nostdin", "-filter_threads", "1", "-threads", "2" }.Concat(arguments)) info.ArgumentList.Add(arg); var process = Process.Start(info) ?? throw new IOException("FFmpeg did not start"); // Drain without retaining lines: FFmpeg errors may contain private input URLs. process.ErrorDataReceived += (_, _) => { }; process.OutputDataReceived += (_, _) => { }; process.BeginErrorReadLine(); process.BeginOutputReadLine(); return process; } private async Task> GenerateSlates(string language) { await slateGate.WaitAsync(lifetime.Token); var folder = Path.Combine(root, "slates-" + language + "-" + Guid.NewGuid().ToString("N")); try { Directory.CreateDirectory(folder); var font = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"; if (!File.Exists(font)) throw new FileNotFoundException("Continuity slates require a Unicode DejaVu Sans font"); var clips = new Dictionary(); foreach (var phase in new[] { "waiting", "retrying" }) { await File.WriteAllTextAsync(Path.Combine(folder, "message.txt"), messages.Text(language, "slate_" + phase), lifetime.Token); var filter = $"drawbox=x=(iw-160)/2:y=260:w=160:h=100:color=0x58bfee:t=6,drawtext=fontfile={font}:text=TV:fontsize=42:fontcolor=0x58bfee:x=(w-tw)/2:y=285," + $"drawtext=fontfile={font}:textfile=message.txt:fontsize=42:fontcolor=white:line_spacing=22:x=(w-tw)/2:y=490"; using var process = Start(new[] { "-f", "lavfi", "-i", $"color=c=0x101723:s={Width}x{Height}:r={FrameRate}", "-f", "lavfi", "-i", "anullsrc=r=48000:cl=stereo", "-vf", filter, "-map", "0:v:0", "-map", "1:a:0", "-t", "6.72" }.Concat(CodecArguments()).Concat(new[] { "-f", "hls", "-hls_time", "2.24", "-hls_list_size", "0", "-hls_segment_filename", "slate-%d.ts", "slate.m3u8" }), folder); try { await process.WaitForExitAsync(lifetime.Token).WaitAsync(TimeSpan.FromSeconds(20), lifetime.Token); } finally { if (!process.HasExited) process.Kill(entireProcessTree: true); } if (process.ExitCode != 0) throw new IOException("FFmpeg slate generation failed"); // Use an interior HLS segment: it has neither AAC encoder priming // nor the final padding of a newly encoded standalone file. clips[phase] = await File.ReadAllBytesAsync(Path.Combine(folder, "slate-1.ts"), lifetime.Token); foreach (var file in Directory.GetFiles(folder, "slate*")) File.Delete(file); } return clips; } finally { try { if (Directory.Exists(folder)) Directory.Delete(folder, recursive: true); } finally { slateGate.Release(); } } } private sealed class Source { private readonly ContinuityStreams owner; private readonly CancellationTokenSource lifetime; private readonly Task task; public readonly string Url; public ContinuityTimeline Timeline { get; } = new(DateTime.UtcNow); public volatile bool Retrying; public Source(ContinuityStreams owner, string url) { this.owner = owner; Url = url; lifetime = CancellationTokenSource.CreateLinkedTokenSource(owner.lifetime.Token); task = Task.Run(Run); } public void Stop() { try { lifetime.Cancel(); } catch (ObjectDisposedException) { } } private async Task Run() { var failures = 0; try { while (!lifetime.IsCancellationRequested) { var epoch = Guid.NewGuid().ToString("N"); var folder = Path.Combine(owner.root, "source-" + epoch); var delivered = false; try { Directory.CreateDirectory(folder); var args = new[] { "-rw_timeout", "10000000", "-analyzeduration", "5000000", "-probesize", "4000000", "-i", Url, "-map", "0:v:0", "-map", "0:a:0", "-sn", "-dn", "-vf", $"bwdif=mode=send_field:parity=auto:deint=interlaced,scale=w={Width}:h={Height}:force_original_aspect_ratio=decrease,pad={Width}:{Height}:(ow-iw)/2:(oh-ih)/2,setsar=1" } .Concat(CodecArguments()).Concat(new[] { "-f", "hls", "-hls_time", "2.24", "-hls_list_size", "6", "-hls_delete_threshold", "8", "-hls_flags", "delete_segments+temp_file+omit_endlist", "-hls_segment_filename", "segment-%09d.ts", "live.m3u8" }); Interlocked.Increment(ref owner.attempts); using var process = owner.Start(args, folder); try { var highest = 0; // Discard the encoder-priming segment. var lastOutput = DateTime.UtcNow; while (!process.HasExited && !lifetime.IsCancellationRequested) { var playlist = Path.Combine(folder, "live.m3u8"); if (File.Exists(playlist)) { var lines = await File.ReadAllLinesAsync(playlist, lifetime.Token); double duration = 0; foreach (var line in lines) { if (line.StartsWith("#EXTINF:", StringComparison.Ordinal)) double.TryParse(line.Substring(8).TrimEnd(','), NumberStyles.Float, CultureInfo.InvariantCulture, out duration); var match = Regex.Match(line, @"^segment-(\d{9})\.ts$"); if (!match.Success || !int.TryParse(match.Groups[1].Value, out var index) || index <= highest) continue; var path = Path.Combine(folder, line); if (!File.Exists(path) || new FileInfo(path).Length > 8 * 1024 * 1024) continue; Timeline.Live(await File.ReadAllBytesAsync(path, lifetime.Token), epoch, duration, DateTime.UtcNow); highest = index; lastOutput = DateTime.UtcNow; delivered = true; Retrying = false; failures = 0; Interlocked.Increment(ref owner.liveSegments); } } if ((DateTime.UtcNow - lastOutput).TotalSeconds > (delivered ? 12 : 35)) throw new IOException("No usable encoded segments"); await Task.Delay(250, lifetime.Token); } } finally { if (!process.HasExited) process.Kill(entireProcessTree: true); await process.WaitForExitAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(5)); if (!lifetime.IsCancellationRequested && process.ExitCode != 0) owner.logger.LogWarning("[Enigma2] Continuity producer exited with code {ExitCode}; retrying while retaining the viewer stream", process.ExitCode); } } catch (Exception error) when (!lifetime.IsCancellationRequested) { owner.logger.LogWarning("[Enigma2] Continuity producer retry ({FailureType}); keeping the viewer stream open", error.GetType().Name); } finally { try { if (Directory.Exists(folder)) Directory.Delete(folder, recursive: true); } catch (IOException) { owner.logger.LogWarning("[Enigma2] Could not remove an inactive continuity cache directory"); } } Retrying = true; failures++; await Task.Delay(TimeSpan.FromSeconds(Math.Min(10, 3 * failures)), lifetime.Token); } } catch (OperationCanceledException) when (lifetime.IsCancellationRequested) { } catch (Exception error) { owner.logger.LogError("[Enigma2] Continuity worker stopped ({FailureType})", error.GetType().Name); } finally { lifetime.Dispose(); } } } } [ApiController] [Route("DecoderIntegration/Continuity")] public sealed class ContinuityController : ControllerBase { private readonly ContinuityStreams streams; public ContinuityController(ContinuityStreams streams) { this.streams = streams; } private bool Local => HttpContext.Connection.RemoteIpAddress is { } remote && (IPAddress.IsLoopback(remote) || remote.IsIPv4MappedToIPv6 && IPAddress.IsLoopback(remote.MapToIPv4())) && !Request.Headers.ContainsKey("Forwarded") && !Request.Headers.ContainsKey("X-Forwarded-For") && !Request.Headers.ContainsKey("X-Real-IP"); [HttpGet("{id}/index.m3u8")] [AllowAnonymous] public IActionResult Playlist(string id) { if (!Local) return StatusCode(403); Response.Headers.CacheControl = "no-store"; return streams.Playlist(id) is { } text ? Content(text, "application/vnd.apple.mpegurl") : NotFound(); } [HttpGet("{id}/{sequence:long}.ts")] [AllowAnonymous] public IActionResult Segment(string id, long sequence) { if (!Local) return StatusCode(403); Response.Headers.CacheControl = "no-store"; return streams.Segment(id, sequence) is { } bytes ? File(bytes, "video/mp2t") : NotFound(); } }