diff --git a/package.json b/package.json
index bb1f232..49a9622 100644
--- a/package.json
+++ b/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": {
diff --git a/server/RdtClient.Service/Services/Downloaders/SimpleDownloader.cs b/server/RdtClient.Service/Services/Downloaders/SimpleDownloader.cs
index c2e56b0..d8444f7 100644
--- a/server/RdtClient.Service/Services/Downloaders/SimpleDownloader.cs
+++ b/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
DownloadComplete;
public event EventHandler 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 buffer = new Byte[64 * 1024];
-
- while (fileStream.Length < response.ContentLength && !_cancelled)
+ var speedLimit = Settings.Get.DownloadMaxSpeed * BufferSize * 1024L;
+
+ if (speedLimit == 0)
{
- 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;
- BytesTotal = responseLength;
+ _nextUpdate = DateTime.UtcNow.AddSeconds(1);
+ _bytesLastUpdate = fileStream.Length;
- if (DateTime.UtcNow > _nextUpdate)
+ timeout = DateTimeOffset.UtcNow.AddHours(1);
+
+ DownloadProgress?.Invoke(this, new DownloadProgressEventArgs
{
- Speed = fileStream.Length - _bytesLastUpdate;
-
- _nextUpdate = DateTime.UtcNow.AddSeconds(1);
- _bytesLastUpdate = fileStream.Length;
-
- timeout = DateTimeOffset.UtcNow.AddHours(1);
-
- DownloadProgress?.Invoke(this, new DownloadProgressEventArgs
- {
- Speed = Speed,
- BytesDone = BytesDone,
- BytesTotal = BytesTotal
- });
- }
- }
- else
- {
- break;
+ Speed = Speed,
+ BytesDone = BytesDone,
+ BytesTotal = BytesTotal
+ });
}
}
diff --git a/server/RdtClient.Service/Services/ThrottledStream.cs b/server/RdtClient.Service/Services/ThrottledStream.cs
new file mode 100644
index 0000000..3087ce9
--- /dev/null
+++ b/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
+{
+ ///
+ /// Class for streaming data with throttling support.
+ /// Taken from Downloader: https://github.com/bezzad/Downloader
+ ///
+ public class ThrottledStream : Stream
+ {
+ private readonly Bandwidth _bandwidth;
+ private readonly Stream _baseStream;
+ private Int64 _bandwidthLimit;
+
+ ///
+ /// Initializes a new instance of the class.
+ ///
+ /// The base stream.
+ /// The maximum bytes per second that can be transferred through the base stream.
+ /// Thrown when is a null reference.
+ /// Thrown when is a negative value.
+ 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;
+
+ ///
+ /// Bandwidth Limit (in B/s)
+ ///
+ /// The maximum bytes per second.
+ public Int64 BandwidthLimit
+ {
+ get => _bandwidthLimit;
+ set => _bandwidthLimit = value <= 0 ? Infinite : value;
+ }
+
+ ///
+ public override Boolean CanRead => _baseStream.CanRead;
+
+ ///
+ public override Boolean CanSeek => _baseStream.CanSeek;
+
+ ///
+ public override Boolean CanWrite => _baseStream.CanWrite;
+
+ ///
+ public override Int64 Length => _baseStream.Length;
+
+ ///
+ public override Int64 Position
+ {
+ get => _baseStream.Position;
+ set => _baseStream.Position = value;
+ }
+
+ ///
+ public override void Flush()
+ {
+ _baseStream.Flush();
+ }
+
+ ///
+ public override Int64 Seek(Int64 offset, SeekOrigin origin)
+ {
+ return _baseStream.Seek(offset, origin);
+ }
+
+ ///
+ public override void SetLength(Int64 value)
+ {
+ _baseStream.SetLength(value);
+ }
+
+ ///
+ public override Int32 Read(Byte[] buffer, Int32 offset, Int32 count)
+ {
+ Throttle(count).Wait();
+
+ return _baseStream.Read(buffer, offset, count);
+ }
+
+ public override async Task ReadAsync(Byte[] buffer,
+ Int32 offset,
+ Int32 count,
+ CancellationToken cancellationToken)
+ {
+ await Throttle(count).ConfigureAwait(false);
+
+ return await _baseStream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+ }
+
+ ///
+ public override void Write(Byte[] buffer, Int32 offset, Int32 count)
+ {
+ Throttle(count).Wait();
+ _baseStream.Write(buffer, offset, count);
+ }
+
+ ///
+ 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);
+ }
+ }
+
+ ///
+ 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);
+ }
+ }
+}
diff --git a/server/RdtClient.Web/RdtClient.Web.csproj b/server/RdtClient.Web/RdtClient.Web.csproj
index e1960d3..1432bfa 100644
--- a/server/RdtClient.Web/RdtClient.Web.csproj
+++ b/server/RdtClient.Web/RdtClient.Web.csproj
@@ -4,7 +4,7 @@
net5.0
Exe
94c24cba-f03f-4453-a671-3640b517c573
- 1.9.7
+ 1.9.8