198 lines
9.9 KiB
C#
198 lines
9.9 KiB
C#
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;
|
|
}
|
|
}
|
|
}
|