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; /// /// A progressive file stream for transferring transcoded files as they are written to. /// public class ProgressiveFileStream : Stream { private readonly Stream _stream; private readonly TranscodingJob? _job; private readonly ITranscodeManager? _transcodeManager; private readonly int _timeoutMs; private bool _disposed; /// /// Initializes a new instance of the class. /// /// The path to the transcoded file. /// The transcoding job information. /// The transcode manager. /// The timeout duration in milliseconds. 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); } /// /// Initializes a new instance of the class. /// /// The stream to progressively copy. /// The timeout duration in milliseconds. public ProgressiveFileStream(Stream stream, int timeoutMs = 30000) { _job = null; _transcodeManager = null; _timeoutMs = timeoutMs; _stream = stream; } /// public override bool CanRead => _stream.CanRead; /// public override bool CanSeek => false; /// public override bool CanWrite => false; /// public override long Length => throw new NotSupportedException(); /// public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } /// public override void Flush() { // Not supported } /// public override int Read(byte[] buffer, int offset, int count) => Read(buffer.AsSpan(offset, count)); /// public override int Read(Span 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; } /// public override async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => await ReadAsync(buffer.AsMemory(offset, count), cancellationToken).ConfigureAwait(false); /// public override async ValueTask ReadAsync(Memory 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; } /// 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(); /// 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); } }