Merge pull request 'feat(core): track active sessions for direct streaming modes (#99)' (#121) from feat/99-session-tracking into main

Merge pull request 'feat(core): track active sessions for direct streaming modes (#99)' (#121) from feat/99-session-tracking into main

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<DirectStreamSession> 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<DirectStreamSession> 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();
+    }
+}

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<DirectStreamSession> GetActiveSessions();
+    IReadOnlyCollection<DirectStreamSession> GetActiveSessions(string channelNumber);
+}

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);

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<string, ConcurrentDictionary<Guid, DirectStreamSession>> _sessions = new();
+
+    public IDisposable Register(string channelNumber, StreamingMode streamingMode)
+    {
+        var session = new DirectStreamSession(Guid.NewGuid(), channelNumber, streamingMode, DateTimeOffset.Now);
+
+        ConcurrentDictionary<Guid, DirectStreamSession> channelSessions =
+            _sessions.GetOrAdd(channelNumber, _ => new ConcurrentDictionary<Guid, DirectStreamSession>());
+
+        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<Guid, DirectStreamSession> channelSessions)
+            ? channelSessions.Count
+            : 0;
+
+    public IReadOnlyCollection<DirectStreamSession> GetActiveSessions() =>
+        _sessions.Values.SelectMany(s => s.Values).ToList();
+
+    public IReadOnlyCollection<DirectStreamSession> GetActiveSessions(string channelNumber) =>
+        _sessions.TryGetValue(channelNumber, out ConcurrentDictionary<Guid, DirectStreamSession> channelSessions)
+            ? channelSessions.Values.ToList()
+            : [];
+
+    private void Remove(DirectStreamSession session)
+    {
+        if (!_sessions.TryGetValue(session.ChannelNumber, out ConcurrentDictionary<Guid, DirectStreamSession> 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);
+            }
+        }
+    }
+}

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<Task> execute = () => result.ExecuteResultAsync(GetActionContext());
+
+        await execute.ShouldThrowAsync<IOException>();
+
+        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<IDirectStreamSessionTracker>();
+        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<string>(), Arg.Any<StreamingMode>());
+    }
+
+    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<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
+        {
+            _readStarted.TrySetResult();
+            return ReadAfterContinue();
+        }
+
+        public override Task<int> 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<int> 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<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) =>
+            ValueTask.FromException<int>(exception);
+
+        public override Task<int> 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();
+    }
+}

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<IptvController> _logger;
     private readonly IMediator _mediator;
@@ -33,9 +34,11 @@ public class IptvController : StreamingControllerBase
         IMediator mediator,
         IGraphicsEngine graphicsEngine,
         ILogger<IptvController> 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<BaseError, PlayoutItemProcessModel> 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"])

ErsatzTV/Controllers/StreamingControllerBase.cs

@@ -16,7 +16,8 @@ public abstract class StreamingControllerBase(IGraphicsEngine graphicsEngine, IL
     protected IActionResult GetProcessResponse(
         Either<BaseError, PlayoutItemProcessModel> 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);
     }
 }

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);
+    }
+}

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<IScannerProxyService, ScannerProxyService>();
         services.AddSingleton<IScriptedPlayoutBuilderService, ScriptedPlayoutBuilderService>();
         services.AddSingleton<IFFmpegSegmenterService, FFmpegSegmenterService>();
+        services.AddSingleton<IDirectStreamSessionTracker, DirectStreamSessionTracker>();
         services.AddSingleton<ITempFilePool, TempFilePool>();
         services.AddSingleton<IHlsPlaylistFilter, HlsPlaylistFilter>();
         services.AddSingleton<RecyclableMemoryStreamManager>();