using System;
using System.Diagnostics;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using MediaBrowser.Controller.MediaEncoding;
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 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);
}
}