Add download speed limiter on the Simple downloader.

CHANGELOG.md

@@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.9.8] - 2021-10-30
+### Added
+- Add speed limit setting on the Simple downloader.
+
 ## [1.9.7] - 2021-10-30
 ### Added
 - Add AllDebrid support.

client/src/app/navbar/navbar.component.html

@@ -55,7 +55,7 @@
           <a class="navbar-item"> Profile </a>
           <a class="navbar-item" (click)="logout()"> Logout </a>
           <hr class="navbar-divider" />
-          <div class="navbar-item">Version 1.9.7</div>
+          <div class="navbar-item">Version 1.9.8</div>
         </div>
       </div>
     </div>

client/src/app/settings/settings.component.html

@@ -147,7 +147,7 @@
     </p>
   </div>
 
-  <div class="field" *ngIf="settingDownloadClient === 'MultiPart'">
+  <div class="field" *ngIf="settingDownloadClient === 'MultiPart' || settingDownloadClient === 'Simple'">
     <label class="label">Download speed (in MB/s)</label>
     <div class="control">
       <input class="input" type="number" max="1000" min="0" step="1" [(ngModel)]="settingDownloadMaxSpeed" />

package.json

@@ -1,6 +1,6 @@
 {
   "name": "rdt-client",
-  "version": "1.9.7",
+  "version": "1.9.8",
   "description": "This is a web interface to manage your torrents on Real-Debrid.",
   "main": "index.js",
   "dependencies": {

server/RdtClient.Service/Services/Downloaders/SimpleDownloader.cs

@@ -1,6 +1,7 @@
 using System;
 using System.IO;
 using System.Net;
+using System.Threading;
 using System.Threading.Tasks;
 using Serilog;
 
@@ -8,6 +9,8 @@ namespace RdtClient.Service.Services.Downloaders
 {
     public class SimpleDownloader : IDownloader
     {
+        private const Int32 BufferSize = 8 * 1024;
+
         public event EventHandler<DownloadCompleteEventArgs> DownloadComplete;
         public event EventHandler<DownloadProgressEventArgs> DownloadProgress;
 
@@ -81,47 +84,51 @@ namespace RdtClient.Service.Services.Downloaders
                         var request = WebRequest.Create(_uri);
                         using var response = await request.GetResponseAsync();
 
-                        await using var stream = response.GetResponseStream();
+                        await using var responseStream = response.GetResponseStream();
 
-                        if (stream == null)
+                        if (responseStream == null)
                         {
                             throw new IOException("No stream");
                         }
 
-                        await using var fileStream = new FileStream(_filePath, FileMode.Create, FileAccess.Write, FileShare.Write);
+                        var speedLimit = Settings.Get.DownloadMaxSpeed * BufferSize * 1024L;
-                        var buffer = new Byte[64 * 1024];
+                        
-
+                        if (speedLimit == 0)
-                        while (fileStream.Length < response.ContentLength && !_cancelled)
                         {
-                            var read = await stream.ReadAsync(buffer.AsMemory(0, buffer.Length));
+                            speedLimit = ThrottledStream.Infinite;
+                        }
 
-                            if (read > 0)
+                        await using var destinationStream = new ThrottledStream(responseStream, speedLimit);
+
+                        await using var fileStream = new FileStream(_filePath, FileMode.Create, FileAccess.Write, FileShare.Write);
+                        
+                        var readSize = 1;
+                        while (readSize > 0 && !_cancelled)
+                        {
+                            using var innerCts = new CancellationTokenSource(1000);
+                            var buffer = new Byte[BufferSize * 8];
+                            readSize = await destinationStream.ReadAsync(buffer, 0, buffer.Length, innerCts.Token).ConfigureAwait(false);
+
+                            await fileStream.WriteAsync(buffer.AsMemory(0, readSize), innerCts.Token);
+
+                            BytesDone = fileStream.Length;
+                            BytesTotal = responseLength;
+
+                            if (DateTime.UtcNow > _nextUpdate)
                             {
-                                await fileStream.WriteAsync(buffer.AsMemory(0, read));
+                                Speed = fileStream.Length - _bytesLastUpdate;
 
-                                BytesDone = fileStream.Length;
+                                _nextUpdate = DateTime.UtcNow.AddSeconds(1);
-                                BytesTotal = responseLength;
+                                _bytesLastUpdate = fileStream.Length;
 
-                                if (DateTime.UtcNow > _nextUpdate)
+                                timeout = DateTimeOffset.UtcNow.AddHours(1);
+
+                                DownloadProgress?.Invoke(this, new DownloadProgressEventArgs
                                 {
-                                    Speed = fileStream.Length - _bytesLastUpdate;
+                                    Speed = Speed,
-
+                                    BytesDone = BytesDone,
-                                    _nextUpdate = DateTime.UtcNow.AddSeconds(1);
+                                    BytesTotal = BytesTotal
-                                    _bytesLastUpdate = fileStream.Length;
+                                });
-
-                                    timeout = DateTimeOffset.UtcNow.AddHours(1);
-
-                                    DownloadProgress?.Invoke(this, new DownloadProgressEventArgs
-                                    {
-                                        Speed = Speed,
-                                        BytesDone = BytesDone,
-                                        BytesTotal = BytesTotal
-                                    });
-                                }
-                            }
-                            else
-                            {
-                                break;
                             }
                         }
 

server/RdtClient.Service/Services/ThrottledStream.cs

@@ -0,0 +1,208 @@
+using System;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace RdtClient.Service.Services
+{
+    /// <summary>
+    ///     Class for streaming data with throttling support.
+    ///     Taken from Downloader: https://github.com/bezzad/Downloader
+    /// </summary>
+    public class ThrottledStream : Stream
+    {
+        private readonly Bandwidth _bandwidth;
+        private readonly Stream _baseStream;
+        private Int64 _bandwidthLimit;
+
+        /// <summary>
+        ///     Initializes a new instance of the <see cref="T:ThrottledStream" /> class.
+        /// </summary>
+        /// <param name="baseStream">The base stream.</param>
+        /// <param name="bandwidthLimit">The maximum bytes per second that can be transferred through the base stream.</param>
+        /// <exception cref="ArgumentNullException">Thrown when <see cref="baseStream" /> is a null reference.</exception>
+        /// <exception cref="ArgumentOutOfRangeException">Thrown when <see cref="BandwidthLimit" /> is a negative value.</exception>
+        public ThrottledStream(Stream baseStream, Int64 bandwidthLimit)
+        {
+            if (bandwidthLimit < 0)
+            {
+                throw new ArgumentOutOfRangeException(nameof(bandwidthLimit),
+                                                      bandwidthLimit,
+                                                      "The maximum number of bytes per second can't be negative.");
+            }
+
+            _baseStream = baseStream ?? throw new ArgumentNullException(nameof(baseStream));
+            BandwidthLimit = bandwidthLimit;
+
+            _bandwidth = new Bandwidth
+            {
+                BandwidthLimit = BandwidthLimit
+            };
+        }
+
+        public static Int64 Infinite => Int64.MaxValue;
+
+        /// <summary>
+        ///     Bandwidth Limit (in B/s)
+        /// </summary>
+        /// <value>The maximum bytes per second.</value>
+        public Int64 BandwidthLimit
+        {
+            get => _bandwidthLimit;
+            set => _bandwidthLimit = value <= 0 ? Infinite : value;
+        }
+
+        /// <inheritdoc />
+        public override Boolean CanRead => _baseStream.CanRead;
+
+        /// <inheritdoc />
+        public override Boolean CanSeek => _baseStream.CanSeek;
+
+        /// <inheritdoc />
+        public override Boolean CanWrite => _baseStream.CanWrite;
+
+        /// <inheritdoc />
+        public override Int64 Length => _baseStream.Length;
+
+        /// <inheritdoc />
+        public override Int64 Position
+        {
+            get => _baseStream.Position;
+            set => _baseStream.Position = value;
+        }
+
+        /// <inheritdoc />
+        public override void Flush()
+        {
+            _baseStream.Flush();
+        }
+
+        /// <inheritdoc />
+        public override Int64 Seek(Int64 offset, SeekOrigin origin)
+        {
+            return _baseStream.Seek(offset, origin);
+        }
+
+        /// <inheritdoc />
+        public override void SetLength(Int64 value)
+        {
+            _baseStream.SetLength(value);
+        }
+
+        /// <inheritdoc />
+        public override Int32 Read(Byte[] buffer, Int32 offset, Int32 count)
+        {
+            Throttle(count).Wait();
+
+            return _baseStream.Read(buffer, offset, count);
+        }
+
+        public override async Task<Int32> ReadAsync(Byte[] buffer,
+                                                    Int32 offset,
+                                                    Int32 count,
+                                                    CancellationToken cancellationToken)
+        {
+            await Throttle(count).ConfigureAwait(false);
+
+            return await _baseStream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+        }
+
+        /// <inheritdoc />
+        public override void Write(Byte[] buffer, Int32 offset, Int32 count)
+        {
+            Throttle(count).Wait();
+            _baseStream.Write(buffer, offset, count);
+        }
+
+        /// <inheritdoc />
+        public override async Task WriteAsync(Byte[] buffer, Int32 offset, Int32 count, CancellationToken cancellationToken)
+        {
+            await Throttle(count).ConfigureAwait(false);
+            await _baseStream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+        }
+
+        private async Task Throttle(Int32 transmissionVolume)
+        {
+            // Make sure the buffer isn't empty.
+            if (BandwidthLimit > 0 && transmissionVolume > 0)
+            {
+                // Calculate the time to sleep.
+                _bandwidth.CalculateSpeed(transmissionVolume);
+                await Sleep(_bandwidth.PopSpeedRetrieveTime()).ConfigureAwait(false);
+            }
+        }
+
+        private async Task Sleep(Int32 time)
+        {
+            if (time > 0)
+            {
+                await Task.Delay(time).ConfigureAwait(false);
+            }
+        }
+
+        /// <inheritdoc />
+        public override String ToString()
+        {
+            return _baseStream.ToString();
+        }
+    }
+
+    public class Bandwidth
+    {
+        private const Double OneSecond = 1000; // millisecond
+        private Int64 _count;
+        private Int32 _lastSecondCheckpoint;
+        private Int64 _lastTransferredBytesCount;
+        private Int32 _speedRetrieveTime;
+
+        public Bandwidth()
+        {
+            BandwidthLimit = Int64.MaxValue;
+            Reset();
+        }
+
+        public Double Speed { get; private set; }
+        public Double AverageSpeed { get; private set; }
+        public Int64 BandwidthLimit { get; set; }
+
+        public void CalculateSpeed(Int64 receivedBytesCount)
+        {
+            var elapsedTime = (Environment.TickCount - _lastSecondCheckpoint) + 1;
+            receivedBytesCount = Interlocked.Add(ref _lastTransferredBytesCount, receivedBytesCount);
+            var momentSpeed = (receivedBytesCount * OneSecond) / elapsedTime; // B/s
+
+            if (OneSecond < elapsedTime)
+            {
+                Speed = momentSpeed;
+                AverageSpeed = ((AverageSpeed * _count) + Speed) / (_count + 1);
+                _count++;
+                SecondCheckpoint();
+            }
+
+            if (momentSpeed >= BandwidthLimit)
+            {
+                var expectedTime = (receivedBytesCount * OneSecond) / BandwidthLimit;
+                Interlocked.Add(ref _speedRetrieveTime, (Int32)expectedTime - elapsedTime);
+            }
+        }
+
+        public Int32 PopSpeedRetrieveTime()
+        {
+            return Interlocked.Exchange(ref _speedRetrieveTime, 0);
+        }
+
+        public void Reset()
+        {
+            SecondCheckpoint();
+            _count = 0;
+            Speed = 0;
+            AverageSpeed = 0;
+        }
+
+        private void SecondCheckpoint()
+        {
+            Interlocked.Exchange(ref _lastSecondCheckpoint, Environment.TickCount);
+            Interlocked.Exchange(ref _lastTransferredBytesCount, 0);
+        }
+    }
+}

server/RdtClient.Web/RdtClient.Web.csproj

@@ -4,7 +4,7 @@
     <TargetFramework>net5.0</TargetFramework>
     <OutputType>Exe</OutputType>
     <UserSecretsId>94c24cba-f03f-4453-a671-3640b517c573</UserSecretsId>
-    <Version>1.9.7</Version>
+    <Version>1.9.8</Version>
   </PropertyGroup>
 
   <ItemGroup>