flywithu/CryptoExchange.Net
1
0
Fork 0
mirror of https://github.com/JKorf/CryptoExchange.Net.git synced 2026-08-16 19:03:01 +00:00
Code Issues Releases Wiki Activity

Feature/ratelimit refactor (#197)

Browse Source
This commit is contained in:
Jan Korf
2024-04-16 14:55:27 +02:00
committed by GitHub
parent 2dbd5be924
commit 1b1961db00
57 changed files with 2294 additions and 688 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CryptoExchange.Net/Sockets/CryptoExchangeWebSocketClient.cs
+28 -16
View File
@@ -2,6 +2,7 @@
using CryptoExchange.Net.Logging.Extensions;
using CryptoExchange.Net.Objects;
using CryptoExchange.Net.Objects.Sockets;
using CryptoExchange.Net.RateLimiting;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Concurrent;
@@ -45,6 +46,7 @@ namespace CryptoExchange.Net.Sockets
private bool _disposed;
private ProcessState _processState;
private DateTime _lastReconnectTime;
private string _baseAddress;
private const int _receiveBufferSize = 1048576;
private const int _sendBufferSize = 4096;
@@ -110,6 +112,9 @@ namespace CryptoExchange.Net.Sockets
/// <inheritdoc />
public event Func<int, Task>? OnRequestSent;
/// <inheritdoc />
public event Func<int, Task>? OnRequestRateLimited;
/// <inheritdoc />
public event Func<Exception, Task>? OnError;
@@ -143,17 +148,19 @@ namespace CryptoExchange.Net.Sockets
_closeSem = new SemaphoreSlim(1, 1);
_socket = CreateSocket();
_baseAddress = $"{Uri.Scheme}://{Uri.Host}";
}
/// <inheritdoc />
public virtual async Task<bool> ConnectAsync()
public virtual async Task<CallResult> ConnectAsync()
{
if (!await ConnectInternalAsync().ConfigureAwait(false))
return false;
var connectResult = await ConnectInternalAsync().ConfigureAwait(false);
if (!connectResult)
return connectResult;
await (OnOpen?.Invoke() ?? Task.CompletedTask).ConfigureAwait(false);
_processTask = ProcessAsync();
return true;
return connectResult;
}
/// <summary>
@@ -188,11 +195,19 @@ namespace CryptoExchange.Net.Sockets
return socket;
}
private async Task<bool> ConnectInternalAsync()
private async Task<CallResult> ConnectInternalAsync()
{
_logger.SocketConnecting(Id);
try
{
if (Parameters.RateLimiter != null)
{
var definition = new RequestDefinition(Id.ToString(), HttpMethod.Get);
var limitResult = await Parameters.RateLimiter.ProcessAsync(_logger, Id, RateLimitItemType.Connection, definition, _baseAddress, null, 1, Parameters.RateLimitingBehaviour, _ctsSource.Token).ConfigureAwait(false);
if (!limitResult)
return new CallResult(new ClientRateLimitError("Connection limit reached"));
}
using CancellationTokenSource tcs = new(TimeSpan.FromSeconds(10));
using var linked = CancellationTokenSource.CreateLinkedTokenSource(tcs.Token, _ctsSource.Token);
await _socket.ConnectAsync(Uri, linked.Token).ConfigureAwait(false);
@@ -204,11 +219,11 @@ namespace CryptoExchange.Net.Sockets
// if _ctsSource was canceled this was already logged
_logger.SocketConnectionFailed(Id, e.Message, e);
}
return false;
return new CallResult(new CantConnectError());
}
_logger.SocketConnected(Id, Uri);
return true;
return new CallResult(null);
}
/// <inheritdoc />
@@ -407,9 +422,9 @@ namespace CryptoExchange.Net.Sockets
/// <returns></returns>
private async Task SendLoopAsync()
{
var requestDefinition = new RequestDefinition(Id.ToString(), HttpMethod.Get);
try
{
var limitKey = Uri.ToString() + "/" + Id.ToString();
while (true)
{
if (_ctsSource.IsCancellationRequested)
@@ -422,16 +437,13 @@ namespace CryptoExchange.Net.Sockets
while (_sendBuffer.TryDequeue(out var data))
{
if (Parameters.RateLimiters != null)
if (Parameters.RateLimiter != null)
{
foreach(var ratelimiter in Parameters.RateLimiters)
var limitResult = await Parameters.RateLimiter.ProcessAsync(_logger, data.Id, RateLimitItemType.Request, requestDefinition, _baseAddress, null, data.Weight, Parameters.RateLimitingBehaviour, _ctsSource.Token).ConfigureAwait(false);
if (!limitResult)
{
var limitResult = await ratelimiter.LimitRequestAsync(_logger, limitKey, HttpMethod.Get, false, null, RateLimitingBehaviour.Wait, data.Weight, _ctsSource.Token).ConfigureAwait(false);
if (limitResult.Success)
{
if (limitResult.Data > 0)
_logger.SocketSendDelayedBecauseOfRateLimit(Id, data.Id, limitResult.Data);
}
await (OnRequestRateLimited?.Invoke(data.Id) ?? Task.CompletedTask).ConfigureAwait(false);
continue;
}
}