jellyfin-server/Jellyfin.Api/Helpers/ProgressiveFileCopier.cs

176 lines
6.6 KiB
C#
Raw Normal View History

2020-06-23 17:48:37 +00:00
using System;
2020-07-29 15:28:53 +00:00
using System.Buffers;
2020-06-23 17:48:37 +00:00
using System.IO;
2020-07-29 15:28:53 +00:00
using System.Runtime.InteropServices;
2020-06-23 17:48:37 +00:00
using System.Threading;
using System.Threading.Tasks;
2020-07-27 19:42:40 +00:00
using Jellyfin.Api.Models.PlaybackDtos;
2020-06-23 17:48:37 +00:00
using MediaBrowser.Controller.Library;
using MediaBrowser.Model.IO;
namespace Jellyfin.Api.Helpers
{
/// <summary>
/// Progressive file copier.
/// </summary>
public class ProgressiveFileCopier
{
2020-07-27 19:42:40 +00:00
private readonly TranscodingJobDto? _job;
2020-06-23 17:48:37 +00:00
private readonly string? _path;
2020-07-27 19:42:40 +00:00
private readonly CancellationToken _cancellationToken;
2020-06-23 17:48:37 +00:00
private readonly IDirectStreamProvider? _directStreamProvider;
2020-07-27 19:42:40 +00:00
private readonly TranscodingJobHelper _transcodingJobHelper;
private long _bytesWritten;
2020-06-23 17:48:37 +00:00
/// <summary>
/// Initializes a new instance of the <see cref="ProgressiveFileCopier"/> class.
/// </summary>
2020-07-27 19:42:40 +00:00
/// <param name="path">The path to copy from.</param>
/// <param name="job">The transcoding job.</param>
/// <param name="transcodingJobHelper">Instance of the <see cref="TranscodingJobHelper"/>.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public ProgressiveFileCopier(string path, TranscodingJobDto? job, TranscodingJobHelper transcodingJobHelper, CancellationToken cancellationToken)
2020-06-23 17:48:37 +00:00
{
_path = path;
2020-07-27 19:42:40 +00:00
_job = job;
_cancellationToken = cancellationToken;
_transcodingJobHelper = transcodingJobHelper;
2020-06-23 17:48:37 +00:00
}
/// <summary>
/// Initializes a new instance of the <see cref="ProgressiveFileCopier"/> class.
/// </summary>
/// <param name="directStreamProvider">Instance of the <see cref="IDirectStreamProvider"/> interface.</param>
2020-07-27 19:42:40 +00:00
/// <param name="job">The transcoding job.</param>
/// <param name="transcodingJobHelper">Instance of the <see cref="TranscodingJobHelper"/>.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public ProgressiveFileCopier(IDirectStreamProvider directStreamProvider, TranscodingJobDto? job, TranscodingJobHelper transcodingJobHelper, CancellationToken cancellationToken)
2020-06-23 17:48:37 +00:00
{
_directStreamProvider = directStreamProvider;
2020-07-27 19:42:40 +00:00
_job = job;
_cancellationToken = cancellationToken;
_transcodingJobHelper = transcodingJobHelper;
2020-06-23 17:48:37 +00:00
}
2020-07-27 19:42:40 +00:00
/// <summary>
/// Gets or sets a value indicating whether allow read end of file.
/// </summary>
public bool AllowEndOfFile { get; set; } = true;
/// <summary>
/// Gets or sets copy start position.
/// </summary>
public long StartPosition { get; set; }
2020-06-23 17:48:37 +00:00
/// <summary>
/// Write source stream to output.
/// </summary>
/// <param name="outputStream">Output stream.</param>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>A <see cref="Task"/>.</returns>
public async Task WriteToAsync(Stream outputStream, CancellationToken cancellationToken)
{
2020-07-27 19:42:40 +00:00
cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cancellationToken).Token;
try
2020-06-23 17:48:37 +00:00
{
2020-07-27 19:42:40 +00:00
if (_directStreamProvider != null)
{
await _directStreamProvider.CopyToAsync(outputStream, cancellationToken).ConfigureAwait(false);
return;
}
var fileOptions = FileOptions.SequentialScan;
var allowAsyncFileRead = false;
// use non-async filestream along with read due to https://github.com/dotnet/corefx/issues/6039
2020-07-29 20:21:32 +00:00
if (!RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
2020-07-27 19:42:40 +00:00
{
fileOptions |= FileOptions.Asynchronous;
allowAsyncFileRead = true;
}
await using var inputStream = new FileStream(_path, FileMode.Open, FileAccess.Read, FileShare.ReadWrite, IODefaults.FileStreamBufferSize, fileOptions);
var eofCount = 0;
const int EmptyReadLimit = 20;
2020-07-27 19:42:40 +00:00
if (StartPosition > 0)
{
inputStream.Position = StartPosition;
}
while (eofCount < EmptyReadLimit || !AllowEndOfFile)
2020-07-27 19:42:40 +00:00
{
var bytesRead = await CopyToInternalAsync(inputStream, outputStream, allowAsyncFileRead, cancellationToken).ConfigureAwait(false);
2020-06-23 17:48:37 +00:00
2020-07-27 19:42:40 +00:00
if (bytesRead == 0)
{
if (_job == null || _job.HasExited)
{
eofCount++;
}
2020-06-23 17:48:37 +00:00
2020-07-27 19:42:40 +00:00
await Task.Delay(100, cancellationToken).ConfigureAwait(false);
}
else
{
eofCount = 0;
}
}
}
finally
2020-06-23 17:48:37 +00:00
{
2020-07-27 19:42:40 +00:00
if (_job != null)
{
_transcodingJobHelper.OnTranscodeEndRequest(_job);
}
2020-06-23 17:48:37 +00:00
}
2020-07-27 19:42:40 +00:00
}
2020-06-23 17:48:37 +00:00
private async Task<int> CopyToInternalAsync(Stream source, Stream destination, bool readAsync, CancellationToken cancellationToken)
2020-07-27 19:42:40 +00:00
{
var array = ArrayPool<byte>.Shared.Rent(IODefaults.CopyToBufferSize);
2020-07-27 19:42:40 +00:00
int bytesRead;
int totalBytesRead = 0;
if (readAsync)
2020-06-23 17:48:37 +00:00
{
bytesRead = await source.ReadAsync(array, 0, array.Length, cancellationToken).ConfigureAwait(false);
}
else
{
bytesRead = source.Read(array, 0, array.Length);
2020-07-27 19:42:40 +00:00
}
while (bytesRead != 0)
2020-07-27 19:42:40 +00:00
{
var bytesToWrite = bytesRead;
if (bytesToWrite > 0)
2020-06-23 17:48:37 +00:00
{
2020-07-27 19:42:40 +00:00
await destination.WriteAsync(array, 0, Convert.ToInt32(bytesToWrite), cancellationToken).ConfigureAwait(false);
_bytesWritten += bytesRead;
totalBytesRead += bytesRead;
if (_job != null)
{
_job.BytesDownloaded = Math.Max(_job.BytesDownloaded ?? _bytesWritten, _bytesWritten);
}
2020-06-23 17:48:37 +00:00
}
if (readAsync)
{
bytesRead = await source.ReadAsync(array, 0, array.Length, cancellationToken).ConfigureAwait(false);
}
else
{
bytesRead = source.Read(array, 0, array.Length);
}
2020-06-23 17:48:37 +00:00
}
2020-07-27 19:42:40 +00:00
return totalBytesRead;
2020-06-23 17:48:37 +00:00
}
}
}