Use the Memory overload for ReadAsync (#6865)
This commit is contained in:
parent
3eec137100
commit
ca887518dd
|
@ -17,7 +17,6 @@ namespace Jellyfin.Api.Helpers
|
||||||
private readonly TranscodingJobDto? _job;
|
private readonly TranscodingJobDto? _job;
|
||||||
private readonly TranscodingJobHelper? _transcodingJobHelper;
|
private readonly TranscodingJobHelper? _transcodingJobHelper;
|
||||||
private readonly int _timeoutMs;
|
private readonly int _timeoutMs;
|
||||||
private int _bytesWritten;
|
|
||||||
private bool _disposed;
|
private bool _disposed;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
|
@ -71,53 +70,58 @@ namespace Jellyfin.Api.Helpers
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
public override void Flush()
|
public override void Flush()
|
||||||
{
|
{
|
||||||
_stream.Flush();
|
// Not supported
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
public override int Read(byte[] buffer, int offset, int count)
|
public override int Read(byte[] buffer, int offset, int count)
|
||||||
|
=> Read(buffer.AsSpan(offset, count));
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public override int Read(Span<byte> buffer)
|
||||||
{
|
{
|
||||||
return _stream.Read(buffer, offset, count);
|
int totalBytesRead = 0;
|
||||||
|
var stopwatch = Stopwatch.StartNew();
|
||||||
|
|
||||||
|
while (KeepReading(stopwatch.ElapsedMilliseconds))
|
||||||
|
{
|
||||||
|
totalBytesRead += _stream.Read(buffer);
|
||||||
|
if (totalBytesRead > 0)
|
||||||
|
{
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
Thread.Sleep(50);
|
||||||
|
}
|
||||||
|
|
||||||
|
UpdateBytesWritten(totalBytesRead);
|
||||||
|
|
||||||
|
return totalBytesRead;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
|
public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
|
||||||
|
=> await ReadAsync(buffer.AsMemory(offset, count), cancellationToken).ConfigureAwait(false);
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
|
||||||
{
|
{
|
||||||
int totalBytesRead = 0;
|
int totalBytesRead = 0;
|
||||||
int remainingBytesToRead = count;
|
|
||||||
var stopwatch = Stopwatch.StartNew();
|
var stopwatch = Stopwatch.StartNew();
|
||||||
|
|
||||||
int newOffset = offset;
|
while (KeepReading(stopwatch.ElapsedMilliseconds))
|
||||||
while (remainingBytesToRead > 0)
|
|
||||||
{
|
{
|
||||||
cancellationToken.ThrowIfCancellationRequested();
|
totalBytesRead += await _stream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
|
||||||
int bytesRead = await _stream.ReadAsync(buffer, newOffset, remainingBytesToRead, cancellationToken).ConfigureAwait(false);
|
if (totalBytesRead > 0)
|
||||||
|
|
||||||
remainingBytesToRead -= bytesRead;
|
|
||||||
newOffset += bytesRead;
|
|
||||||
|
|
||||||
if (bytesRead > 0)
|
|
||||||
{
|
{
|
||||||
_bytesWritten += bytesRead;
|
break;
|
||||||
totalBytesRead += bytesRead;
|
|
||||||
|
|
||||||
if (_job != null)
|
|
||||||
{
|
|
||||||
_job.BytesDownloaded = Math.Max(_job.BytesDownloaded ?? _bytesWritten, _bytesWritten);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
else
|
|
||||||
{
|
|
||||||
// If the job is null it's a live stream and will require user action to close, but don't keep it open indefinitely
|
|
||||||
if (_job?.HasExited ?? stopwatch.ElapsedMilliseconds > _timeoutMs)
|
|
||||||
{
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
|
|
||||||
await Task.Delay(50, cancellationToken).ConfigureAwait(false);
|
await Task.Delay(50, cancellationToken).ConfigureAwait(false);
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
UpdateBytesWritten(totalBytesRead);
|
||||||
|
|
||||||
return totalBytesRead;
|
return totalBytesRead;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -159,5 +163,19 @@ namespace Jellyfin.Api.Helpers
|
||||||
base.Dispose(disposing);
|
base.Dispose(disposing);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private void UpdateBytesWritten(int totalBytesRead)
|
||||||
|
{
|
||||||
|
if (_job != null)
|
||||||
|
{
|
||||||
|
_job.BytesDownloaded += totalBytesRead;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private bool KeepReading(long elapsed)
|
||||||
|
{
|
||||||
|
// If the job is null it's a live stream and will require user action to close, but don't keep it open indefinitely
|
||||||
|
return !_job?.HasExited ?? elapsed < _timeoutMs;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -134,7 +134,7 @@ namespace Jellyfin.Api.Models.PlaybackDtos
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Gets or sets bytes downloaded.
|
/// Gets or sets bytes downloaded.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
public long? BytesDownloaded { get; set; }
|
public long BytesDownloaded { get; set; }
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Gets or sets bytes transcoded.
|
/// Gets or sets bytes transcoded.
|
||||||
|
|
|
@ -141,7 +141,7 @@ namespace Jellyfin.Api.Models.PlaybackDtos
|
||||||
|
|
||||||
private bool IsThrottleAllowed(TranscodingJobDto job, int thresholdSeconds)
|
private bool IsThrottleAllowed(TranscodingJobDto job, int thresholdSeconds)
|
||||||
{
|
{
|
||||||
var bytesDownloaded = job.BytesDownloaded ?? 0;
|
var bytesDownloaded = job.BytesDownloaded;
|
||||||
var transcodingPositionTicks = job.TranscodingPositionTicks ?? 0;
|
var transcodingPositionTicks = job.TranscodingPositionTicks ?? 0;
|
||||||
var downloadPositionTicks = job.DownloadPositionTicks ?? 0;
|
var downloadPositionTicks = job.DownloadPositionTicks ?? 0;
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue
Block a user