Use the Memory overload for ReadAsync (#6865)

3 files changed, 49 insertions, 31 deletions

diff --git a/Jellyfin.Api/Helpers/ProgressiveFileStream.cs b/Jellyfin.Api/Helpers/ProgressiveFileStream.cs
index 61e18220a..3fa07720a 100644
--- a/Jellyfin.Api/Helpers/ProgressiveFileStream.cs
+++ b/Jellyfin.Api/Helpers/ProgressiveFileStream.cs
@@ -17,7 +17,6 @@ namespace Jellyfin.Api.Helpers
         private readonly TranscodingJobDto? _job;
         private readonly TranscodingJobHelper? _transcodingJobHelper;
         private readonly int _timeoutMs;
-        private int _bytesWritten;
         private bool _disposed;
 
         /// <summary>
@@ -71,53 +70,58 @@ namespace Jellyfin.Api.Helpers
         /// <inheritdoc />
         public override void Flush()
         {
-            _stream.Flush();
+            // Not supported
         }
 
         /// <inheritdoc />
         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 />
         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 remainingBytesToRead = count;
             var stopwatch = Stopwatch.StartNew();
 
-            int newOffset = offset;
-            while (remainingBytesToRead > 0)
+            while (KeepReading(stopwatch.ElapsedMilliseconds))
             {
-                cancellationToken.ThrowIfCancellationRequested();
-                int bytesRead = await _stream.ReadAsync(buffer, newOffset, remainingBytesToRead, cancellationToken).ConfigureAwait(false);
-
-                remainingBytesToRead -= bytesRead;
-                newOffset += bytesRead;
-
-                if (bytesRead > 0)
+                totalBytesRead += await _stream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
+                if (totalBytesRead > 0)
                 {
-                    _bytesWritten += bytesRead;
-                    totalBytesRead += bytesRead;
-
-                    if (_job != null)
-                    {
-                        _job.BytesDownloaded = Math.Max(_job.BytesDownloaded ?? _bytesWritten, _bytesWritten);
-                    }
+                    break;
                 }
-                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;
         }
 
@@ -159,5 +163,19 @@ namespace Jellyfin.Api.Helpers
                 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;
+        }
     }
 }
diff --git a/Jellyfin.Api/Models/PlaybackDtos/TranscodingJobDto.cs b/Jellyfin.Api/Models/PlaybackDtos/TranscodingJobDto.cs
index fed837b85..ab67c8732 100644
--- a/Jellyfin.Api/Models/PlaybackDtos/TranscodingJobDto.cs
+++ b/Jellyfin.Api/Models/PlaybackDtos/TranscodingJobDto.cs
@@ -134,7 +134,7 @@ namespace Jellyfin.Api.Models.PlaybackDtos
         /// <summary>
         /// Gets or sets bytes downloaded.
         /// </summary>
-        public long? BytesDownloaded { get; set; }
+        public long BytesDownloaded { get; set; }
 
         /// <summary>
         /// Gets or sets bytes transcoded.
diff --git a/Jellyfin.Api/Models/PlaybackDtos/TranscodingThrottler.cs b/Jellyfin.Api/Models/PlaybackDtos/TranscodingThrottler.cs
index 0136d9f86..7a1ca252c 100644
--- a/Jellyfin.Api/Models/PlaybackDtos/TranscodingThrottler.cs
+++ b/Jellyfin.Api/Models/PlaybackDtos/TranscodingThrottler.cs
@@ -141,7 +141,7 @@ namespace Jellyfin.Api.Models.PlaybackDtos
 
         private bool IsThrottleAllowed(TranscodingJobDto job, int thresholdSeconds)
         {
-            var bytesDownloaded = job.BytesDownloaded ?? 0;
+            var bytesDownloaded = job.BytesDownloaded;
             var transcodingPositionTicks = job.TranscodingPositionTicks ?? 0;
             var downloadPositionTicks = job.DownloadPositionTicks ?? 0;