Merge remote-tracking branch 'origin/master' into HEAD

1 files changed, 70 insertions, 55 deletions

diff --git a/Jellyfin.Api/Helpers/ProgressiveFileStream.cs b/Jellyfin.Api/Helpers/ProgressiveFileStream.cs
index 824870c7e..3fa07720a 100644
--- a/Jellyfin.Api/Helpers/ProgressiveFileStream.cs
+++ b/Jellyfin.Api/Helpers/ProgressiveFileStream.cs
@@ -1,6 +1,6 @@
 using System;
+using System.Diagnostics;
 using System.IO;
-using System.Runtime.InteropServices;
 using System.Threading;
 using System.Threading.Tasks;
 using Jellyfin.Api.Models.PlaybackDtos;
@@ -13,11 +13,10 @@ namespace Jellyfin.Api.Helpers
     /// </summary>
     public class ProgressiveFileStream : Stream
     {
-        private readonly FileStream _fileStream;
+        private readonly Stream _stream;
         private readonly TranscodingJobDto? _job;
-        private readonly TranscodingJobHelper _transcodingJobHelper;
-        private readonly bool _allowAsyncFileRead;
-        private int _bytesWritten;
+        private readonly TranscodingJobHelper? _transcodingJobHelper;
+        private readonly int _timeoutMs;
         private bool _disposed;
 
         /// <summary>
@@ -26,27 +25,31 @@ namespace Jellyfin.Api.Helpers
         /// <param name="filePath">The path to the transcoded file.</param>
         /// <param name="job">The transcoding job information.</param>
         /// <param name="transcodingJobHelper">The transcoding job helper.</param>
-        public ProgressiveFileStream(string filePath, TranscodingJobDto? job, TranscodingJobHelper transcodingJobHelper)
+        /// <param name="timeoutMs">The timeout duration in milliseconds.</param>
+        public ProgressiveFileStream(string filePath, TranscodingJobDto? job, TranscodingJobHelper transcodingJobHelper, int timeoutMs = 30000)
         {
             _job = job;
             _transcodingJobHelper = transcodingJobHelper;
-            _bytesWritten = 0;
+            _timeoutMs = timeoutMs;
 
-            var fileOptions = FileOptions.SequentialScan;
-            _allowAsyncFileRead = false;
-
-            // use non-async filestream along with read due to https://github.com/dotnet/corefx/issues/6039
-            if (!RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
-            {
-                fileOptions |= FileOptions.Asynchronous;
-                _allowAsyncFileRead = true;
-            }
+            _stream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite, IODefaults.FileStreamBufferSize, FileOptions.Asynchronous | FileOptions.SequentialScan);
+        }
 
-            _fileStream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite, IODefaults.FileStreamBufferSize, fileOptions);
+        /// <summary>
+        /// Initializes a new instance of the <see cref="ProgressiveFileStream"/> class.
+        /// </summary>
+        /// <param name="stream">The stream to progressively copy.</param>
+        /// <param name="timeoutMs">The timeout duration in milliseconds.</param>
+        public ProgressiveFileStream(Stream stream, int timeoutMs = 30000)
+        {
+            _job = null;
+            _transcodingJobHelper = null;
+            _timeoutMs = timeoutMs;
+            _stream = stream;
         }
 
         /// <inheritdoc />
-        public override bool CanRead => _fileStream.CanRead;
+        public override bool CanRead => _stream.CanRead;
 
         /// <inheritdoc />
         public override bool CanSeek => false;
@@ -67,60 +70,58 @@ namespace Jellyfin.Api.Helpers
         /// <inheritdoc />
         public override void Flush()
         {
-            _fileStream.Flush();
+            // Not supported
         }
 
         /// <inheritdoc />
         public override int Read(byte[] buffer, int offset, int count)
-        {
-            return _fileStream.Read(buffer, offset, count);
-        }
+            => Read(buffer.AsSpan(offset, count));
 
         /// <inheritdoc />
-        public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        public override int Read(Span<byte> buffer)
         {
             int totalBytesRead = 0;
-            int remainingBytesToRead = count;
+            var stopwatch = Stopwatch.StartNew();
 
-            int newOffset = offset;
-            while (remainingBytesToRead > 0)
+            while (KeepReading(stopwatch.ElapsedMilliseconds))
             {
-                cancellationToken.ThrowIfCancellationRequested();
-                int bytesRead;
-                if (_allowAsyncFileRead)
+                totalBytesRead += _stream.Read(buffer);
+                if (totalBytesRead > 0)
                 {
-                    bytesRead = await _fileStream.ReadAsync(buffer, newOffset, remainingBytesToRead, cancellationToken).ConfigureAwait(false);
-                }
-                else
-                {
-                    bytesRead = _fileStream.Read(buffer, newOffset, remainingBytesToRead);
+                    break;
                 }
 
-                remainingBytesToRead -= bytesRead;
-                newOffset += bytesRead;
+                Thread.Sleep(50);
+            }
 
-                if (bytesRead > 0)
-                {
-                    _bytesWritten += bytesRead;
-                    totalBytesRead += bytesRead;
+            UpdateBytesWritten(totalBytesRead);
 
-                    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
-                    if (_job?.HasExited ?? false)
-                    {
-                        break;
-                    }
+            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);
 
-                    await Task.Delay(50, cancellationToken).ConfigureAwait(false);
+        /// <inheritdoc />
+        public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
+        {
+            int totalBytesRead = 0;
+            var stopwatch = Stopwatch.StartNew();
+
+            while (KeepReading(stopwatch.ElapsedMilliseconds))
+            {
+                totalBytesRead += await _stream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
+                if (totalBytesRead > 0)
+                {
+                    break;
                 }
+
+                await Task.Delay(50, cancellationToken).ConfigureAwait(false);
             }
 
+            UpdateBytesWritten(totalBytesRead);
+
             return totalBytesRead;
         }
 
@@ -148,11 +149,11 @@ namespace Jellyfin.Api.Helpers
             {
                 if (disposing)
                 {
-                    _fileStream.Dispose();
+                    _stream.Dispose();
 
                     if (_job != null)
                     {
-                        _transcodingJobHelper.OnTranscodeEndRequest(_job);
+                        _transcodingJobHelper?.OnTranscodeEndRequest(_job);
                     }
                 }
             }
@@ -162,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;
+        }
     }
 }