using System; using System.Diagnostics; using System.IO; using System.Threading; using System.Threading.Tasks; using Jellyfin.Api.Models.PlaybackDtos; using MediaBrowser.Model.IO; namespace Jellyfin.Api.Helpers { /// /// A progressive file stream for transferring transcoded files as they are written to. /// public class ProgressiveFileStream : Stream { private readonly Stream _stream; private readonly TranscodingJobDto? _job; private readonly TranscodingJobHelper? _transcodingJobHelper; 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 transcoding job helper. /// The timeout duration in milliseconds. public ProgressiveFileStream(string filePath, TranscodingJobDto? job, TranscodingJobHelper transcodingJobHelper, int timeoutMs = 30000) { _job = job; _transcodingJobHelper = transcodingJobHelper; _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; _transcodingJobHelper = 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 != null) { _transcodingJobHelper?.OnTranscodeEndRequest(_job); } } } finally { _disposed = true; base.Dispose(disposing); } } private void UpdateBytesWritten(int totalBytesRead) { if (_job != 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); } } }