diff --git a/ErsatzTV.Core.Tests/Streaming/DirectStreamSessionTrackerTests.cs b/ErsatzTV.Core.Tests/Streaming/DirectStreamSessionTrackerTests.cs new file mode 100644 index 000000000..48649ab23 --- /dev/null +++ b/ErsatzTV.Core.Tests/Streaming/DirectStreamSessionTrackerTests.cs @@ -0,0 +1,98 @@ +using ErsatzTV.Core.Domain; +using ErsatzTV.Core.Interfaces.Streaming; +using ErsatzTV.Core.Streaming; +using NUnit.Framework; +using Shouldly; + +namespace ErsatzTV.Core.Tests.Streaming; + +[TestFixture] +public class DirectStreamSessionTrackerTests +{ + [Test] + public void Should_Track_Concurrent_Viewers_Per_Channel() + { + var tracker = new DirectStreamSessionTracker(); + + using IDisposable session1 = tracker.Register("1", StreamingMode.TransportStream); + using IDisposable session2 = tracker.Register("1", StreamingMode.HttpLiveStreamingDirect); + using IDisposable session3 = tracker.Register("2", StreamingMode.TransportStream); + + tracker.IsActive("1").ShouldBeTrue(); + tracker.GetViewerCount("1").ShouldBe(2); + tracker.GetViewerCount("2").ShouldBe(1); + tracker.GetActiveSessions().Count.ShouldBe(3); + } + + [Test] + public void Should_Remove_Only_Disposed_Session() + { + var tracker = new DirectStreamSessionTracker(); + + IDisposable session1 = tracker.Register("1", StreamingMode.TransportStream); + IDisposable session2 = tracker.Register("1", StreamingMode.TransportStream); + + session1.Dispose(); + + tracker.IsActive("1").ShouldBeTrue(); + tracker.GetViewerCount("1").ShouldBe(1); + + session2.Dispose(); + + tracker.IsActive("1").ShouldBeFalse(); + tracker.GetViewerCount("1").ShouldBe(0); + } + + [Test] + public void Should_Dispose_Registration_Only_Once() + { + var tracker = new DirectStreamSessionTracker(); + + IDisposable session = tracker.Register("1", StreamingMode.TransportStream); + + session.Dispose(); + session.Dispose(); + + tracker.IsActive("1").ShouldBeFalse(); + tracker.GetViewerCount("1").ShouldBe(0); + } + + [Test] + public void Should_Filter_Active_Sessions_By_Channel() + { + var tracker = new DirectStreamSessionTracker(); + + using IDisposable session1 = tracker.Register("1", StreamingMode.TransportStream); + using IDisposable session2 = tracker.Register("2", StreamingMode.HttpLiveStreamingDirect); + + IReadOnlyCollection sessions = tracker.GetActiveSessions("2"); + + sessions.Count.ShouldBe(1); + sessions.Single().ChannelNumber.ShouldBe("2"); + sessions.Single().StreamingMode.ShouldBe(StreamingMode.HttpLiveStreamingDirect); + } + + [Test] + public void Should_Not_Orphan_Session_When_Last_Registration_Is_Removed_During_Register() + { + DirectStreamSessionTracker tracker = null; + IDisposable existingSession = null; + + tracker = new TestDirectStreamSessionTracker(() => existingSession?.Dispose()); + existingSession = tracker.Register("1", StreamingMode.TransportStream); + + using IDisposable newSession = tracker.Register("1", StreamingMode.HttpLiveStreamingDirect); + + tracker.IsActive("1").ShouldBeTrue(); + tracker.GetViewerCount("1").ShouldBe(1); + + IReadOnlyCollection sessions = tracker.GetActiveSessions("1"); + sessions.Count.ShouldBe(1); + sessions.Single().StreamingMode.ShouldBe(StreamingMode.HttpLiveStreamingDirect); + } + + private sealed class TestDirectStreamSessionTracker(Action onRegisteringSession) : DirectStreamSessionTracker + { + protected override void OnRegisteringSession() => onRegisteringSession(); + } +} diff --git a/ErsatzTV.Core/Interfaces/Streaming/IDirectStreamSessionTracker.cs b/ErsatzTV.Core/Interfaces/Streaming/IDirectStreamSessionTracker.cs new file mode 100644 index 000000000..496ff9e05 --- /dev/null +++ b/ErsatzTV.Core/Interfaces/Streaming/IDirectStreamSessionTracker.cs @@ -0,0 +1,13 @@ +using ErsatzTV.Core.Domain; +using ErsatzTV.Core.Streaming; + +namespace ErsatzTV.Core.Interfaces.Streaming; + +public interface IDirectStreamSessionTracker +{ + IDisposable Register(string channelNumber, StreamingMode streamingMode); + bool IsActive(string channelNumber); + int GetViewerCount(string channelNumber); + IReadOnlyCollection GetActiveSessions(); + IReadOnlyCollection GetActiveSessions(string channelNumber); +} diff --git a/ErsatzTV.Core/Streaming/DirectStreamSession.cs b/ErsatzTV.Core/Streaming/DirectStreamSession.cs new file mode 100644 index 000000000..c6d1b2c9a --- /dev/null +++ b/ErsatzTV.Core/Streaming/DirectStreamSession.cs @@ -0,0 +1,9 @@ +using ErsatzTV.Core.Domain; + +namespace ErsatzTV.Core.Streaming; + +public record DirectStreamSession( + Guid Id, + string ChannelNumber, + StreamingMode StreamingMode, + DateTimeOffset StartedAt); diff --git a/ErsatzTV.Core/Streaming/DirectStreamSessionTracker.cs b/ErsatzTV.Core/Streaming/DirectStreamSessionTracker.cs new file mode 100644 index 000000000..e0c8bc18e --- /dev/null +++ b/ErsatzTV.Core/Streaming/DirectStreamSessionTracker.cs @@ -0,0 +1,66 @@ +using System.Collections.Concurrent; +using ErsatzTV.Core.Domain; +using ErsatzTV.Core.Interfaces.Streaming; + +namespace ErsatzTV.Core.Streaming; + +public class DirectStreamSessionTracker : IDirectStreamSessionTracker +{ + private readonly ConcurrentDictionary> _sessions = new(); + + public IDisposable Register(string channelNumber, StreamingMode streamingMode) + { + var session = new DirectStreamSession(Guid.NewGuid(), channelNumber, streamingMode, DateTimeOffset.Now); + + ConcurrentDictionary channelSessions = + _sessions.GetOrAdd(channelNumber, _ => new ConcurrentDictionary()); + + OnRegisteringSession(); + + channelSessions.TryAdd(session.Id, session); + + return new Registration(this, session); + } + + protected virtual void OnRegisteringSession() + { + } + + public bool IsActive(string channelNumber) => GetViewerCount(channelNumber) > 0; + + public int GetViewerCount(string channelNumber) => + _sessions.TryGetValue(channelNumber, out ConcurrentDictionary channelSessions) + ? channelSessions.Count + : 0; + + public IReadOnlyCollection GetActiveSessions() => + _sessions.Values.SelectMany(s => s.Values).ToList(); + + public IReadOnlyCollection GetActiveSessions(string channelNumber) => + _sessions.TryGetValue(channelNumber, out ConcurrentDictionary channelSessions) + ? channelSessions.Values.ToList() + : []; + + private void Remove(DirectStreamSession session) + { + if (!_sessions.TryGetValue(session.ChannelNumber, out ConcurrentDictionary channelSessions)) + { + return; + } + + channelSessions.TryRemove(session.Id, out _); + } + + private sealed class Registration(DirectStreamSessionTracker tracker, DirectStreamSession session) : IDisposable + { + private int _disposed; + + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) == 0) + { + tracker.Remove(session); + } + } + } +} diff --git a/ErsatzTV.Tests/Controllers/TrackedFileStreamResultTests.cs b/ErsatzTV.Tests/Controllers/TrackedFileStreamResultTests.cs new file mode 100644 index 000000000..e05fafba4 --- /dev/null +++ b/ErsatzTV.Tests/Controllers/TrackedFileStreamResultTests.cs @@ -0,0 +1,183 @@ +using ErsatzTV.Controllers; +using ErsatzTV.Core.Domain; +using ErsatzTV.Core.Interfaces.Streaming; +using ErsatzTV.Core.Streaming; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Mvc; +using Microsoft.AspNetCore.Routing; +using Microsoft.Extensions.DependencyInjection; +using NSubstitute; +using NUnit.Framework; +using Shouldly; + +namespace ErsatzTV.Tests.Controllers; + +[TestFixture] +public class TrackedFileStreamResultTests +{ + [Test] + public async Task Should_Track_Session_Only_While_Result_Executes() + { + var tracker = new DirectStreamSessionTracker(); + var stream = new BlockingReadStream(); + TrackedFileStreamResult result = new( + stream, + "video/mp2t", + tracker, + "1", + StreamingMode.TransportStream); + + ActionContext context = GetActionContext(); + + tracker.IsActive("1").ShouldBeFalse(); + + Task execute = result.ExecuteResultAsync(context); + await stream.WaitForRead(); + + tracker.IsActive("1").ShouldBeTrue(); + tracker.GetViewerCount("1").ShouldBe(1); + + stream.Complete(); + await execute; + + tracker.IsActive("1").ShouldBeFalse(); + tracker.GetViewerCount("1").ShouldBe(0); + } + + [Test] + public async Task Should_Remove_Session_When_Response_Stream_Fails() + { + var tracker = new DirectStreamSessionTracker(); + TrackedFileStreamResult result = new( + new ThrowingReadStream(new IOException("stream failed")), + "video/mp2t", + tracker, + "1", + StreamingMode.HttpLiveStreamingDirect); + + Func execute = () => result.ExecuteResultAsync(GetActionContext()); + + await execute.ShouldThrowAsync(); + + tracker.IsActive("1").ShouldBeFalse(); + tracker.GetViewerCount("1").ShouldBe(0); + } + + [Test] + public async Task Should_Remove_Session_When_Response_Stream_Is_Aborted() + { + var tracker = new DirectStreamSessionTracker(); + TrackedFileStreamResult result = new( + new ThrowingReadStream(new OperationCanceledException("client aborted")), + "video/mp2t", + tracker, + "1", + StreamingMode.HttpLiveStreamingDirect); + + await result.ExecuteResultAsync(GetActionContext()); + + tracker.IsActive("1").ShouldBeFalse(); + tracker.GetViewerCount("1").ShouldBe(0); + } + + [Test] + public async Task Should_Not_Track_Session_For_Head_Request() + { + IDirectStreamSessionTracker tracker = Substitute.For(); + var stream = new BlockingReadStream(); + TrackedFileStreamResult result = new( + stream, + "video/mp2t", + tracker, + "1", + StreamingMode.TransportStream); + + await result.ExecuteResultAsync(GetActionContext(HttpMethods.Head)); + + tracker.DidNotReceive().Register(Arg.Any(), Arg.Any()); + } + + private static ActionContext GetActionContext(string method = "GET") + { + var httpContext = new DefaultHttpContext(); + httpContext.Request.Method = method; + httpContext.RequestServices = new ServiceCollection() + .AddLogging() + .AddControllers() + .Services + .BuildServiceProvider(); + httpContext.Response.Body = new MemoryStream(); + + return new ActionContext(httpContext, new RouteData(), new Microsoft.AspNetCore.Mvc.Abstractions.ActionDescriptor()); + } + + private sealed class BlockingReadStream : Stream + { + private readonly TaskCompletionSource _continue = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _readStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => false; + public override long Length => 0; + public override long Position { get; set; } + + public Task WaitForRead() => _readStarted.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + public void Complete() => _continue.SetResult(); + + public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) + { + _readStarted.TrySetResult(); + return ReadAfterContinue(); + } + + public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => + ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask(); + + public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + + public override void Flush() + { + } + + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + + public override void SetLength(long value) => throw new NotSupportedException(); + + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + + private async ValueTask ReadAfterContinue() + { + await _continue.Task; + return 0; + } + } + + private sealed class ThrowingReadStream(Exception exception) : Stream + { + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => false; + public override long Length => 0; + public override long Position { get; set; } + + public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) => + ValueTask.FromException(exception); + + public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => + ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask(); + + public override int Read(byte[] buffer, int offset, int count) => throw exception; + + public override void Flush() + { + } + + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + + public override void SetLength(long value) => throw new NotSupportedException(); + + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + } +} diff --git a/ErsatzTV/Controllers/IptvController.cs b/ErsatzTV/Controllers/IptvController.cs index 21bf3ec93..d70ed71af 100644 --- a/ErsatzTV/Controllers/IptvController.cs +++ b/ErsatzTV/Controllers/IptvController.cs @@ -25,6 +25,7 @@ namespace ErsatzTV.Controllers; [ServiceFilter(typeof(ConditionalIptvAuthorizeFilter))] public class IptvController : StreamingControllerBase { + private readonly IDirectStreamSessionTracker _directStreamSessionTracker; private readonly IFFmpegSegmenterService _ffmpegSegmenterService; private readonly ILogger _logger; private readonly IMediator _mediator; @@ -33,9 +34,11 @@ public class IptvController : StreamingControllerBase IMediator mediator, IGraphicsEngine graphicsEngine, ILogger logger, - IFFmpegSegmenterService ffmpegSegmenterService) + IFFmpegSegmenterService ffmpegSegmenterService, + IDirectStreamSessionTracker directStreamSessionTracker) : base(graphicsEngine, logger) { + _directStreamSessionTracker = directStreamSessionTracker; _mediator = mediator; _logger = logger; _ffmpegSegmenterService = ffmpegSegmenterService; @@ -150,7 +153,14 @@ public class IptvController : StreamingControllerBase } process.Start(); - return new FileStreamResult(process.StandardOutput.BaseStream, "video/mp2t"); + return mode == "ts-legacy" + ? new TrackedFileStreamResult( + process.StandardOutput.BaseStream, + "video/mp2t", + _directStreamSessionTracker, + channelNumber, + StreamingMode.TransportStream) + : new FileStreamResult(process.StandardOutput.BaseStream, "video/mp2t"); }, error => BadRequest(error.Value))); } @@ -353,7 +363,11 @@ public class IptvController : StreamingControllerBase Either result = await _mediator.Send(request); - return GetProcessResponse(result, channelNumber, StreamingMode.HttpLiveStreamingDirect); + return GetProcessResponse( + result, + channelNumber, + StreamingMode.HttpLiveStreamingDirect, + _directStreamSessionTracker); } private string AccessTokenQuery() => string.IsNullOrWhiteSpace(Request.Query["access_token"]) diff --git a/ErsatzTV/Controllers/StreamingControllerBase.cs b/ErsatzTV/Controllers/StreamingControllerBase.cs index 7d08044f2..d95cffcca 100644 --- a/ErsatzTV/Controllers/StreamingControllerBase.cs +++ b/ErsatzTV/Controllers/StreamingControllerBase.cs @@ -16,7 +16,8 @@ public abstract class StreamingControllerBase(IGraphicsEngine graphicsEngine, IL protected IActionResult GetProcessResponse( Either result, string channelNumber, - StreamingMode mode) + StreamingMode mode, + IDirectStreamSessionTracker directStreamSessionTracker = null) { foreach (BaseError error in result.LeftToSeq()) { @@ -30,14 +31,18 @@ public abstract class StreamingControllerBase(IGraphicsEngine graphicsEngine, IL foreach (PlayoutItemProcessModel processModel in result.RightToSeq()) { - return StartPlayout(processModel); + return StartPlayout(processModel, channelNumber, mode, directStreamSessionTracker); } // this will never happen return new NotFoundResult(); } - private FileStreamResult StartPlayout(PlayoutItemProcessModel processModel) + private FileStreamResult StartPlayout( + PlayoutItemProcessModel processModel, + string channelNumber, + StreamingMode mode, + IDirectStreamSessionTracker directStreamSessionTracker) { // for process counter var ffmpegProcess = new FFmpegProcess(); @@ -86,6 +91,10 @@ public abstract class StreamingControllerBase(IGraphicsEngine graphicsEngine, IL null, TaskScheduler.Default); - return new FileStreamResult(pipe.Reader.AsStream(), "video/mp2t"); + Stream stream = pipe.Reader.AsStream(); + + return directStreamSessionTracker is null + ? new FileStreamResult(stream, "video/mp2t") + : new TrackedFileStreamResult(stream, "video/mp2t", directStreamSessionTracker, channelNumber, mode); } } diff --git a/ErsatzTV/Controllers/TrackedFileStreamResult.cs b/ErsatzTV/Controllers/TrackedFileStreamResult.cs new file mode 100644 index 000000000..8f13bfc29 --- /dev/null +++ b/ErsatzTV/Controllers/TrackedFileStreamResult.cs @@ -0,0 +1,27 @@ +using ErsatzTV.Core.Domain; +using ErsatzTV.Core.Interfaces.Streaming; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Mvc; + +namespace ErsatzTV.Controllers; + +public class TrackedFileStreamResult( + Stream fileStream, + string contentType, + IDirectStreamSessionTracker directStreamSessionTracker, + string channelNumber, + StreamingMode streamingMode) : FileStreamResult(fileStream, contentType) +{ + public override async Task ExecuteResultAsync(ActionContext context) + { + if (HttpMethods.IsHead(context.HttpContext.Request.Method)) + { + await base.ExecuteResultAsync(context); + return; + } + + using IDisposable registration = directStreamSessionTracker.Register(channelNumber, streamingMode); + + await base.ExecuteResultAsync(context); + } +} diff --git a/ErsatzTV/Startup.cs b/ErsatzTV/Startup.cs index 71f265cca..8a5dcac03 100644 --- a/ErsatzTV/Startup.cs +++ b/ErsatzTV/Startup.cs @@ -44,6 +44,7 @@ using ErsatzTV.Core.Scheduling.Engine; using ErsatzTV.Core.Scheduling.ScriptedScheduling; using ErsatzTV.Core.Scheduling.YamlScheduling; using ErsatzTV.Core.Search; +using ErsatzTV.Core.Streaming; using ErsatzTV.Core.Trakt; using ErsatzTV.Core.Troubleshooting; using ErsatzTV.FFmpeg.Capabilities; @@ -819,6 +820,7 @@ public class Startup services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); + services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); services.AddSingleton();