jellyfin-server/Emby.Server.Implementations/IO/StreamHelper.cs

181 lines
6.1 KiB
C#
Raw Normal View History

2019-11-01 17:38:54 +00:00
#pragma warning disable CS1591
using System;
using System.Buffers;
using System.IO;
2018-09-12 17:26:21 +00:00
using System.Threading;
using System.Threading.Tasks;
using MediaBrowser.Model.IO;
namespace Emby.Server.Implementations.IO
{
public class StreamHelper : IStreamHelper
{
private const int StreamCopyToBufferSize = 81920;
2018-09-12 17:26:21 +00:00
public async Task CopyToAsync(Stream source, Stream destination, int bufferSize, Action onStarted, CancellationToken cancellationToken)
{
byte[] buffer = ArrayPool<byte>.Shared.Rent(bufferSize);
try
2018-09-12 17:26:21 +00:00
{
int read;
while ((read = await source.ReadAsync(buffer, 0, buffer.Length, cancellationToken).ConfigureAwait(false)) != 0)
{
cancellationToken.ThrowIfCancellationRequested();
2018-09-12 17:26:21 +00:00
await destination.WriteAsync(buffer, 0, read, cancellationToken).ConfigureAwait(false);
2018-09-12 17:26:21 +00:00
if (onStarted != null)
{
onStarted();
onStarted = null;
}
2018-09-12 17:26:21 +00:00
}
}
finally
{
ArrayPool<byte>.Shared.Return(buffer);
}
2018-09-12 17:26:21 +00:00
}
public async Task CopyToAsync(Stream source, Stream destination, int bufferSize, int emptyReadLimit, CancellationToken cancellationToken)
{
byte[] buffer = ArrayPool<byte>.Shared.Rent(bufferSize);
try
2018-09-12 17:26:21 +00:00
{
if (emptyReadLimit <= 0)
2018-09-12 17:26:21 +00:00
{
int read;
while ((read = await source.ReadAsync(buffer, 0, buffer.Length, cancellationToken).ConfigureAwait(false)) != 0)
{
cancellationToken.ThrowIfCancellationRequested();
2018-09-12 17:26:21 +00:00
await destination.WriteAsync(buffer, 0, read, cancellationToken).ConfigureAwait(false);
}
2018-09-12 17:26:21 +00:00
return;
}
2018-09-12 17:26:21 +00:00
var eofCount = 0;
2018-09-12 17:26:21 +00:00
while (eofCount < emptyReadLimit)
{
cancellationToken.ThrowIfCancellationRequested();
2018-09-12 17:26:21 +00:00
var bytesRead = await source.ReadAsync(buffer, 0, buffer.Length, cancellationToken).ConfigureAwait(false);
2018-09-12 17:26:21 +00:00
if (bytesRead == 0)
{
eofCount++;
await Task.Delay(50, cancellationToken).ConfigureAwait(false);
}
else
{
eofCount = 0;
2018-09-12 17:26:21 +00:00
await destination.WriteAsync(buffer, 0, bytesRead, cancellationToken).ConfigureAwait(false);
}
2018-09-12 17:26:21 +00:00
}
}
finally
{
ArrayPool<byte>.Shared.Return(buffer);
}
2018-09-12 17:26:21 +00:00
}
public async Task<int> CopyToAsync(Stream source, Stream destination, CancellationToken cancellationToken)
{
byte[] buffer = ArrayPool<byte>.Shared.Rent(StreamCopyToBufferSize);
try
2018-09-12 17:26:21 +00:00
{
int totalBytesRead = 0;
2018-09-12 17:26:21 +00:00
int bytesRead;
while ((bytesRead = await source.ReadAsync(buffer, 0, buffer.Length, cancellationToken).ConfigureAwait(false)) != 0)
2018-09-12 17:26:21 +00:00
{
var bytesToWrite = bytesRead;
2018-09-12 17:26:21 +00:00
if (bytesToWrite > 0)
{
await destination.WriteAsync(buffer, 0, Convert.ToInt32(bytesToWrite), cancellationToken).ConfigureAwait(false);
totalBytesRead += bytesRead;
}
2018-09-12 17:26:21 +00:00
}
return totalBytesRead;
}
finally
{
ArrayPool<byte>.Shared.Return(buffer);
}
2018-09-12 17:26:21 +00:00
}
public async Task CopyToAsync(Stream source, Stream destination, long copyLength, CancellationToken cancellationToken)
{
byte[] buffer = ArrayPool<byte>.Shared.Rent(StreamCopyToBufferSize);
try
2018-09-12 17:26:21 +00:00
{
int bytesRead;
2018-09-12 17:26:21 +00:00
while ((bytesRead = await source.ReadAsync(buffer, 0, buffer.Length, cancellationToken).ConfigureAwait(false)) != 0)
2018-09-12 17:26:21 +00:00
{
var bytesToWrite = Math.Min(bytesRead, copyLength);
2018-09-12 17:26:21 +00:00
if (bytesToWrite > 0)
{
await destination.WriteAsync(buffer, 0, Convert.ToInt32(bytesToWrite), cancellationToken).ConfigureAwait(false);
}
2018-09-12 17:26:21 +00:00
copyLength -= bytesToWrite;
if (copyLength <= 0)
{
break;
}
2018-09-12 17:26:21 +00:00
}
}
finally
{
ArrayPool<byte>.Shared.Return(buffer);
}
2018-09-12 17:26:21 +00:00
}
public async Task CopyUntilCancelled(Stream source, Stream target, int bufferSize, CancellationToken cancellationToken)
{
byte[] buffer = ArrayPool<byte>.Shared.Rent(bufferSize);
try
2018-09-12 17:26:21 +00:00
{
while (!cancellationToken.IsCancellationRequested)
2018-09-12 17:26:21 +00:00
{
var bytesRead = await CopyToAsyncInternal(source, target, buffer, cancellationToken).ConfigureAwait(false);
if (bytesRead == 0)
{
await Task.Delay(100, cancellationToken).ConfigureAwait(false);
}
2018-09-12 17:26:21 +00:00
}
}
finally
{
ArrayPool<byte>.Shared.Return(buffer);
}
2018-09-12 17:26:21 +00:00
}
private static async Task<int> CopyToAsyncInternal(Stream source, Stream destination, byte[] buffer, CancellationToken cancellationToken)
{
int bytesRead;
int totalBytesRead = 0;
while ((bytesRead = await source.ReadAsync(buffer, 0, buffer.Length, cancellationToken).ConfigureAwait(false)) != 0)
{
await destination.WriteAsync(buffer, 0, bytesRead, cancellationToken).ConfigureAwait(false);
2018-09-12 17:26:21 +00:00
totalBytesRead += bytesRead;
}
return totalBytesRead;
}
}
}