335 lines
21 KiB
C#
335 lines
21 KiB
C#
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;
|
|
|
|
/// <summary>Plugin-owned FFmpeg failover. One producer per channel; per-viewer language and lifetime.</summary>
|
|
public sealed class ContinuityStreams : IDisposable
|
|
{
|
|
private readonly string root;
|
|
private Func<string> encoderPath;
|
|
private readonly int port;
|
|
private readonly StartupMessages messages;
|
|
private readonly ILogger logger;
|
|
private readonly object gate = new();
|
|
private readonly Dictionary<string, Source> sources = new();
|
|
private readonly Dictionary<string, Viewer> viewers = new();
|
|
private readonly ConcurrentDictionary<string, Lazy<Task<Dictionary<string, byte[]>>>> 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<ContinuityStreams> logger)
|
|
: this(encoder.EncoderPath, Path.Combine(paths.CachePath, "enigma2-continuity-v1"), host.HttpPort, messages, logger, Plugin.Instance?.Configuration?.PreserveSourceResolution ?? true) { 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, bool preserveResolution = true)
|
|
{
|
|
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<string, byte[]> Slates;
|
|
public DvbSubtitleVersions SubtitleVersions;
|
|
public DateTime LastAccess = DateTime.UtcNow;
|
|
}
|
|
public async Task<MediaSourceInfo> Open(string relayUrl, string language, CancellationToken token, BroadcastTracks tracks = null, bool useSlates = true)
|
|
{
|
|
language = StartupMessages.ResolveLanguage(language, null, "en");
|
|
tracks ??= BroadcastTracks.Fallback();
|
|
var profile = useSlates ? LosslessSlateProfile.Create(tracks) : null;
|
|
useSlates = profile != null;
|
|
Dictionary<string, byte[]> clips = null;
|
|
if (useSlates)
|
|
{
|
|
try { clips = await CachedSlates(language, token, tracks); }
|
|
catch (Exception error) when (error is not OperationCanceledException && !token.IsCancellationRequested)
|
|
{ useSlates = false; logger.LogWarning("[Enigma2] Status picture unavailable ({FailureType}); preserving the original broadcast", error.GetType().Name); }
|
|
}
|
|
var sourceKey = relayUrl + "|" + useSlates + "|" + tracks.Key;
|
|
token.ThrowIfCancellationRequested();
|
|
Viewer viewer;
|
|
lock (gate)
|
|
{
|
|
ObjectDisposedException.ThrowIf(disposed, this);
|
|
if (viewers.Count >= 64) throw new InvalidOperationException("Continuity viewer limit reached");
|
|
if (!sources.TryGetValue(sourceKey, out var source))
|
|
{
|
|
if (sources.Count >= 4) throw new InvalidOperationException("Continuity channel limit reached");
|
|
source = new(this, relayUrl, sourceKey, tracks, useSlates); sources.Add(sourceKey, source);
|
|
}
|
|
viewer = new() { Source = source, Slates = clips, SubtitleVersions = new(tracks) }; viewers.Add(viewer.Id, viewer);
|
|
}
|
|
try
|
|
{
|
|
if (!useSlates) await viewer.Source.Ready.Task.WaitAsync(TimeSpan.FromSeconds(StartupBudget.Seconds), token);
|
|
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 = tracks.Output,
|
|
DefaultAudioStreamIndex = tracks.Output.FirstOrDefault(s => s.Type == MediaStreamType.Audio && s.IsDefault)?.Index
|
|
?? tracks.Output.FirstOrDefault(s => s.Type == MediaStreamType.Audio)?.Index,
|
|
DefaultSubtitleStreamIndex = tracks.Output.FirstOrDefault(s => s.Type == MediaStreamType.Subtitle && s.IsDefault)?.Index
|
|
};
|
|
}
|
|
catch { Close(viewer.Id); throw; }
|
|
}
|
|
private async Task<Dictionary<string, byte[]>> CachedSlates(string language, CancellationToken token, BroadcastTracks tracks = null)
|
|
{
|
|
tracks ??= PreviewTracks();
|
|
var key = language + "-" + tracks.Key;
|
|
if (slates.Count >= 64)
|
|
foreach (var stale in slates.Keys.Where(k => k != key).Take(slates.Count - 63)) slates.TryRemove(stale, out _);
|
|
var cached = slates.GetOrAdd(key, _ => new(() => GenerateSlates(language, tracks), LazyThreadSafetyMode.ExecutionAndPublication));
|
|
Dictionary<string, byte[]> clips;
|
|
try { clips = await cached.Value.WaitAsync(token); }
|
|
catch { if (cached.IsValueCreated && cached.Value.IsFaulted) slates.TryRemove(new KeyValuePair<string, Lazy<Task<Dictionary<string, byte[]>>>>(key, cached)); throw; }
|
|
return clips;
|
|
}
|
|
public async Task<byte[]> 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.Key); 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;
|
|
var transport = TransportStreamClock.Rebase(segment.Slate is { } slate ? viewer.Slates[slate] : segment.Video, segment.Start90k,
|
|
alignAudio: segment.Slate != null, offset90k: segment.ClockOffset90k);
|
|
viewer.SubtitleVersions.Apply(transport, sequence);
|
|
return transport;
|
|
}
|
|
}
|
|
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) if (source.UseSlates) 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 BroadcastTracks PreviewTracks() => new(new MediaStream[] {
|
|
new BroadcastMediaStream { Type = MediaStreamType.Video, Index = -1, Codec = "h264", Width = Width, Height = Height,
|
|
Profile = "Constrained Baseline", PixelFormat = "yuv420p", AverageFrameRate = FrameRate, SampleAspectRatio = "1:1", FrameRateRational = "50/1" },
|
|
new MediaStream { Type = MediaStreamType.Audio, Index = -1, Codec = "aac", Profile = "LC", Channels = 2, SampleRate = 48000, IsDefault = true }
|
|
});
|
|
// These arguments encode the generated status image only. Live AV always uses -c copy.
|
|
private static IEnumerable<string> SlateCodecArguments(LosslessSlateProfile profile) => new[] {
|
|
"-c:v", "libx264", "-threads", "2", "-preset", "ultrafast", "-tune", "zerolatency", "-crf", "20",
|
|
"-profile:v", profile.Profile, "-pix_fmt", "yuv420p", "-r", profile.RateArgument,
|
|
"-g", profile.Gop.ToString(CultureInfo.InvariantCulture), "-keyint_min", profile.Gop.ToString(CultureInfo.InvariantCulture), "-sc_threshold", "0",
|
|
"-c:a", "aac", "-ar", "48000", "-ac", "2", "-b:a", "128k" };
|
|
private Process Start(IEnumerable<string> 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<Dictionary<string, byte[]>> GenerateSlates(string language, BroadcastTracks tracks)
|
|
{
|
|
await slateGate.WaitAsync(lifetime.Token);
|
|
var folder = Path.Combine(root, "slates-" + language + "-" + Guid.NewGuid().ToString("N"));
|
|
try
|
|
{
|
|
Directory.CreateDirectory(folder);
|
|
var profile = LosslessSlateProfile.Create(tracks) ?? throw new InvalidOperationException("No lossless status profile");
|
|
var seconds = profile.SegmentSeconds.ToString("0.########", CultureInfo.InvariantCulture);
|
|
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<string, byte[]>();
|
|
foreach (var phase in new[] { "waiting", "retrying" })
|
|
{
|
|
await File.WriteAllTextAsync(Path.Combine(folder, "message.txt"), messages.Text(language, "slate_" + phase), lifetime.Token);
|
|
var size = Math.Max(12, profile.Height / 28);
|
|
var filter = $"setsar={profile.Aspect},drawtext=fontfile={font}:text=TV:fontsize={size}:fontcolor=0x58bfee:x=(w-tw)/2:y=h*0.25," +
|
|
$"drawtext=fontfile={font}:textfile=message.txt:fontsize={size}:fontcolor=white:line_spacing={size / 2}:x=(w-tw)/2:y=h*0.45";
|
|
using var process = Start(new[] { "-f", "lavfi", "-i", $"color=c=0x101723:s={profile.Width}x{profile.Height}:r={profile.RateArgument}", "-f", "lavfi", "-i", "anullsrc=r=48000:cl=stereo", "-vf", filter,
|
|
"-map", "0:v:0" }.Concat(Enumerable.Range(0, tracks.AudioCount).SelectMany(_ => new[] { "-map", "1:a:0" })).Concat(new[] { "-t", (profile.SegmentSeconds * 3).ToString("0.########", CultureInfo.InvariantCulture) }).Concat(SlateCodecArguments(profile)).Concat(tracks.MetadataArguments(audioVideoOnly: true)).Concat(new[] { "-f", "hls", "-hls_time", seconds, "-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] = BroadcastSlateProgram.Apply(await File.ReadAllBytesAsync(Path.Combine(folder, "slate-1.ts"), lifetime.Token), tracks);
|
|
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, Key;
|
|
public readonly BroadcastTracks Tracks;
|
|
public readonly bool UseSlates;
|
|
public TaskCompletionSource Ready { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
public ContinuityTimeline Timeline { get; }
|
|
public volatile bool Retrying;
|
|
public Source(ContinuityStreams owner, string url, string key, BroadcastTracks tracks, bool useSlates)
|
|
{ this.owner = owner; Url = url; Key = key; Tracks = tracks; UseSlates = useSlates; Timeline = new(DateTime.UtcNow, useSlates, LosslessSlateProfile.Create(tracks)?.SegmentSeconds ?? ContinuityTimeline.SegmentSeconds); 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", (StartupBudget.StallSeconds * 1000000L).ToString(CultureInfo.InvariantCulture), "-analyzeduration", "5000000", "-probesize", "4000000", "-i", Url }
|
|
.Concat(Tracks.MapArguments()).Concat(new[] { "-dn" });
|
|
args = args.Concat(new[] { "-c", "copy" });
|
|
// MPEG-TS segmenter preserves DVB/teletext subtitles, unlike the HLS muxer's WebVTT-only subtitle output.
|
|
args = args.Concat(Tracks.MetadataArguments()).Concat(new[] { "-max_interleave_delta", "1000000", "-f", "segment", "-segment_format", "mpegts", "-segment_time", "2.24",
|
|
"-segment_list", "live.m3u8", "-segment_list_type", "m3u8", "-segment_list_size", "6", "-segment_list_flags", "+live", "segment-%09d.ts" });
|
|
Interlocked.Increment(ref owner.attempts);
|
|
using var process = owner.Start(args, folder);
|
|
try
|
|
{
|
|
var highest = -1; // Stream copy has no encoder-priming segment to discard.
|
|
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 > 32 * 1024 * 1024) continue;
|
|
Timeline.Live(await File.ReadAllBytesAsync(path, lifetime.Token), epoch, duration, DateTime.UtcNow);
|
|
highest = index;
|
|
foreach (var obsolete in Directory.EnumerateFiles(folder, "segment-*.ts"))
|
|
if (int.TryParse(Path.GetFileNameWithoutExtension(obsolete).AsSpan(8), out var number) && number < highest - 14) File.Delete(obsolete);
|
|
lastOutput = DateTime.UtcNow; delivered = true; Ready.TrySetResult(); Retrying = false; failures = 0; Interlocked.Increment(ref owner.liveSegments);
|
|
}
|
|
}
|
|
if ((DateTime.UtcNow - lastOutput).TotalSeconds > (delivered ? StartupBudget.StallSeconds + 5 : StartupBudget.Seconds)) 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();
|
|
}
|
|
}
|