Merge pull request #11045 from barronpm/livetv-recordingsmanager

LiveTV Recordings Refactor

5 files changed, 688 insertions, 0 deletions

diff --git a/src/Jellyfin.LiveTv/IO/DirectRecorder.cs b/src/Jellyfin.LiveTv/IO/DirectRecorder.cs
new file mode 100644
index 000000000..c4ec6de40
--- /dev/null
+++ b/src/Jellyfin.LiveTv/IO/DirectRecorder.cs
@@ -0,0 +1,118 @@
+#pragma warning disable CS1591
+
+using System;
+using System.IO;
+using System.Net.Http;
+using System.Threading;
+using System.Threading.Tasks;
+using MediaBrowser.Common.Net;
+using MediaBrowser.Controller.Library;
+using MediaBrowser.Controller.Streaming;
+using MediaBrowser.Model.Dto;
+using MediaBrowser.Model.IO;
+using Microsoft.Extensions.Logging;
+
+namespace Jellyfin.LiveTv.IO
+{
+    public sealed class DirectRecorder : IRecorder
+    {
+        private readonly ILogger _logger;
+        private readonly IHttpClientFactory _httpClientFactory;
+        private readonly IStreamHelper _streamHelper;
+
+        public DirectRecorder(ILogger logger, IHttpClientFactory httpClientFactory, IStreamHelper streamHelper)
+        {
+            _logger = logger;
+            _httpClientFactory = httpClientFactory;
+            _streamHelper = streamHelper;
+        }
+
+        public string GetOutputPath(MediaSourceInfo mediaSource, string targetFile)
+        {
+            return targetFile;
+        }
+
+        public Task Record(IDirectStreamProvider? directStreamProvider, MediaSourceInfo mediaSource, string targetFile, TimeSpan duration, Action onStarted, CancellationToken cancellationToken)
+        {
+            if (directStreamProvider is not null)
+            {
+                return RecordFromDirectStreamProvider(directStreamProvider, targetFile, duration, onStarted, cancellationToken);
+            }
+
+            return RecordFromMediaSource(mediaSource, targetFile, duration, onStarted, cancellationToken);
+        }
+
+        private async Task RecordFromDirectStreamProvider(IDirectStreamProvider directStreamProvider, string targetFile, TimeSpan duration, Action onStarted, CancellationToken cancellationToken)
+        {
+            Directory.CreateDirectory(Path.GetDirectoryName(targetFile) ?? throw new ArgumentException("Path can't be a root directory.", nameof(targetFile)));
+
+            var output = new FileStream(
+                targetFile,
+                FileMode.CreateNew,
+                FileAccess.Write,
+                FileShare.Read,
+                IODefaults.FileStreamBufferSize,
+                FileOptions.Asynchronous);
+
+            await using (output.ConfigureAwait(false))
+            {
+                onStarted();
+
+                _logger.LogInformation("Copying recording to file {FilePath}", targetFile);
+
+                // The media source is infinite so we need to handle stopping ourselves
+                using var durationToken = new CancellationTokenSource(duration);
+                using var cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, durationToken.Token);
+                var linkedCancellationToken = cancellationTokenSource.Token;
+                var fileStream = new ProgressiveFileStream(directStreamProvider.GetStream());
+                await using (fileStream.ConfigureAwait(false))
+                {
+                    await _streamHelper.CopyToAsync(
+                        fileStream,
+                        output,
+                        IODefaults.CopyToBufferSize,
+                        1000,
+                        linkedCancellationToken).ConfigureAwait(false);
+                }
+            }
+
+            _logger.LogInformation("Recording completed: {FilePath}", targetFile);
+        }
+
+        private async Task RecordFromMediaSource(MediaSourceInfo mediaSource, string targetFile, TimeSpan duration, Action onStarted, CancellationToken cancellationToken)
+        {
+            using var response = await _httpClientFactory.CreateClient(NamedClient.Default)
+                .GetAsync(mediaSource.Path, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false);
+
+            _logger.LogInformation("Opened recording stream from tuner provider");
+
+            Directory.CreateDirectory(Path.GetDirectoryName(targetFile) ?? throw new ArgumentException("Path can't be a root directory.", nameof(targetFile)));
+
+            var output = new FileStream(targetFile, FileMode.CreateNew, FileAccess.Write, FileShare.Read, IODefaults.CopyToBufferSize, FileOptions.Asynchronous);
+            await using (output.ConfigureAwait(false))
+            {
+                onStarted();
+
+                _logger.LogInformation("Copying recording stream to file {0}", targetFile);
+
+                // The media source if infinite so we need to handle stopping ourselves
+                using var durationToken = new CancellationTokenSource(duration);
+                using var linkedCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, durationToken.Token);
+                cancellationToken = linkedCancellationToken.Token;
+
+                await _streamHelper.CopyUntilCancelled(
+                    await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false),
+                    output,
+                    IODefaults.CopyToBufferSize,
+                    cancellationToken).ConfigureAwait(false);
+
+                _logger.LogInformation("Recording completed to file {0}", targetFile);
+            }
+        }
+
+        /// <inheritdoc />
+        public void Dispose()
+        {
+        }
+    }
+}
diff --git a/src/Jellyfin.LiveTv/IO/EncodedRecorder.cs b/src/Jellyfin.LiveTv/IO/EncodedRecorder.cs
new file mode 100644
index 000000000..ff00c8999
--- /dev/null
+++ b/src/Jellyfin.LiveTv/IO/EncodedRecorder.cs
@@ -0,0 +1,362 @@
+#nullable disable
+
+#pragma warning disable CS1591
+
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Globalization;
+using System.IO;
+using System.Text;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Jellyfin.Extensions;
+using Jellyfin.Extensions.Json;
+using MediaBrowser.Common;
+using MediaBrowser.Common.Configuration;
+using MediaBrowser.Controller;
+using MediaBrowser.Controller.Configuration;
+using MediaBrowser.Controller.Library;
+using MediaBrowser.Controller.MediaEncoding;
+using MediaBrowser.Model.Dto;
+using MediaBrowser.Model.IO;
+using Microsoft.Extensions.Logging;
+
+namespace Jellyfin.LiveTv.IO
+{
+    public class EncodedRecorder : IRecorder
+    {
+        private readonly ILogger _logger;
+        private readonly IMediaEncoder _mediaEncoder;
+        private readonly IServerApplicationPaths _appPaths;
+        private readonly TaskCompletionSource<bool> _taskCompletionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly IServerConfigurationManager _serverConfigurationManager;
+        private readonly JsonSerializerOptions _jsonOptions = JsonDefaults.Options;
+        private bool _hasExited;
+        private FileStream _logFileStream;
+        private string _targetPath;
+        private Process _process;
+        private bool _disposed;
+
+        public EncodedRecorder(
+            ILogger logger,
+            IMediaEncoder mediaEncoder,
+            IServerApplicationPaths appPaths,
+            IServerConfigurationManager serverConfigurationManager)
+        {
+            _logger = logger;
+            _mediaEncoder = mediaEncoder;
+            _appPaths = appPaths;
+            _serverConfigurationManager = serverConfigurationManager;
+        }
+
+        private static bool CopySubtitles => false;
+
+        public string GetOutputPath(MediaSourceInfo mediaSource, string targetFile)
+        {
+            return Path.ChangeExtension(targetFile, ".ts");
+        }
+
+        public async Task Record(IDirectStreamProvider directStreamProvider, MediaSourceInfo mediaSource, string targetFile, TimeSpan duration, Action onStarted, CancellationToken cancellationToken)
+        {
+            // The media source is infinite so we need to handle stopping ourselves
+            using var durationToken = new CancellationTokenSource(duration);
+            using var cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, durationToken.Token);
+
+            await RecordFromFile(mediaSource, mediaSource.Path, targetFile, onStarted, cancellationTokenSource.Token).ConfigureAwait(false);
+
+            _logger.LogInformation("Recording completed to file {Path}", targetFile);
+        }
+
+        private async Task RecordFromFile(MediaSourceInfo mediaSource, string inputFile, string targetFile, Action onStarted, CancellationToken cancellationToken)
+        {
+            _targetPath = targetFile;
+            Directory.CreateDirectory(Path.GetDirectoryName(targetFile));
+
+            var processStartInfo = new ProcessStartInfo
+            {
+                CreateNoWindow = true,
+                UseShellExecute = false,
+
+                RedirectStandardError = true,
+                RedirectStandardInput = true,
+
+                FileName = _mediaEncoder.EncoderPath,
+                Arguments = GetCommandLineArgs(mediaSource, inputFile, targetFile),
+
+                WindowStyle = ProcessWindowStyle.Hidden,
+                ErrorDialog = false
+            };
+
+            _logger.LogInformation("{Filename} {Arguments}", processStartInfo.FileName, processStartInfo.Arguments);
+
+            var logFilePath = Path.Combine(_appPaths.LogDirectoryPath, "record-transcode-" + Guid.NewGuid() + ".txt");
+            Directory.CreateDirectory(Path.GetDirectoryName(logFilePath));
+
+            // FFMpeg writes debug/error info to stderr. This is useful when debugging so let's put it in the log directory.
+            _logFileStream = new FileStream(logFilePath, FileMode.CreateNew, FileAccess.Write, FileShare.Read, IODefaults.FileStreamBufferSize, FileOptions.Asynchronous);
+
+            await JsonSerializer.SerializeAsync(_logFileStream, mediaSource, _jsonOptions, cancellationToken).ConfigureAwait(false);
+            await _logFileStream.WriteAsync(Encoding.UTF8.GetBytes(Environment.NewLine + Environment.NewLine + processStartInfo.FileName + " " + processStartInfo.Arguments + Environment.NewLine + Environment.NewLine), cancellationToken).ConfigureAwait(false);
+
+            _process = new Process
+            {
+                StartInfo = processStartInfo,
+                EnableRaisingEvents = true
+            };
+            _process.Exited += (_, _) => OnFfMpegProcessExited(_process);
+
+            _process.Start();
+
+            cancellationToken.Register(Stop);
+
+            onStarted();
+
+            // Important - don't await the log task or we won't be able to kill ffmpeg when the user stops playback
+            _ = StartStreamingLog(_process.StandardError.BaseStream, _logFileStream);
+
+            _logger.LogInformation("ffmpeg recording process started for {Path}", _targetPath);
+
+            // Block until ffmpeg exits
+            await _taskCompletionSource.Task.ConfigureAwait(false);
+        }
+
+        private string GetCommandLineArgs(MediaSourceInfo mediaSource, string inputTempFile, string targetFile)
+        {
+            string videoArgs;
+            if (EncodeVideo(mediaSource))
+            {
+                const int MaxBitrate = 25000000;
+                videoArgs = string.Format(
+                    CultureInfo.InvariantCulture,
+                    "-codec:v:0 libx264 -force_key_frames \"expr:gte(t,n_forced*5)\" {0} -pix_fmt yuv420p -preset superfast -crf 23 -b:v {1} -maxrate {1} -bufsize ({1}*2) -vsync -1 -profile:v high -level 41",
+                    GetOutputSizeParam(),
+                    MaxBitrate);
+            }
+            else
+            {
+                videoArgs = "-codec:v:0 copy";
+            }
+
+            videoArgs += " -fflags +genpts";
+
+            var flags = new List<string>();
+            if (mediaSource.IgnoreDts)
+            {
+                flags.Add("+igndts");
+            }
+
+            if (mediaSource.IgnoreIndex)
+            {
+                flags.Add("+ignidx");
+            }
+
+            if (mediaSource.GenPtsInput)
+            {
+                flags.Add("+genpts");
+            }
+
+            var inputModifier = "-async 1 -vsync -1";
+
+            if (flags.Count > 0)
+            {
+                inputModifier += " -fflags " + string.Join(string.Empty, flags);
+            }
+
+            if (mediaSource.ReadAtNativeFramerate)
+            {
+                inputModifier += " -re";
+            }
+
+            if (mediaSource.RequiresLooping)
+            {
+                inputModifier += " -stream_loop -1 -reconnect_at_eof 1 -reconnect_streamed 1 -reconnect_delay_max 2";
+            }
+
+            var analyzeDurationSeconds = 5;
+            var analyzeDuration = " -analyzeduration " +
+                  (analyzeDurationSeconds * 1000000).ToString(CultureInfo.InvariantCulture);
+            inputModifier += analyzeDuration;
+
+            var subtitleArgs = CopySubtitles ? " -codec:s copy" : " -sn";
+
+            // var outputParam = string.Equals(Path.GetExtension(targetFile), ".mp4", StringComparison.OrdinalIgnoreCase) ?
+            //    " -f mp4 -movflags frag_keyframe+empty_moov" :
+            //    string.Empty;
+
+            var outputParam = string.Empty;
+
+            var threads = EncodingHelper.GetNumberOfThreads(null, _serverConfigurationManager.GetEncodingOptions(), null);
+            var commandLineArgs = string.Format(
+                CultureInfo.InvariantCulture,
+                "-i \"{0}\" {2} -map_metadata -1 -threads {6} {3}{4}{5} -y \"{1}\"",
+                inputTempFile,
+                targetFile.Replace("\"", "\\\"", StringComparison.Ordinal), // Escape quotes in filename
+                videoArgs,
+                GetAudioArgs(mediaSource),
+                subtitleArgs,
+                outputParam,
+                threads);
+
+            return inputModifier + " " + commandLineArgs;
+        }
+
+        private static string GetAudioArgs(MediaSourceInfo mediaSource)
+        {
+            return "-codec:a:0 copy";
+
+            // var audioChannels = 2;
+            // var audioStream = mediaStreams.FirstOrDefault(i => i.Type == MediaStreamType.Audio);
+            // if (audioStream is not null)
+            // {
+            //    audioChannels = audioStream.Channels ?? audioChannels;
+            // }
+            // return "-codec:a:0 aac -strict experimental -ab 320000";
+        }
+
+        private static bool EncodeVideo(MediaSourceInfo mediaSource)
+        {
+            return false;
+        }
+
+        protected string GetOutputSizeParam()
+            => "-vf \"yadif=0:-1:0\"";
+
+        private void Stop()
+        {
+            if (!_hasExited)
+            {
+                try
+                {
+                    _logger.LogInformation("Stopping ffmpeg recording process for {Path}", _targetPath);
+
+                    _process.StandardInput.WriteLine("q");
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogError(ex, "Error stopping recording transcoding job for {Path}", _targetPath);
+                }
+
+                if (_hasExited)
+                {
+                    return;
+                }
+
+                try
+                {
+                    _logger.LogInformation("Calling recording process.WaitForExit for {Path}", _targetPath);
+
+                    if (_process.WaitForExit(10000))
+                    {
+                        return;
+                    }
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogError(ex, "Error waiting for recording process to exit for {Path}", _targetPath);
+                }
+
+                if (_hasExited)
+                {
+                    return;
+                }
+
+                try
+                {
+                    _logger.LogInformation("Killing ffmpeg recording process for {Path}", _targetPath);
+
+                    _process.Kill();
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogError(ex, "Error killing recording transcoding job for {Path}", _targetPath);
+                }
+            }
+        }
+
+        /// <summary>
+        /// Processes the exited.
+        /// </summary>
+        private void OnFfMpegProcessExited(Process process)
+        {
+            using (process)
+            {
+                _hasExited = true;
+
+                _logFileStream?.Dispose();
+                _logFileStream = null;
+
+                var exitCode = process.ExitCode;
+
+                _logger.LogInformation("FFMpeg recording exited with code {ExitCode} for {Path}", exitCode, _targetPath);
+
+                if (exitCode == 0)
+                {
+                    _taskCompletionSource.TrySetResult(true);
+                }
+                else
+                {
+                    _taskCompletionSource.TrySetException(
+                        new FfmpegException(
+                            string.Format(
+                                CultureInfo.InvariantCulture,
+                                "Recording for {0} failed. Exit code {1}",
+                                _targetPath,
+                                exitCode)));
+                }
+            }
+        }
+
+        private async Task StartStreamingLog(Stream source, FileStream target)
+        {
+            try
+            {
+                using (var reader = new StreamReader(source))
+                {
+                    await foreach (var line in reader.ReadAllLinesAsync().ConfigureAwait(false))
+                    {
+                        var bytes = Encoding.UTF8.GetBytes(Environment.NewLine + line);
+
+                        await target.WriteAsync(bytes.AsMemory()).ConfigureAwait(false);
+                        await target.FlushAsync().ConfigureAwait(false);
+                    }
+                }
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Error reading ffmpeg recording log");
+            }
+        }
+
+        /// <inheritdoc />
+        public void Dispose()
+        {
+            Dispose(true);
+            GC.SuppressFinalize(this);
+        }
+
+        /// <summary>
+        /// Releases unmanaged and optionally managed resources.
+        /// </summary>
+        /// <param name="disposing"><c>true</c> to release both managed and unmanaged resources; <c>false</c> to release only unmanaged resources.</param>
+        protected virtual void Dispose(bool disposing)
+        {
+            if (_disposed)
+            {
+                return;
+            }
+
+            if (disposing)
+            {
+                _logFileStream?.Dispose();
+                _process?.Dispose();
+            }
+
+            _logFileStream = null;
+            _process = null;
+
+            _disposed = true;
+        }
+    }
+}
diff --git a/src/Jellyfin.LiveTv/IO/ExclusiveLiveStream.cs b/src/Jellyfin.LiveTv/IO/ExclusiveLiveStream.cs
new file mode 100644
index 000000000..394b9cf11
--- /dev/null
+++ b/src/Jellyfin.LiveTv/IO/ExclusiveLiveStream.cs
@@ -0,0 +1,61 @@
+#nullable disable
+
+#pragma warning disable CA1711
+#pragma warning disable CS1591
+
+using System;
+using System.Globalization;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using MediaBrowser.Controller.Library;
+using MediaBrowser.Model.Dto;
+
+namespace Jellyfin.LiveTv.IO
+{
+    public sealed class ExclusiveLiveStream : ILiveStream
+    {
+        private readonly Func<Task> _closeFn;
+
+        public ExclusiveLiveStream(MediaSourceInfo mediaSource, Func<Task> closeFn)
+        {
+            MediaSource = mediaSource;
+            EnableStreamSharing = false;
+            _closeFn = closeFn;
+            ConsumerCount = 1;
+            UniqueId = Guid.NewGuid().ToString("N", CultureInfo.InvariantCulture);
+        }
+
+        public int ConsumerCount { get; set; }
+
+        public string OriginalStreamId { get; set; }
+
+        public string TunerHostId => null;
+
+        public bool EnableStreamSharing { get; set; }
+
+        public MediaSourceInfo MediaSource { get; set; }
+
+        public string UniqueId { get; }
+
+        public Task Close()
+        {
+            return _closeFn();
+        }
+
+        public Stream GetStream()
+        {
+            throw new NotSupportedException();
+        }
+
+        public Task Open(CancellationToken openCancellationToken)
+        {
+            return Task.CompletedTask;
+        }
+
+        /// <inheritdoc />
+        public void Dispose()
+        {
+        }
+    }
+}
diff --git a/src/Jellyfin.LiveTv/IO/IRecorder.cs b/src/Jellyfin.LiveTv/IO/IRecorder.cs
new file mode 100644
index 000000000..ab4506414
--- /dev/null
+++ b/src/Jellyfin.LiveTv/IO/IRecorder.cs
@@ -0,0 +1,27 @@
+#pragma warning disable CS1591
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using MediaBrowser.Controller.Library;
+using MediaBrowser.Model.Dto;
+
+namespace Jellyfin.LiveTv.IO
+{
+    public interface IRecorder : IDisposable
+    {
+        /// <summary>
+        /// Records the specified media source.
+        /// </summary>
+        /// <param name="directStreamProvider">The direct stream provider, or <c>null</c>.</param>
+        /// <param name="mediaSource">The media source.</param>
+        /// <param name="targetFile">The target file.</param>
+        /// <param name="duration">The duration to record.</param>
+        /// <param name="onStarted">An action to perform when recording starts.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>A <see cref="Task"/> that represents the recording operation.</returns>
+        Task Record(IDirectStreamProvider? directStreamProvider, MediaSourceInfo mediaSource, string targetFile, TimeSpan duration, Action onStarted, CancellationToken cancellationToken);
+
+        string GetOutputPath(MediaSourceInfo mediaSource, string targetFile);
+    }
+}
diff --git a/src/Jellyfin.LiveTv/IO/StreamHelper.cs b/src/Jellyfin.LiveTv/IO/StreamHelper.cs
new file mode 100644
index 000000000..7947807ba
--- /dev/null
+++ b/src/Jellyfin.LiveTv/IO/StreamHelper.cs
@@ -0,0 +1,120 @@
+#pragma warning disable CS1591
+
+using System;
+using System.Buffers;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using MediaBrowser.Model.IO;
+
+namespace Jellyfin.LiveTv.IO
+{
+    public class StreamHelper : IStreamHelper
+    {
+        public async Task CopyToAsync(Stream source, Stream destination, int bufferSize, Action? onStarted, CancellationToken cancellationToken)
+        {
+            byte[] buffer = ArrayPool<byte>.Shared.Rent(bufferSize);
+            try
+            {
+                int read;
+                while ((read = await source.ReadAsync(buffer, cancellationToken).ConfigureAwait(false)) != 0)
+                {
+                    cancellationToken.ThrowIfCancellationRequested();
+
+                    await destination.WriteAsync(buffer.AsMemory(0, read), cancellationToken).ConfigureAwait(false);
+
+                    if (onStarted is not null)
+                    {
+                        onStarted();
+                        onStarted = null;
+                    }
+                }
+            }
+            finally
+            {
+                ArrayPool<byte>.Shared.Return(buffer);
+            }
+        }
+
+        public async Task CopyToAsync(Stream source, Stream destination, int bufferSize, int emptyReadLimit, CancellationToken cancellationToken)
+        {
+            byte[] buffer = ArrayPool<byte>.Shared.Rent(bufferSize);
+            try
+            {
+                if (emptyReadLimit <= 0)
+                {
+                    int read;
+                    while ((read = await source.ReadAsync(buffer, cancellationToken).ConfigureAwait(false)) != 0)
+                    {
+                        cancellationToken.ThrowIfCancellationRequested();
+
+                        await destination.WriteAsync(buffer.AsMemory(0, read), cancellationToken).ConfigureAwait(false);
+                    }
+
+                    return;
+                }
+
+                var eofCount = 0;
+
+                while (eofCount < emptyReadLimit)
+                {
+                    cancellationToken.ThrowIfCancellationRequested();
+
+                    var bytesRead = await source.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
+
+                    if (bytesRead == 0)
+                    {
+                        eofCount++;
+                        await Task.Delay(50, cancellationToken).ConfigureAwait(false);
+                    }
+                    else
+                    {
+                        eofCount = 0;
+
+                        await destination.WriteAsync(buffer.AsMemory(0, bytesRead), cancellationToken).ConfigureAwait(false);
+                    }
+                }
+            }
+            finally
+            {
+                ArrayPool<byte>.Shared.Return(buffer);
+            }
+        }
+
+        public async Task CopyUntilCancelled(Stream source, Stream target, int bufferSize, CancellationToken cancellationToken)
+        {
+            byte[] buffer = ArrayPool<byte>.Shared.Rent(bufferSize);
+            try
+            {
+                while (!cancellationToken.IsCancellationRequested)
+                {
+                    var bytesRead = await CopyToAsyncInternal(source, target, buffer, cancellationToken).ConfigureAwait(false);
+
+                    if (bytesRead == 0)
+                    {
+                        await Task.Delay(100, cancellationToken).ConfigureAwait(false);
+                    }
+                }
+            }
+            finally
+            {
+                ArrayPool<byte>.Shared.Return(buffer);
+            }
+        }
+
+        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, cancellationToken).ConfigureAwait(false)) != 0)
+            {
+                await destination.WriteAsync(buffer.AsMemory(0, bytesRead), cancellationToken).ConfigureAwait(false);
+
+                totalBytesRead += bytesRead;
+            }
+
+            return totalBytesRead;
+        }
+    }
+}