Merge pull request #10758 from barronpm/transcode-manager

Add ITranscodeManager

7 files changed, 1039 insertions, 0 deletions

diff --git a/MediaBrowser.Controller/MediaEncoding/ITranscodeManager.cs b/MediaBrowser.Controller/MediaEncoding/ITranscodeManager.cs
new file mode 100644
index 0000000000..c19a12ae7a
--- /dev/null
+++ b/MediaBrowser.Controller/MediaEncoding/ITranscodeManager.cs
@@ -0,0 +1,104 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using MediaBrowser.Controller.Streaming;
+
+namespace MediaBrowser.Controller.MediaEncoding;
+
+/// <summary>
+/// A service for managing media transcoding.
+/// </summary>
+public interface ITranscodeManager
+{
+    /// <summary>
+    /// Get transcoding job.
+    /// </summary>
+    /// <param name="playSessionId">Playback session id.</param>
+    /// <returns>The transcoding job.</returns>
+    public TranscodingJob? GetTranscodingJob(string playSessionId);
+
+    /// <summary>
+    /// Get transcoding job.
+    /// </summary>
+    /// <param name="path">Path to the transcoding file.</param>
+    /// <param name="type">The <see cref="TranscodingJobType"/>.</param>
+    /// <returns>The transcoding job.</returns>
+    public TranscodingJob? GetTranscodingJob(string path, TranscodingJobType type);
+
+    /// <summary>
+    /// Ping transcoding job.
+    /// </summary>
+    /// <param name="playSessionId">Play session id.</param>
+    /// <param name="isUserPaused">Is user paused.</param>
+    /// <exception cref="ArgumentNullException">Play session id is null.</exception>
+    public void PingTranscodingJob(string playSessionId, bool? isUserPaused);
+
+    /// <summary>
+    /// Kills the single transcoding job.
+    /// </summary>
+    /// <param name="deviceId">The device id.</param>
+    /// <param name="playSessionId">The play session identifier.</param>
+    /// <param name="deleteFiles">The delete files.</param>
+    /// <returns>Task.</returns>
+    public Task KillTranscodingJobs(string deviceId, string? playSessionId, Func<string, bool> deleteFiles);
+
+    /// <summary>
+    /// Report the transcoding progress to the session manager.
+    /// </summary>
+    /// <param name="job">The <see cref="TranscodingJob"/> of which the progress will be reported.</param>
+    /// <param name="state">The <see cref="StreamState"/> of the current transcoding job.</param>
+    /// <param name="transcodingPosition">The current transcoding position.</param>
+    /// <param name="framerate">The framerate of the transcoding job.</param>
+    /// <param name="percentComplete">The completion percentage of the transcode.</param>
+    /// <param name="bytesTranscoded">The number of bytes transcoded.</param>
+    /// <param name="bitRate">The bitrate of the transcoding job.</param>
+    public void ReportTranscodingProgress(
+        TranscodingJob job,
+        StreamState state,
+        TimeSpan? transcodingPosition,
+        float? framerate,
+        double? percentComplete,
+        long? bytesTranscoded,
+        int? bitRate);
+
+    /// <summary>
+    /// Starts FFMpeg.
+    /// </summary>
+    /// <param name="state">The state.</param>
+    /// <param name="outputPath">The output path.</param>
+    /// <param name="commandLineArguments">The command line arguments for FFmpeg.</param>
+    /// <param name="userId">The user id.</param>
+    /// <param name="transcodingJobType">The <see cref="TranscodingJobType"/>.</param>
+    /// <param name="cancellationTokenSource">The cancellation token source.</param>
+    /// <param name="workingDirectory">The working directory.</param>
+    /// <returns>Task.</returns>
+    public Task<TranscodingJob> StartFfMpeg(
+        StreamState state,
+        string outputPath,
+        string commandLineArguments,
+        Guid userId,
+        TranscodingJobType transcodingJobType,
+        CancellationTokenSource cancellationTokenSource,
+        string? workingDirectory = null);
+
+    /// <summary>
+    /// Called when [transcode begin request].
+    /// </summary>
+    /// <param name="path">The path.</param>
+    /// <param name="type">The type.</param>
+    /// <returns>The <see cref="TranscodingJob"/>.</returns>
+    public TranscodingJob? OnTranscodeBeginRequest(string path, TranscodingJobType type);
+
+    /// <summary>
+    /// Called when [transcode end].
+    /// </summary>
+    /// <param name="job">The transcode job.</param>
+    public void OnTranscodeEndRequest(TranscodingJob job);
+
+    /// <summary>
+    /// Gets the transcoding lock.
+    /// </summary>
+    /// <param name="outputPath">The output path of the transcoded file.</param>
+    /// <returns>A <see cref="SemaphoreSlim"/>.</returns>
+    public SemaphoreSlim GetTranscodingLock(string outputPath);
+}
diff --git a/MediaBrowser.Controller/MediaEncoding/TranscodingJob.cs b/MediaBrowser.Controller/MediaEncoding/TranscodingJob.cs
new file mode 100644
index 0000000000..1e6d5933c8
--- /dev/null
+++ b/MediaBrowser.Controller/MediaEncoding/TranscodingJob.cs
@@ -0,0 +1,280 @@
+using System;
+using System.Diagnostics;
+using System.Threading;
+using MediaBrowser.Model.Dto;
+using Microsoft.Extensions.Logging;
+
+namespace MediaBrowser.Controller.MediaEncoding;
+
+/// <summary>
+/// Class TranscodingJob.
+/// </summary>
+public sealed class TranscodingJob : IDisposable
+{
+    private readonly ILogger<TranscodingJob> _logger;
+    private readonly object _processLock = new();
+    private readonly object _timerLock = new();
+
+    private Timer? _killTimer;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="TranscodingJob"/> class.
+    /// </summary>
+    /// <param name="logger">Instance of the <see cref="ILogger{TranscodingJobDto}"/> interface.</param>
+    public TranscodingJob(ILogger<TranscodingJob> logger)
+    {
+        _logger = logger;
+    }
+
+    /// <summary>
+    /// Gets or sets the play session identifier.
+    /// </summary>
+    public string? PlaySessionId { get; set; }
+
+    /// <summary>
+    /// Gets or sets the live stream identifier.
+    /// </summary>
+    public string? LiveStreamId { get; set; }
+
+    /// <summary>
+    /// Gets or sets a value indicating whether is live output.
+    /// </summary>
+    public bool IsLiveOutput { get; set; }
+
+    /// <summary>
+    /// Gets or sets the path.
+    /// </summary>
+    public MediaSourceInfo? MediaSource { get; set; }
+
+    /// <summary>
+    /// Gets or sets path.
+    /// </summary>
+    public string? Path { get; set; }
+
+    /// <summary>
+    /// Gets or sets the type.
+    /// </summary>
+    public TranscodingJobType Type { get; set; }
+
+    /// <summary>
+    /// Gets or sets the process.
+    /// </summary>
+    public Process? Process { get; set; }
+
+    /// <summary>
+    /// Gets or sets the active request count.
+    /// </summary>
+    public int ActiveRequestCount { get; set; }
+
+    /// <summary>
+    /// Gets or sets device id.
+    /// </summary>
+    public string? DeviceId { get; set; }
+
+    /// <summary>
+    /// Gets or sets cancellation token source.
+    /// </summary>
+    public CancellationTokenSource? CancellationTokenSource { get; set; }
+
+    /// <summary>
+    /// Gets or sets a value indicating whether has exited.
+    /// </summary>
+    public bool HasExited { get; set; }
+
+    /// <summary>
+    /// Gets or sets exit code.
+    /// </summary>
+    public int ExitCode { get; set; }
+
+    /// <summary>
+    /// Gets or sets a value indicating whether is user paused.
+    /// </summary>
+    public bool IsUserPaused { get; set; }
+
+    /// <summary>
+    /// Gets or sets id.
+    /// </summary>
+    public string? Id { get; set; }
+
+    /// <summary>
+    /// Gets or sets framerate.
+    /// </summary>
+    public float? Framerate { get; set; }
+
+    /// <summary>
+    /// Gets or sets completion percentage.
+    /// </summary>
+    public double? CompletionPercentage { get; set; }
+
+    /// <summary>
+    /// Gets or sets bytes downloaded.
+    /// </summary>
+    public long BytesDownloaded { get; set; }
+
+    /// <summary>
+    /// Gets or sets bytes transcoded.
+    /// </summary>
+    public long? BytesTranscoded { get; set; }
+
+    /// <summary>
+    /// Gets or sets bit rate.
+    /// </summary>
+    public int? BitRate { get; set; }
+
+    /// <summary>
+    /// Gets or sets transcoding position ticks.
+    /// </summary>
+    public long? TranscodingPositionTicks { get; set; }
+
+    /// <summary>
+    /// Gets or sets download position ticks.
+    /// </summary>
+    public long? DownloadPositionTicks { get; set; }
+
+    /// <summary>
+    /// Gets or sets transcoding throttler.
+    /// </summary>
+    public TranscodingThrottler? TranscodingThrottler { get; set; }
+
+    /// <summary>
+    /// Gets or sets last ping date.
+    /// </summary>
+    public DateTime LastPingDate { get; set; }
+
+    /// <summary>
+    /// Gets or sets ping timeout.
+    /// </summary>
+    public int PingTimeout { get; set; }
+
+    /// <summary>
+    /// Stop kill timer.
+    /// </summary>
+    public void StopKillTimer()
+    {
+        lock (_timerLock)
+        {
+            _killTimer?.Change(Timeout.Infinite, Timeout.Infinite);
+        }
+    }
+
+    /// <summary>
+    /// Dispose kill timer.
+    /// </summary>
+    public void DisposeKillTimer()
+    {
+        lock (_timerLock)
+        {
+            if (_killTimer is not null)
+            {
+                _killTimer.Dispose();
+                _killTimer = null;
+            }
+        }
+    }
+
+    /// <summary>
+    /// Start kill timer.
+    /// </summary>
+    /// <param name="callback">Callback action.</param>
+    public void StartKillTimer(Action<object?> callback)
+    {
+        StartKillTimer(callback, PingTimeout);
+    }
+
+    /// <summary>
+    /// Start kill timer.
+    /// </summary>
+    /// <param name="callback">Callback action.</param>
+    /// <param name="intervalMs">Callback interval.</param>
+    public void StartKillTimer(Action<object?> callback, int intervalMs)
+    {
+        if (HasExited)
+        {
+            return;
+        }
+
+        lock (_timerLock)
+        {
+            if (_killTimer is null)
+            {
+                _logger.LogDebug("Starting kill timer at {0}ms. JobId {1} PlaySessionId {2}", intervalMs, Id, PlaySessionId);
+                _killTimer = new Timer(new TimerCallback(callback), this, intervalMs, Timeout.Infinite);
+            }
+            else
+            {
+                _logger.LogDebug("Changing kill timer to {0}ms. JobId {1} PlaySessionId {2}", intervalMs, Id, PlaySessionId);
+                _killTimer.Change(intervalMs, Timeout.Infinite);
+            }
+        }
+    }
+
+    /// <summary>
+    /// Change kill timer if started.
+    /// </summary>
+    public void ChangeKillTimerIfStarted()
+    {
+        if (HasExited)
+        {
+            return;
+        }
+
+        lock (_timerLock)
+        {
+            if (_killTimer is not null)
+            {
+                var intervalMs = PingTimeout;
+
+                _logger.LogDebug("Changing kill timer to {0}ms. JobId {1} PlaySessionId {2}", intervalMs, Id, PlaySessionId);
+                _killTimer.Change(intervalMs, Timeout.Infinite);
+            }
+        }
+    }
+
+    /// <summary>
+    /// Stops the transcoding job.
+    /// </summary>
+    public void Stop()
+    {
+        lock (_processLock)
+        {
+#pragma warning disable CA1849 // Can't await in lock block
+            TranscodingThrottler?.Stop().GetAwaiter().GetResult();
+
+            var process = Process;
+
+            if (!HasExited)
+            {
+                try
+                {
+                    _logger.LogInformation("Stopping ffmpeg process with q command for {Path}", Path);
+
+                    process!.StandardInput.WriteLine("q");
+
+                    // Need to wait because killing is asynchronous.
+                    if (!process.WaitForExit(5000))
+                    {
+                        _logger.LogInformation("Killing FFmpeg process for {Path}", Path);
+                        process.Kill();
+                    }
+                }
+                catch (InvalidOperationException)
+                {
+                }
+            }
+#pragma warning restore CA1849
+        }
+    }
+
+    /// <inheritdoc />
+    public void Dispose()
+    {
+        Process?.Dispose();
+        Process = null;
+        _killTimer?.Dispose();
+        _killTimer = null;
+        CancellationTokenSource?.Dispose();
+        CancellationTokenSource = null;
+        TranscodingThrottler?.Dispose();
+        TranscodingThrottler = null;
+    }
+}
diff --git a/MediaBrowser.Controller/MediaEncoding/TranscodingThrottler.cs b/MediaBrowser.Controller/MediaEncoding/TranscodingThrottler.cs
new file mode 100644
index 0000000000..813f13eaef
--- /dev/null
+++ b/MediaBrowser.Controller/MediaEncoding/TranscodingThrottler.cs
@@ -0,0 +1,218 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using MediaBrowser.Common.Configuration;
+using MediaBrowser.Model.Configuration;
+using MediaBrowser.Model.IO;
+using Microsoft.Extensions.Logging;
+
+namespace MediaBrowser.Controller.MediaEncoding;
+
+/// <summary>
+/// Transcoding throttler.
+/// </summary>
+public class TranscodingThrottler : IDisposable
+{
+    private readonly TranscodingJob _job;
+    private readonly ILogger<TranscodingThrottler> _logger;
+    private readonly IConfigurationManager _config;
+    private readonly IFileSystem _fileSystem;
+    private readonly IMediaEncoder _mediaEncoder;
+    private Timer? _timer;
+    private bool _isPaused;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="TranscodingThrottler"/> class.
+    /// </summary>
+    /// <param name="job">Transcoding job dto.</param>
+    /// <param name="logger">Instance of the <see cref="ILogger{TranscodingThrottler}"/> interface.</param>
+    /// <param name="config">Instance of the <see cref="IConfigurationManager"/> interface.</param>
+    /// <param name="fileSystem">Instance of the <see cref="IFileSystem"/> interface.</param>
+    /// <param name="mediaEncoder">Instance of the <see cref="IMediaEncoder"/> interface.</param>
+    public TranscodingThrottler(TranscodingJob job, ILogger<TranscodingThrottler> logger, IConfigurationManager config, IFileSystem fileSystem, IMediaEncoder mediaEncoder)
+    {
+        _job = job;
+        _logger = logger;
+        _config = config;
+        _fileSystem = fileSystem;
+        _mediaEncoder = mediaEncoder;
+    }
+
+    /// <summary>
+    /// Start timer.
+    /// </summary>
+    public void Start()
+    {
+        _timer = new Timer(TimerCallback, null, 5000, 5000);
+    }
+
+    /// <summary>
+    /// Unpause transcoding.
+    /// </summary>
+    /// <returns>A <see cref="Task"/>.</returns>
+    public async Task UnpauseTranscoding()
+    {
+        if (_isPaused)
+        {
+            _logger.LogDebug("Sending resume command to ffmpeg");
+
+            try
+            {
+                var resumeKey = _mediaEncoder.IsPkeyPauseSupported ? "u" : Environment.NewLine;
+                await _job.Process!.StandardInput.WriteAsync(resumeKey).ConfigureAwait(false);
+                _isPaused = false;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Error resuming transcoding");
+            }
+        }
+    }
+
+    /// <summary>
+    /// Stop throttler.
+    /// </summary>
+    /// <returns>A <see cref="Task"/>.</returns>
+    public async Task Stop()
+    {
+        DisposeTimer();
+        await UnpauseTranscoding().ConfigureAwait(false);
+    }
+
+    /// <summary>
+    /// Dispose throttler.
+    /// </summary>
+    public void Dispose()
+    {
+        Dispose(true);
+        GC.SuppressFinalize(this);
+    }
+
+    /// <summary>
+    /// Dispose throttler.
+    /// </summary>
+    /// <param name="disposing">Disposing.</param>
+    protected virtual void Dispose(bool disposing)
+    {
+        if (disposing)
+        {
+            DisposeTimer();
+        }
+    }
+
+    private EncodingOptions GetOptions()
+    {
+        return _config.GetEncodingOptions();
+    }
+
+    private async void TimerCallback(object? state)
+    {
+        if (_job.HasExited)
+        {
+            DisposeTimer();
+            return;
+        }
+
+        var options = GetOptions();
+
+        if (options.EnableThrottling && IsThrottleAllowed(_job, options.ThrottleDelaySeconds))
+        {
+            await PauseTranscoding().ConfigureAwait(false);
+        }
+        else
+        {
+            await UnpauseTranscoding().ConfigureAwait(false);
+        }
+    }
+
+    private async Task PauseTranscoding()
+    {
+        if (!_isPaused)
+        {
+            var pauseKey = _mediaEncoder.IsPkeyPauseSupported ? "p" : "c";
+
+            _logger.LogDebug("Sending pause command [{Key}] to ffmpeg", pauseKey);
+
+            try
+            {
+                await _job.Process!.StandardInput.WriteAsync(pauseKey).ConfigureAwait(false);
+                _isPaused = true;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Error pausing transcoding");
+            }
+        }
+    }
+
+    private bool IsThrottleAllowed(TranscodingJob job, int thresholdSeconds)
+    {
+        var bytesDownloaded = job.BytesDownloaded;
+        var transcodingPositionTicks = job.TranscodingPositionTicks ?? 0;
+        var downloadPositionTicks = job.DownloadPositionTicks ?? 0;
+
+        var path = job.Path ?? throw new ArgumentException("Path can't be null.");
+
+        var gapLengthInTicks = TimeSpan.FromSeconds(thresholdSeconds).Ticks;
+
+        if (downloadPositionTicks > 0 && transcodingPositionTicks > 0)
+        {
+            // HLS - time-based consideration
+
+            var targetGap = gapLengthInTicks;
+            var gap = transcodingPositionTicks - downloadPositionTicks;
+
+            if (gap < targetGap)
+            {
+                _logger.LogDebug("Not throttling transcoder gap {0} target gap {1}", gap, targetGap);
+                return false;
+            }
+
+            _logger.LogDebug("Throttling transcoder gap {0} target gap {1}", gap, targetGap);
+            return true;
+        }
+
+        if (bytesDownloaded > 0 && transcodingPositionTicks > 0)
+        {
+            // Progressive Streaming - byte-based consideration
+
+            try
+            {
+                var bytesTranscoded = job.BytesTranscoded ?? _fileSystem.GetFileInfo(path).Length;
+
+                // Estimate the bytes the transcoder should be ahead
+                double gapFactor = gapLengthInTicks;
+                gapFactor /= transcodingPositionTicks;
+                var targetGap = bytesTranscoded * gapFactor;
+
+                var gap = bytesTranscoded - bytesDownloaded;
+
+                if (gap < targetGap)
+                {
+                    _logger.LogDebug("Not throttling transcoder gap {0} target gap {1} bytes downloaded {2}", gap, targetGap, bytesDownloaded);
+                    return false;
+                }
+
+                _logger.LogDebug("Throttling transcoder gap {0} target gap {1} bytes downloaded {2}", gap, targetGap, bytesDownloaded);
+                return true;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Error getting output size");
+                return false;
+            }
+        }
+
+        _logger.LogDebug("No throttle data for {Path}", path);
+        return false;
+    }
+
+    private void DisposeTimer()
+    {
+        if (_timer is not null)
+        {
+            _timer.Dispose();
+            _timer = null;
+        }
+    }
+}
diff --git a/MediaBrowser.Controller/Streaming/ProgressiveFileStream.cs b/MediaBrowser.Controller/Streaming/ProgressiveFileStream.cs
new file mode 100644
index 0000000000..f44dc92d71
--- /dev/null
+++ b/MediaBrowser.Controller/Streaming/ProgressiveFileStream.cs
@@ -0,0 +1,182 @@
+using System;
+using System.Diagnostics;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using MediaBrowser.Controller.MediaEncoding;
+using MediaBrowser.Model.IO;
+
+namespace MediaBrowser.Controller.Streaming;
+
+/// <summary>
+/// A progressive file stream for transferring transcoded files as they are written to.
+/// </summary>
+public class ProgressiveFileStream : Stream
+{
+    private readonly Stream _stream;
+    private readonly TranscodingJob? _job;
+    private readonly ITranscodeManager? _transcodeManager;
+    private readonly int _timeoutMs;
+    private bool _disposed;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="ProgressiveFileStream"/> class.
+    /// </summary>
+    /// <param name="filePath">The path to the transcoded file.</param>
+    /// <param name="job">The transcoding job information.</param>
+    /// <param name="transcodeManager">The transcode manager.</param>
+    /// <param name="timeoutMs">The timeout duration in milliseconds.</param>
+    public ProgressiveFileStream(string filePath, TranscodingJob? job, ITranscodeManager transcodeManager, int timeoutMs = 30000)
+    {
+        _job = job;
+        _transcodeManager = transcodeManager;
+        _timeoutMs = timeoutMs;
+
+        _stream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite, IODefaults.FileStreamBufferSize, FileOptions.Asynchronous | FileOptions.SequentialScan);
+    }
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="ProgressiveFileStream"/> class.
+    /// </summary>
+    /// <param name="stream">The stream to progressively copy.</param>
+    /// <param name="timeoutMs">The timeout duration in milliseconds.</param>
+    public ProgressiveFileStream(Stream stream, int timeoutMs = 30000)
+    {
+        _job = null;
+        _transcodeManager = null;
+        _timeoutMs = timeoutMs;
+        _stream = stream;
+    }
+
+    /// <inheritdoc />
+    public override bool CanRead => _stream.CanRead;
+
+    /// <inheritdoc />
+    public override bool CanSeek => false;
+
+    /// <inheritdoc />
+    public override bool CanWrite => false;
+
+    /// <inheritdoc />
+    public override long Length => throw new NotSupportedException();
+
+    /// <inheritdoc />
+    public override long Position
+    {
+        get => throw new NotSupportedException();
+        set => throw new NotSupportedException();
+    }
+
+    /// <inheritdoc />
+    public override void Flush()
+    {
+        // Not supported
+    }
+
+    /// <inheritdoc />
+    public override int Read(byte[] buffer, int offset, int count)
+        => Read(buffer.AsSpan(offset, count));
+
+    /// <inheritdoc />
+    public override int Read(Span<byte> buffer)
+    {
+        int totalBytesRead = 0;
+        var stopwatch = Stopwatch.StartNew();
+
+        while (true)
+        {
+            totalBytesRead += _stream.Read(buffer);
+            if (StopReading(totalBytesRead, stopwatch.ElapsedMilliseconds))
+            {
+                break;
+            }
+
+            Thread.Sleep(50);
+        }
+
+        UpdateBytesWritten(totalBytesRead);
+
+        return totalBytesRead;
+    }
+
+    /// <inheritdoc />
+    public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        => await ReadAsync(buffer.AsMemory(offset, count), cancellationToken).ConfigureAwait(false);
+
+    /// <inheritdoc />
+    public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
+    {
+        int totalBytesRead = 0;
+        var stopwatch = Stopwatch.StartNew();
+
+        while (true)
+        {
+            totalBytesRead += await _stream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
+            if (StopReading(totalBytesRead, stopwatch.ElapsedMilliseconds))
+            {
+                break;
+            }
+
+            await Task.Delay(50, cancellationToken).ConfigureAwait(false);
+        }
+
+        UpdateBytesWritten(totalBytesRead);
+
+        return totalBytesRead;
+    }
+
+    /// <inheritdoc />
+    public override long Seek(long offset, SeekOrigin origin)
+        => throw new NotSupportedException();
+
+    /// <inheritdoc />
+    public override void SetLength(long value)
+        => throw new NotSupportedException();
+
+    /// <inheritdoc />
+    public override void Write(byte[] buffer, int offset, int count)
+        => throw new NotSupportedException();
+
+    /// <inheritdoc />
+    protected override void Dispose(bool disposing)
+    {
+        if (_disposed)
+        {
+            return;
+        }
+
+        try
+        {
+            if (disposing)
+            {
+                _stream.Dispose();
+
+                if (_job is not null)
+                {
+                    _transcodeManager?.OnTranscodeEndRequest(_job);
+                }
+            }
+        }
+        finally
+        {
+            _disposed = true;
+            base.Dispose(disposing);
+        }
+    }
+
+    private void UpdateBytesWritten(int totalBytesRead)
+    {
+        if (_job is not null)
+        {
+            _job.BytesDownloaded += totalBytesRead;
+        }
+    }
+
+    private bool StopReading(int bytesRead, long elapsed)
+    {
+        // It should stop reading when anything has been successfully read or if the job has exited
+        // If the job is null, however, it's a live stream and will require user action to close,
+        // but don't keep it open indefinitely if it isn't reading anything
+        return bytesRead > 0 || (_job?.HasExited ?? elapsed >= _timeoutMs);
+    }
+}
diff --git a/MediaBrowser.Controller/Streaming/StreamState.cs b/MediaBrowser.Controller/Streaming/StreamState.cs
new file mode 100644
index 0000000000..b5dbe29ec7
--- /dev/null
+++ b/MediaBrowser.Controller/Streaming/StreamState.cs
@@ -0,0 +1,183 @@
+using System;
+using MediaBrowser.Controller.Library;
+using MediaBrowser.Controller.MediaEncoding;
+using MediaBrowser.Model.Dlna;
+
+namespace MediaBrowser.Controller.Streaming;
+
+/// <summary>
+/// The stream state dto.
+/// </summary>
+public class StreamState : EncodingJobInfo, IDisposable
+{
+    private readonly IMediaSourceManager _mediaSourceManager;
+    private readonly ITranscodeManager _transcodeManager;
+    private bool _disposed;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="StreamState" /> class.
+    /// </summary>
+    /// <param name="mediaSourceManager">Instance of the <see cref="IMediaSourceManager" /> interface.</param>
+    /// <param name="transcodingType">The <see cref="TranscodingJobType" />.</param>
+    /// <param name="transcodeManager">The <see cref="ITranscodeManager" /> singleton.</param>
+    public StreamState(IMediaSourceManager mediaSourceManager, TranscodingJobType transcodingType, ITranscodeManager transcodeManager)
+        : base(transcodingType)
+    {
+        _mediaSourceManager = mediaSourceManager;
+        _transcodeManager = transcodeManager;
+    }
+
+    /// <summary>
+    /// Gets or sets the requested url.
+    /// </summary>
+    public string? RequestedUrl { get; set; }
+
+    /// <summary>
+    /// Gets or sets the request.
+    /// </summary>
+    public StreamingRequestDto Request
+    {
+        get => (StreamingRequestDto)BaseRequest;
+        set
+        {
+            BaseRequest = value;
+            IsVideoRequest = VideoRequest is not null;
+        }
+    }
+
+    /// <summary>
+    /// Gets the video request.
+    /// </summary>
+    public VideoRequestDto? VideoRequest => Request as VideoRequestDto;
+
+    /// <summary>
+    /// Gets or sets the direct stream provicer.
+    /// </summary>
+    /// <remarks>
+    /// Deprecated.
+    /// </remarks>
+    public IDirectStreamProvider? DirectStreamProvider { get; set; }
+
+    /// <summary>
+    /// Gets or sets the path to wait for.
+    /// </summary>
+    public string? WaitForPath { get; set; }
+
+    /// <summary>
+    /// Gets a value indicating whether the request outputs video.
+    /// </summary>
+    public bool IsOutputVideo => Request is VideoRequestDto;
+
+    /// <summary>
+    /// Gets the segment length.
+    /// </summary>
+    public int SegmentLength
+    {
+        get
+        {
+            if (Request.SegmentLength.HasValue)
+            {
+                return Request.SegmentLength.Value;
+            }
+
+            if (EncodingHelper.IsCopyCodec(OutputVideoCodec))
+            {
+                var userAgent = UserAgent ?? string.Empty;
+
+                if (userAgent.Contains("AppleTV", StringComparison.OrdinalIgnoreCase)
+                    || userAgent.Contains("cfnetwork", StringComparison.OrdinalIgnoreCase)
+                    || userAgent.Contains("ipad", StringComparison.OrdinalIgnoreCase)
+                    || userAgent.Contains("iphone", StringComparison.OrdinalIgnoreCase)
+                    || userAgent.Contains("ipod", StringComparison.OrdinalIgnoreCase))
+                {
+                    return 6;
+                }
+
+                if (IsSegmentedLiveStream)
+                {
+                    return 3;
+                }
+
+                return 6;
+            }
+
+            return 3;
+        }
+    }
+
+    /// <summary>
+    /// Gets the minimum number of segments.
+    /// </summary>
+    public int MinSegments
+    {
+        get
+        {
+            if (Request.MinSegments.HasValue)
+            {
+                return Request.MinSegments.Value;
+            }
+
+            return SegmentLength >= 10 ? 2 : 3;
+        }
+    }
+
+    /// <summary>
+    /// Gets or sets the user agent.
+    /// </summary>
+    public string? UserAgent { get; set; }
+
+    /// <summary>
+    /// Gets or sets a value indicating whether to estimate the content length.
+    /// </summary>
+    public bool EstimateContentLength { get; set; }
+
+    /// <summary>
+    /// Gets or sets the transcode seek info.
+    /// </summary>
+    public TranscodeSeekInfo TranscodeSeekInfo { get; set; }
+
+    /// <summary>
+    /// Gets or sets the transcoding job.
+    /// </summary>
+    public TranscodingJob? TranscodingJob { get; set; }
+
+    /// <inheritdoc />
+    public void Dispose()
+    {
+        Dispose(true);
+        GC.SuppressFinalize(this);
+    }
+
+    /// <inheritdoc />
+    public override void ReportTranscodingProgress(TimeSpan? transcodingPosition, float? framerate, double? percentComplete, long? bytesTranscoded, int? bitRate)
+    {
+        _transcodeManager.ReportTranscodingProgress(TranscodingJob!, this, transcodingPosition, framerate, percentComplete, bytesTranscoded, bitRate);
+    }
+
+    /// <summary>
+    /// Disposes the stream state.
+    /// </summary>
+    /// <param name="disposing">Whether the object is currently being disposed.</param>
+    protected virtual void Dispose(bool disposing)
+    {
+        if (_disposed)
+        {
+            return;
+        }
+
+        if (disposing)
+        {
+            // REVIEW: Is this the right place for this?
+            if (MediaSource.RequiresClosing
+                && string.IsNullOrWhiteSpace(Request.LiveStreamId)
+                && !string.IsNullOrWhiteSpace(MediaSource.LiveStreamId))
+            {
+                _mediaSourceManager.CloseLiveStream(MediaSource.LiveStreamId).GetAwaiter().GetResult();
+            }
+        }
+
+        TranscodingJob = null;
+
+        _disposed = true;
+    }
+}
diff --git a/MediaBrowser.Controller/Streaming/StreamingRequestDto.cs b/MediaBrowser.Controller/Streaming/StreamingRequestDto.cs
new file mode 100644
index 0000000000..e47ef65f06
--- /dev/null
+++ b/MediaBrowser.Controller/Streaming/StreamingRequestDto.cs
@@ -0,0 +1,49 @@
+using MediaBrowser.Controller.MediaEncoding;
+
+namespace MediaBrowser.Controller.Streaming;
+
+/// <summary>
+/// The audio streaming request dto.
+/// </summary>
+public class StreamingRequestDto : BaseEncodingJobOptions
+{
+    /// <summary>
+    /// Gets or sets the params.
+    /// </summary>
+    public string? Params { get; set; }
+
+    /// <summary>
+    /// Gets or sets the play session id.
+    /// </summary>
+    public string? PlaySessionId { get; set; }
+
+    /// <summary>
+    /// Gets or sets the tag.
+    /// </summary>
+    public string? Tag { get; set; }
+
+    /// <summary>
+    /// Gets or sets the segment container.
+    /// </summary>
+    public string? SegmentContainer { get; set; }
+
+    /// <summary>
+    /// Gets or sets the segment length.
+    /// </summary>
+    public int? SegmentLength { get; set; }
+
+    /// <summary>
+    /// Gets or sets the min segments.
+    /// </summary>
+    public int? MinSegments { get; set; }
+
+    /// <summary>
+    /// Gets or sets the position of the requested segment in ticks.
+    /// </summary>
+    public long CurrentRuntimeTicks { get; set; }
+
+    /// <summary>
+    /// Gets or sets the actual segment length in ticks.
+    /// </summary>
+    public long ActualSegmentLengthTicks { get; set; }
+}
diff --git a/MediaBrowser.Controller/Streaming/VideoRequestDto.cs b/MediaBrowser.Controller/Streaming/VideoRequestDto.cs
new file mode 100644
index 0000000000..44dc831fdc
--- /dev/null
+++ b/MediaBrowser.Controller/Streaming/VideoRequestDto.cs
@@ -0,0 +1,23 @@
+namespace MediaBrowser.Controller.Streaming;
+
+/// <summary>
+/// The video request dto.
+/// </summary>
+public class VideoRequestDto : StreamingRequestDto
+{
+    /// <summary>
+    /// Gets a value indicating whether this instance has fixed resolution.
+    /// </summary>
+    /// <value><c>true</c> if this instance has fixed resolution; otherwise, <c>false</c>.</value>
+    public bool HasFixedResolution => Width.HasValue || Height.HasValue;
+
+    /// <summary>
+    /// Gets or sets a value indicating whether to enable subtitles in the manifest.
+    /// </summary>
+    public bool EnableSubtitlesInManifest { get; set; }
+
+    /// <summary>
+    /// Gets or sets a value indicating whether to enable trickplay images.
+    /// </summary>
+    public bool EnableTrickplay { get; set; }
+}