mirror of
https://github.com/JKorf/CryptoExchange.Net.git
synced 2026-08-12 00:43:03 +00:00
Compare commits
8 Commits
8.8.0
..
9.0.0-beta3
| Author | SHA1 | Date | |
|---|---|---|---|
| 3d6267da93 | |||
| 8def7f32af | |||
| ac295de9f6 | |||
| d412e0895e | |||
| 1f9e2b4fcb | |||
| b13cff5a95 | |||
| 4c050744ad | |||
| 3b15c35a02 |
@@ -106,6 +106,7 @@ namespace CryptoExchange.Net.UnitTests
|
||||
for(var i = 1; i <= 10; i++)
|
||||
{
|
||||
evnt.Set();
|
||||
await Task.Delay(1); // Wait for the continuation.
|
||||
Assert.That(10 - i == waiters.Count(w => w.Status != TaskStatus.RanToCompletion));
|
||||
}
|
||||
|
||||
|
||||
@@ -176,12 +176,12 @@ namespace CryptoExchange.Net.UnitTests
|
||||
|
||||
for (var i = 0; i < requests + 1; i++)
|
||||
{
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, default);
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, null, default);
|
||||
Assert.That(i == requests? triggered : !triggered);
|
||||
}
|
||||
triggered = false;
|
||||
await Task.Delay((int)Math.Round(perSeconds * 1000) + 10);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, default);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, null, default);
|
||||
Assert.That(!triggered);
|
||||
}
|
||||
|
||||
@@ -201,7 +201,7 @@ namespace CryptoExchange.Net.UnitTests
|
||||
rateLimiter.RateLimitTriggered += (x) => { evnt = x; };
|
||||
for (var i = 0; i < 2; i++)
|
||||
{
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, default);
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, null, default);
|
||||
bool expected = i == 1 ? (expectLimiting ? evnt.DelayTime > TimeSpan.Zero : evnt == null) : evnt == null;
|
||||
Assert.That(expected);
|
||||
}
|
||||
@@ -222,9 +222,9 @@ namespace CryptoExchange.Net.UnitTests
|
||||
RateLimitEvent evnt = null;
|
||||
rateLimiter.RateLimitTriggered += (x) => { evnt = x; };
|
||||
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition1, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, default);
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition1, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, null, default);
|
||||
Assert.That(evnt == null);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition2, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, default);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition2, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, null, default);
|
||||
Assert.That(expectLimiting ? evnt != null : evnt == null);
|
||||
}
|
||||
|
||||
@@ -243,12 +243,12 @@ namespace CryptoExchange.Net.UnitTests
|
||||
|
||||
for (var i = 0; i < requests + 1; i++)
|
||||
{
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, default);
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, null, default);
|
||||
Assert.That(i == requests ? triggered : !triggered);
|
||||
}
|
||||
triggered = false;
|
||||
await Task.Delay((int)Math.Round(perSeconds * 1000) + 10);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, default);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, null, default);
|
||||
Assert.That(!triggered);
|
||||
}
|
||||
|
||||
@@ -266,7 +266,7 @@ namespace CryptoExchange.Net.UnitTests
|
||||
rateLimiter.RateLimitTriggered += (x) => { evnt = x; };
|
||||
for (var i = 0; i < 2; i++)
|
||||
{
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, default);
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, null, default);
|
||||
bool expected = i == 1 ? (expectLimited ? evnt.DelayTime > TimeSpan.Zero : evnt == null) : evnt == null;
|
||||
Assert.That(expected);
|
||||
}
|
||||
@@ -286,7 +286,7 @@ namespace CryptoExchange.Net.UnitTests
|
||||
rateLimiter.RateLimitTriggered += (x) => { evnt = x; };
|
||||
for (var i = 0; i < 2; i++)
|
||||
{
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, default);
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, null, default);
|
||||
bool expected = i == 1 ? (expectLimited ? evnt.DelayTime > TimeSpan.Zero : evnt == null) : evnt == null;
|
||||
Assert.That(expected);
|
||||
}
|
||||
@@ -309,9 +309,9 @@ namespace CryptoExchange.Net.UnitTests
|
||||
RateLimitEvent evnt = null;
|
||||
rateLimiter.RateLimitTriggered += (x) => { evnt = x; };
|
||||
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition1, "https://test.com", key1, 1, RateLimitingBehaviour.Wait, default);
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition1, "https://test.com", key1, 1, RateLimitingBehaviour.Wait, null, default);
|
||||
Assert.That(evnt == null);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition2, "https://test.com", key2, 1, RateLimitingBehaviour.Wait, default);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition2, "https://test.com", key2, 1, RateLimitingBehaviour.Wait, null, default);
|
||||
Assert.That(expectLimited ? evnt != null : evnt == null);
|
||||
}
|
||||
|
||||
@@ -328,9 +328,9 @@ namespace CryptoExchange.Net.UnitTests
|
||||
RateLimitEvent evnt = null;
|
||||
rateLimiter.RateLimitTriggered += (x) => { evnt = x; };
|
||||
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition1, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, default);
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition1, "https://test.com", "123", 1, RateLimitingBehaviour.Wait, null, default);
|
||||
Assert.That(evnt == null);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition2, "https://test.com", null, 1, RateLimitingBehaviour.Wait, default);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition2, "https://test.com", null, 1, RateLimitingBehaviour.Wait, null, default);
|
||||
Assert.That(expectLimited ? evnt != null : evnt == null);
|
||||
}
|
||||
|
||||
@@ -348,9 +348,9 @@ namespace CryptoExchange.Net.UnitTests
|
||||
RateLimitEvent evnt = null;
|
||||
rateLimiter.RateLimitTriggered += (x) => { evnt = x; };
|
||||
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition1, host1, "123", 1, RateLimitingBehaviour.Wait, default);
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition1, host1, "123", 1, RateLimitingBehaviour.Wait, null, default);
|
||||
Assert.That(evnt == null);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition1, host2, "123", 1, RateLimitingBehaviour.Wait, default);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Request, requestDefinition1, host2, "123", 1, RateLimitingBehaviour.Wait, null, default);
|
||||
Assert.That(expectLimited ? evnt != null : evnt == null);
|
||||
}
|
||||
|
||||
@@ -365,9 +365,9 @@ namespace CryptoExchange.Net.UnitTests
|
||||
RateLimitEvent evnt = null;
|
||||
rateLimiter.RateLimitTriggered += (x) => { evnt = x; };
|
||||
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Connection, new RequestDefinition("1", HttpMethod.Get), host1, "123", 1, RateLimitingBehaviour.Wait, default);
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Connection, new RequestDefinition("1", HttpMethod.Get), host1, "123", 1, RateLimitingBehaviour.Wait, null, default);
|
||||
Assert.That(evnt == null);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Connection, new RequestDefinition("1", HttpMethod.Get), host2, "123", 1, RateLimitingBehaviour.Wait, default);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Connection, new RequestDefinition("1", HttpMethod.Get), host2, "123", 1, RateLimitingBehaviour.Wait, null, default);
|
||||
Assert.That(expectLimited ? evnt != null : evnt == null);
|
||||
}
|
||||
|
||||
@@ -381,8 +381,8 @@ namespace CryptoExchange.Net.UnitTests
|
||||
rateLimiter.RateLimitTriggered += (x) => { evnt = x; };
|
||||
var ct = new CancellationTokenSource(TimeSpan.FromSeconds(0.2));
|
||||
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Connection, new RequestDefinition("1", HttpMethod.Get), "https://test.com", "123", 1, RateLimitingBehaviour.Wait, ct.Token);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Connection, new RequestDefinition("1", HttpMethod.Get), "https://test.com", "123", 1, RateLimitingBehaviour.Wait, ct.Token);
|
||||
var result1 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Connection, new RequestDefinition("1", HttpMethod.Get), "https://test.com", "123", 1, RateLimitingBehaviour.Wait, null, ct.Token);
|
||||
var result2 = await rateLimiter.ProcessAsync(new TraceLogger(), 1, RateLimitItemType.Connection, new RequestDefinition("1", HttpMethod.Get), "https://test.com", "123", 1, RateLimitingBehaviour.Wait, null, ct.Token);
|
||||
Assert.That(result2.Error, Is.TypeOf<CancellationRequestedError>());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -155,6 +155,7 @@ namespace CryptoExchange.Net.Clients
|
||||
/// <param name="additionalHeaders">Additional headers for this request</param>
|
||||
/// <param name="weight">Override the request weight for this request definition, for example when the weight depends on the parameters</param>
|
||||
/// <param name="weightSingleLimiter">Specify the weight to apply to the individual rate limit guard for this request</param>
|
||||
/// <param name="rateLimitKeySuffix">An additional optional suffix for the key selector. Can be used to make rate limiting work based on parameters.</param>
|
||||
/// <returns></returns>
|
||||
protected virtual Task<WebCallResult<T>> SendAsync<T>(
|
||||
string baseAddress,
|
||||
@@ -163,7 +164,8 @@ namespace CryptoExchange.Net.Clients
|
||||
CancellationToken cancellationToken,
|
||||
Dictionary<string, string>? additionalHeaders = null,
|
||||
int? weight = null,
|
||||
int? weightSingleLimiter = null)
|
||||
int? weightSingleLimiter = null,
|
||||
string? rateLimitKeySuffix = null)
|
||||
{
|
||||
var parameterPosition = definition.ParameterPosition ?? ParameterPositions[definition.Method];
|
||||
return SendAsync<T>(
|
||||
@@ -174,7 +176,8 @@ namespace CryptoExchange.Net.Clients
|
||||
cancellationToken,
|
||||
additionalHeaders,
|
||||
weight,
|
||||
weightSingleLimiter);
|
||||
weightSingleLimiter,
|
||||
rateLimitKeySuffix);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -189,6 +192,7 @@ namespace CryptoExchange.Net.Clients
|
||||
/// <param name="additionalHeaders">Additional headers for this request</param>
|
||||
/// <param name="weight">Override the request weight for this request definition, for example when the weight depends on the parameters</param>
|
||||
/// <param name="weightSingleLimiter">Specify the weight to apply to the individual rate limit guard for this request</param>
|
||||
/// <param name="rateLimitKeySuffix">An additional optional suffix for the key selector. Can be used to make rate limiting work based on parameters.</param>
|
||||
/// <returns></returns>
|
||||
protected virtual async Task<WebCallResult<T>> SendAsync<T>(
|
||||
string baseAddress,
|
||||
@@ -198,7 +202,8 @@ namespace CryptoExchange.Net.Clients
|
||||
CancellationToken cancellationToken,
|
||||
Dictionary<string, string>? additionalHeaders = null,
|
||||
int? weight = null,
|
||||
int? weightSingleLimiter = null)
|
||||
int? weightSingleLimiter = null,
|
||||
string? rateLimitKeySuffix = null)
|
||||
{
|
||||
string? cacheKey = null;
|
||||
if (ShouldCache(definition))
|
||||
@@ -222,7 +227,7 @@ namespace CryptoExchange.Net.Clients
|
||||
currentTry++;
|
||||
var requestId = ExchangeHelpers.NextId();
|
||||
|
||||
var prepareResult = await PrepareAsync(requestId, baseAddress, definition, cancellationToken, additionalHeaders, weight, weightSingleLimiter).ConfigureAwait(false);
|
||||
var prepareResult = await PrepareAsync(requestId, baseAddress, definition, cancellationToken, additionalHeaders, weight, weightSingleLimiter, rateLimitKeySuffix).ConfigureAwait(false);
|
||||
if (!prepareResult)
|
||||
return new WebCallResult<T>(prepareResult.Error!);
|
||||
|
||||
@@ -236,10 +241,17 @@ namespace CryptoExchange.Net.Clients
|
||||
_logger.RestApiSendRequest(request.RequestId, definition, request.Content, string.IsNullOrEmpty(request.Uri.Query) ? "-" : request.Uri.Query, string.Join(", ", request.GetHeaders().Select(h => h.Key + $"=[{string.Join(",", h.Value)}]")));
|
||||
TotalRequestsMade++;
|
||||
var result = await GetResponseAsync<T>(request, definition.RateLimitGate, cancellationToken).ConfigureAwait(false);
|
||||
if (!result)
|
||||
_logger.RestApiErrorReceived(result.RequestId, result.ResponseStatusCode, (long)Math.Floor(result.ResponseTime!.Value.TotalMilliseconds), result.Error?.ToString());
|
||||
if (result.Error is not CancellationRequestedError)
|
||||
{
|
||||
if (!result)
|
||||
_logger.RestApiErrorReceived(result.RequestId, result.ResponseStatusCode, (long)Math.Floor(result.ResponseTime!.Value.TotalMilliseconds), result.Error?.ToString());
|
||||
else
|
||||
_logger.RestApiResponseReceived(result.RequestId, result.ResponseStatusCode, (long)Math.Floor(result.ResponseTime!.Value.TotalMilliseconds), OutputOriginalData ? result.OriginalData : "[Data only available when OutputOriginal = true]");
|
||||
}
|
||||
else
|
||||
_logger.RestApiResponseReceived(result.RequestId, result.ResponseStatusCode, (long)Math.Floor(result.ResponseTime!.Value.TotalMilliseconds), OutputOriginalData ? result.OriginalData : "[Data only available when OutputOriginal = true]");
|
||||
{
|
||||
_logger.RestApiCancellationRequested(result.RequestId);
|
||||
}
|
||||
|
||||
if (await ShouldRetryRequestAsync(definition.RateLimitGate, result, currentTry).ConfigureAwait(false))
|
||||
continue;
|
||||
@@ -264,6 +276,7 @@ namespace CryptoExchange.Net.Clients
|
||||
/// <param name="additionalHeaders">Additional headers for this request</param>
|
||||
/// <param name="weight">Override the request weight for this request</param>
|
||||
/// <param name="weightSingleLimiter">Specify the weight to apply to the individual rate limit guard for this request</param>
|
||||
/// <param name="rateLimitKeySuffix">An additional optional suffix for the key selector</param>
|
||||
/// <returns></returns>
|
||||
/// <exception cref="Exception"></exception>
|
||||
protected virtual async Task<CallResult> PrepareAsync(
|
||||
@@ -273,7 +286,8 @@ namespace CryptoExchange.Net.Clients
|
||||
CancellationToken cancellationToken,
|
||||
Dictionary<string, string>? additionalHeaders = null,
|
||||
int? weight = null,
|
||||
int? weightSingleLimiter = null)
|
||||
int? weightSingleLimiter = null,
|
||||
string? rateLimitKeySuffix = null)
|
||||
{
|
||||
// Time sync
|
||||
if (definition.Authenticated)
|
||||
@@ -308,7 +322,7 @@ namespace CryptoExchange.Net.Clients
|
||||
|
||||
if (ClientOptions.RateLimiterEnabled)
|
||||
{
|
||||
var limitResult = await definition.RateLimitGate.ProcessAsync(_logger, requestId, RateLimitItemType.Request, definition, baseAddress, AuthenticationProvider?._credentials.Key, requestWeight, ClientOptions.RateLimitingBehaviour, cancellationToken).ConfigureAwait(false);
|
||||
var limitResult = await definition.RateLimitGate.ProcessAsync(_logger, requestId, RateLimitItemType.Request, definition, baseAddress, AuthenticationProvider?._credentials.Key, requestWeight, ClientOptions.RateLimitingBehaviour, rateLimitKeySuffix, cancellationToken).ConfigureAwait(false);
|
||||
if (!limitResult)
|
||||
return new CallResult(limitResult.Error!);
|
||||
}
|
||||
@@ -323,7 +337,7 @@ namespace CryptoExchange.Net.Clients
|
||||
if (ClientOptions.RateLimiterEnabled)
|
||||
{
|
||||
var singleRequestWeight = weightSingleLimiter ?? 1;
|
||||
var limitResult = await definition.RateLimitGate.ProcessSingleAsync(_logger, requestId, definition.LimitGuard, RateLimitItemType.Request, definition, baseAddress, AuthenticationProvider?._credentials.Key, singleRequestWeight, ClientOptions.RateLimitingBehaviour, cancellationToken).ConfigureAwait(false);
|
||||
var limitResult = await definition.RateLimitGate.ProcessSingleAsync(_logger, requestId, definition.LimitGuard, RateLimitItemType.Request, definition, baseAddress, AuthenticationProvider?._credentials.Key, singleRequestWeight, ClientOptions.RateLimitingBehaviour, rateLimitKeySuffix, cancellationToken).ConfigureAwait(false);
|
||||
if (!limitResult)
|
||||
return new CallResult(limitResult.Error!);
|
||||
}
|
||||
@@ -609,7 +623,7 @@ namespace CryptoExchange.Net.Clients
|
||||
|
||||
if (ClientOptions.RateLimiterEnabled)
|
||||
{
|
||||
var limitResult = await gate.ProcessAsync(_logger, requestId, RateLimitItemType.Request, new RequestDefinition(uri.AbsolutePath.TrimStart('/'), method) { Authenticated = signed }, uri.Host, AuthenticationProvider?._credentials.Key, requestWeight, ClientOptions.RateLimitingBehaviour, cancellationToken).ConfigureAwait(false);
|
||||
var limitResult = await gate.ProcessAsync(_logger, requestId, RateLimitItemType.Request, new RequestDefinition(uri.AbsolutePath.TrimStart('/'), method) { Authenticated = signed }, uri.Host, AuthenticationProvider?._credentials.Key, requestWeight, ClientOptions.RateLimitingBehaviour, null, cancellationToken).ConfigureAwait(false);
|
||||
if (!limitResult)
|
||||
return new CallResult<IRequest>(limitResult.Error!);
|
||||
}
|
||||
|
||||
@@ -600,7 +600,8 @@ namespace CryptoExchange.Net.Clients
|
||||
RateLimiter = ClientOptions.RateLimiterEnabled ? RateLimiter : null,
|
||||
RateLimitingBehavior = ClientOptions.RateLimitingBehaviour,
|
||||
Proxy = ClientOptions.Proxy,
|
||||
Timeout = ApiOptions.SocketNoDataTimeout ?? ClientOptions.SocketNoDataTimeout
|
||||
Timeout = ApiOptions.SocketNoDataTimeout ?? ClientOptions.SocketNoDataTimeout,
|
||||
ReceiveBufferSize = ClientOptions.ReceiveBufferSize,
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
@@ -670,8 +671,11 @@ namespace CryptoExchange.Net.Clients
|
||||
var tasks = new List<Task>();
|
||||
{
|
||||
var socketList = socketConnections.Values;
|
||||
foreach (var connection in socketList.Where(s => !s.DedicatedRequestConnection.IsDedicatedRequestConnection))
|
||||
tasks.Add(connection.CloseAsync());
|
||||
foreach (var connection in socketList)
|
||||
{
|
||||
foreach(var subscription in connection.Subscriptions.Where(x => x.UserSubscription))
|
||||
tasks.Add(connection.CloseAsync(subscription));
|
||||
}
|
||||
}
|
||||
|
||||
await Task.WhenAll(tasks.ToArray()).ConfigureAwait(false);
|
||||
|
||||
@@ -21,6 +21,7 @@ namespace CryptoExchange.Net.Logging.Extensions
|
||||
private static readonly Action<ILogger, string, Exception?> _restApiCheckingCache;
|
||||
private static readonly Action<ILogger, string, Exception?> _restApiCacheHit;
|
||||
private static readonly Action<ILogger, string, Exception?> _restApiCacheNotHit;
|
||||
private static readonly Action<ILogger, int?, Exception?> _restApiCancellationRequested;
|
||||
|
||||
static RestApiClientLoggingExtensions()
|
||||
{
|
||||
@@ -83,6 +84,12 @@ namespace CryptoExchange.Net.Logging.Extensions
|
||||
LogLevel.Trace,
|
||||
new EventId(4011, "RestApiCacheNotHit"),
|
||||
"Cache not hit for key {Key}");
|
||||
|
||||
_restApiCancellationRequested = LoggerMessage.Define<int?>(
|
||||
LogLevel.Debug,
|
||||
new EventId(4012, "RestApiCancellationRequested"),
|
||||
"[Req {RequestId}] Request cancelled by user");
|
||||
|
||||
}
|
||||
|
||||
public static void RestApiErrorReceived(this ILogger logger, int? requestId, HttpStatusCode? responseStatusCode, long responseTime, string? error)
|
||||
@@ -144,5 +151,9 @@ namespace CryptoExchange.Net.Logging.Extensions
|
||||
{
|
||||
_restApiCacheNotHit(logger, key, null);
|
||||
}
|
||||
public static void RestApiCancellationRequested(this ILogger logger, int? requestId)
|
||||
{
|
||||
_restApiCancellationRequested(logger, requestId, null);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -32,44 +32,51 @@ namespace CryptoExchange.Net.Objects
|
||||
/// Wait for the AutoResetEvent to be set
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
public Task<bool> WaitAsync(TimeSpan? timeout = null, CancellationToken ct = default)
|
||||
public async Task<bool> WaitAsync(TimeSpan? timeout = null, CancellationToken ct = default)
|
||||
{
|
||||
lock (_waits)
|
||||
CancellationTokenRegistration registration = default;
|
||||
try
|
||||
{
|
||||
if (_signaled)
|
||||
Task<bool> waiter = _completed;
|
||||
lock (_waits)
|
||||
{
|
||||
if(_reset)
|
||||
_signaled = false;
|
||||
return _completed;
|
||||
}
|
||||
else
|
||||
{
|
||||
if (ct.IsCancellationRequested)
|
||||
return _completed;
|
||||
|
||||
var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
if (timeout.HasValue)
|
||||
if (_signaled)
|
||||
{
|
||||
var timeoutSource = new CancellationTokenSource(timeout.Value);
|
||||
var cancellationSource = CancellationTokenSource.CreateLinkedTokenSource(timeoutSource.Token, ct);
|
||||
ct = cancellationSource.Token;
|
||||
if (_reset)
|
||||
_signaled = false;
|
||||
}
|
||||
|
||||
var registration = ct.Register(() =>
|
||||
else if (!ct.IsCancellationRequested)
|
||||
{
|
||||
lock (_waits)
|
||||
var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
if (timeout.HasValue)
|
||||
{
|
||||
tcs.TrySetResult(false);
|
||||
|
||||
// Not the cleanest but it works
|
||||
_waits = new Queue<TaskCompletionSource<bool>>(_waits.Where(i => i != tcs));
|
||||
var timeoutSource = new CancellationTokenSource(timeout.Value);
|
||||
var cancellationSource = CancellationTokenSource.CreateLinkedTokenSource(timeoutSource.Token, ct);
|
||||
ct = cancellationSource.Token;
|
||||
}
|
||||
}, useSynchronizationContext: false);
|
||||
|
||||
|
||||
_waits.Enqueue(tcs);
|
||||
return tcs.Task;
|
||||
registration = ct.Register(() =>
|
||||
{
|
||||
lock (_waits)
|
||||
{
|
||||
tcs.TrySetResult(false);
|
||||
|
||||
// Not the cleanest but it works
|
||||
_waits = new Queue<TaskCompletionSource<bool>>(_waits.Where(i => i != tcs));
|
||||
}
|
||||
}, useSynchronizationContext: false);
|
||||
|
||||
|
||||
_waits.Enqueue(tcs);
|
||||
waiter = tcs.Task;
|
||||
}
|
||||
}
|
||||
|
||||
return await waiter.ConfigureAwait(false);
|
||||
}
|
||||
finally
|
||||
{
|
||||
registration.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -52,6 +52,15 @@ namespace CryptoExchange.Net.Objects.Options
|
||||
/// </summary>
|
||||
public TimeSpan? ConnectDelayAfterRateLimited { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// The buffer size to use for receiving data. Leave unset to use the default buffer size.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Only specify this if you are creating a significant amount of connections and understand the typical message length we receive from the exchange.
|
||||
/// Setting this too low can increase memory consumption and allocations.
|
||||
/// </remarks>
|
||||
public int? ReceiveBufferSize { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Create a copy of this options
|
||||
/// </summary>
|
||||
@@ -72,6 +81,7 @@ namespace CryptoExchange.Net.Objects.Options
|
||||
item.RequestTimeout = RequestTimeout;
|
||||
item.RateLimitingBehaviour = RateLimitingBehaviour;
|
||||
item.RateLimiterEnabled = RateLimiterEnabled;
|
||||
item.ReceiveBufferSize = ReceiveBufferSize;
|
||||
return item;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -64,6 +64,11 @@ namespace CryptoExchange.Net.Objects.Sockets
|
||||
/// </summary>
|
||||
public Encoding Encoding { get; set; } = Encoding.UTF8;
|
||||
|
||||
/// <summary>
|
||||
/// The buffer size to use for receiving data
|
||||
/// </summary>
|
||||
public int? ReceiveBufferSize { get; set; } = null;
|
||||
|
||||
/// <summary>
|
||||
/// ctor
|
||||
/// </summary>
|
||||
|
||||
@@ -90,7 +90,7 @@ namespace CryptoExchange.Net.RateLimiting.Guards
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public LimitCheck Check(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight)
|
||||
public LimitCheck Check(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight, string? keySuffix)
|
||||
{
|
||||
foreach(var filter in _filters)
|
||||
{
|
||||
@@ -101,7 +101,7 @@ namespace CryptoExchange.Net.RateLimiting.Guards
|
||||
if (type == RateLimitItemType.Connection)
|
||||
requestWeight = _connectionWeight ?? requestWeight;
|
||||
|
||||
var key = _keySelector(definition, host, apiKey);
|
||||
var key = _keySelector(definition, host, apiKey) + keySuffix;
|
||||
if (!_trackers.TryGetValue(key, out var tracker))
|
||||
{
|
||||
tracker = CreateTracker();
|
||||
@@ -116,7 +116,7 @@ namespace CryptoExchange.Net.RateLimiting.Guards
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public RateLimitState ApplyWeight(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight)
|
||||
public RateLimitState ApplyWeight(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight, string? keySuffix)
|
||||
{
|
||||
foreach (var filter in _filters)
|
||||
{
|
||||
@@ -127,7 +127,7 @@ namespace CryptoExchange.Net.RateLimiting.Guards
|
||||
if (type == RateLimitItemType.Connection)
|
||||
requestWeight = _connectionWeight ?? requestWeight;
|
||||
|
||||
var key = _keySelector(definition, host, apiKey);
|
||||
var key = _keySelector(definition, host, apiKey) + keySuffix;
|
||||
var tracker = _trackers[key];
|
||||
tracker.ApplyWeight(requestWeight);
|
||||
return RateLimitState.Applied(Limit, TimeSpan, tracker.Current);
|
||||
|
||||
@@ -42,7 +42,7 @@ namespace CryptoExchange.Net.RateLimiting.Guards
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public LimitCheck Check(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight)
|
||||
public LimitCheck Check(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight, string? keySuffix)
|
||||
{
|
||||
if (type != Type)
|
||||
return LimitCheck.NotApplicable;
|
||||
@@ -55,7 +55,7 @@ namespace CryptoExchange.Net.RateLimiting.Guards
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public RateLimitState ApplyWeight(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight)
|
||||
public RateLimitState ApplyWeight(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight, string? keySuffix)
|
||||
{
|
||||
return RateLimitState.NotApplied;
|
||||
}
|
||||
|
||||
@@ -19,7 +19,7 @@ namespace CryptoExchange.Net.RateLimiting.Guards
|
||||
/// <summary>
|
||||
/// Endpoint limit per API key
|
||||
/// </summary>
|
||||
public static Func<RequestDefinition, string, string?, string> PerApiKey { get; } = new Func<RequestDefinition, string, string?, string>((def, host, key) => def.Path + def.Method);
|
||||
public static Func<RequestDefinition, string, string?, string> PerApiKey { get; } = new Func<RequestDefinition, string, string?, string>((def, host, key) => def.Path + def.Method + key);
|
||||
|
||||
private readonly Dictionary<string, IWindowTracker> _trackers;
|
||||
private readonly RateLimitWindowType _windowType;
|
||||
@@ -53,9 +53,9 @@ namespace CryptoExchange.Net.RateLimiting.Guards
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public LimitCheck Check(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight)
|
||||
public LimitCheck Check(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight, string? keySuffix)
|
||||
{
|
||||
var key = _keySelector(definition, host, apiKey);
|
||||
var key = _keySelector(definition, host, apiKey) + keySuffix;
|
||||
if (!_trackers.TryGetValue(key, out var tracker))
|
||||
{
|
||||
tracker = CreateTracker();
|
||||
@@ -70,9 +70,9 @@ namespace CryptoExchange.Net.RateLimiting.Guards
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public RateLimitState ApplyWeight(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight)
|
||||
public RateLimitState ApplyWeight(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight, string? keySuffix)
|
||||
{
|
||||
var key = _keySelector(definition, host, apiKey);
|
||||
var key = _keySelector(definition, host, apiKey) + keySuffix;
|
||||
var tracker = _trackers[key];
|
||||
tracker.ApplyWeight(requestWeight);
|
||||
return RateLimitState.Applied(_limit, _period, tracker.Current);
|
||||
|
||||
@@ -53,9 +53,10 @@ namespace CryptoExchange.Net.RateLimiting.Interfaces
|
||||
/// <param name="apiKey">The API key</param>
|
||||
/// <param name="requestWeight">Request weight</param>
|
||||
/// <param name="behaviour">Behaviour when rate limit is hit</param>
|
||||
/// <param name="keySuffix">An additional optional suffix for the key selector. Can be used to make rate limiting work based on parameters.</param>
|
||||
/// <param name="ct">Cancelation token</param>
|
||||
/// <returns>Error if RateLimitingBehaviour is Fail and rate limit is hit</returns>
|
||||
Task<CallResult> ProcessAsync(ILogger logger, int itemId, RateLimitItemType type, RequestDefinition definition, string baseAddress, string? apiKey, int requestWeight, RateLimitingBehaviour behaviour, CancellationToken ct);
|
||||
Task<CallResult> ProcessAsync(ILogger logger, int itemId, RateLimitItemType type, RequestDefinition definition, string baseAddress, string? apiKey, int requestWeight, RateLimitingBehaviour behaviour, string? keySuffix, CancellationToken ct);
|
||||
|
||||
/// <summary>
|
||||
/// Enforces the rate limit as defined in the request definition. When a rate limit is hit will wait for the rate limit to pass if RateLimitingBehaviour is Wait, or return an error if it is set to Fail
|
||||
@@ -69,8 +70,9 @@ namespace CryptoExchange.Net.RateLimiting.Interfaces
|
||||
/// <param name="apiKey">The API key</param>
|
||||
/// <param name="behaviour">Behaviour when rate limit is hit</param>
|
||||
/// <param name="requestWeight">The weight to apply to the limit guard</param>
|
||||
/// <param name="keySuffix">An additional optional suffix for the key selector. Can be used to make rate limiting work based on parameters.</param>
|
||||
/// <param name="ct">Cancelation token</param>
|
||||
/// <returns>Error if RateLimitingBehaviour is Fail and rate limit is hit</returns>
|
||||
Task<CallResult> ProcessSingleAsync(ILogger logger, int itemId, IRateLimitGuard guard, RateLimitItemType type, RequestDefinition definition, string baseAddress, string? apiKey, int requestWeight, RateLimitingBehaviour behaviour, CancellationToken ct);
|
||||
Task<CallResult> ProcessSingleAsync(ILogger logger, int itemId, IRateLimitGuard guard, RateLimitItemType type, RequestDefinition definition, string baseAddress, string? apiKey, int requestWeight, RateLimitingBehaviour behaviour, string? keySuffix, CancellationToken ct);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -25,8 +25,9 @@ namespace CryptoExchange.Net.RateLimiting.Interfaces
|
||||
/// <param name="host">The host address</param>
|
||||
/// <param name="apiKey">The API key</param>
|
||||
/// <param name="requestWeight">The request weight</param>
|
||||
/// <param name="keySuffix">An additional optional suffix for the key selector. Can be used to make rate limiting work based on parameters.</param>
|
||||
/// <returns></returns>
|
||||
LimitCheck Check(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight);
|
||||
LimitCheck Check(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight, string? keySuffix);
|
||||
|
||||
/// <summary>
|
||||
/// Apply the request to this guard with the specified weight
|
||||
@@ -36,7 +37,8 @@ namespace CryptoExchange.Net.RateLimiting.Interfaces
|
||||
/// <param name="host">The host address</param>
|
||||
/// <param name="apiKey">The API key</param>
|
||||
/// <param name="requestWeight">The request weight</param>
|
||||
/// <param name="keySuffix">An additional optional suffix for the key selector. Can be used to make rate limiting work based on parameters.</param>
|
||||
/// <returns></returns>
|
||||
RateLimitState ApplyWeight(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight);
|
||||
RateLimitState ApplyWeight(RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight, string? keySuffix);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -37,14 +37,14 @@ namespace CryptoExchange.Net.RateLimiting
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<CallResult> ProcessAsync(ILogger logger, int itemId, RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight, RateLimitingBehaviour rateLimitingBehaviour, CancellationToken ct)
|
||||
public async Task<CallResult> ProcessAsync(ILogger logger, int itemId, RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight, RateLimitingBehaviour rateLimitingBehaviour, string? keySuffix, CancellationToken ct)
|
||||
{
|
||||
await _semaphore.WaitAsync(ct).ConfigureAwait(false);
|
||||
bool release = true;
|
||||
_waitingCount++;
|
||||
try
|
||||
{
|
||||
return await CheckGuardsAsync(_guards, logger, itemId, type, definition, host, apiKey, requestWeight, rateLimitingBehaviour, ct).ConfigureAwait(false);
|
||||
return await CheckGuardsAsync(_guards, logger, itemId, type, definition, host, apiKey, requestWeight, rateLimitingBehaviour, keySuffix, ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (TaskCanceledException)
|
||||
{
|
||||
@@ -71,6 +71,7 @@ namespace CryptoExchange.Net.RateLimiting
|
||||
string? apiKey,
|
||||
int requestWeight,
|
||||
RateLimitingBehaviour rateLimitingBehaviour,
|
||||
string? keySuffix,
|
||||
CancellationToken ct)
|
||||
{
|
||||
await _semaphore.WaitAsync(ct).ConfigureAwait(false);
|
||||
@@ -78,7 +79,7 @@ namespace CryptoExchange.Net.RateLimiting
|
||||
_waitingCount++;
|
||||
try
|
||||
{
|
||||
return await CheckGuardsAsync(new IRateLimitGuard[] { guard }, logger, itemId, type, definition, host, apiKey, requestWeight, rateLimitingBehaviour, ct).ConfigureAwait(false);
|
||||
return await CheckGuardsAsync(new IRateLimitGuard[] { guard }, logger, itemId, type, definition, host, apiKey, requestWeight, rateLimitingBehaviour, keySuffix, ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (TaskCanceledException)
|
||||
{
|
||||
@@ -94,12 +95,12 @@ namespace CryptoExchange.Net.RateLimiting
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<CallResult> CheckGuardsAsync(IEnumerable<IRateLimitGuard> guards, ILogger logger, int itemId, RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight, RateLimitingBehaviour rateLimitingBehaviour, CancellationToken ct)
|
||||
private async Task<CallResult> CheckGuardsAsync(IEnumerable<IRateLimitGuard> guards, ILogger logger, int itemId, RateLimitItemType type, RequestDefinition definition, string host, string? apiKey, int requestWeight, RateLimitingBehaviour rateLimitingBehaviour, string? keySuffix, CancellationToken ct)
|
||||
{
|
||||
foreach (var guard in guards)
|
||||
{
|
||||
// Check if a wait is needed for this guard
|
||||
var result = guard.Check(type, definition, host, apiKey, requestWeight);
|
||||
var result = guard.Check(type, definition, host, apiKey, requestWeight, keySuffix);
|
||||
if (result.Delay != TimeSpan.Zero && rateLimitingBehaviour == RateLimitingBehaviour.Fail)
|
||||
{
|
||||
// Delay is needed and limit behaviour is to fail the request
|
||||
@@ -126,14 +127,14 @@ namespace CryptoExchange.Net.RateLimiting
|
||||
RateLimitTriggered?.Invoke(new RateLimitEvent(itemId, _name, guard.Description, definition, host, result.Current, requestWeight, result.Limit, result.Period, result.Delay, rateLimitingBehaviour));
|
||||
await Task.Delay((int)result.Delay.TotalMilliseconds + 1, ct).ConfigureAwait(false);
|
||||
await _semaphore.WaitAsync(ct).ConfigureAwait(false);
|
||||
return await CheckGuardsAsync(guards, logger, itemId, type, definition, host, apiKey, requestWeight, rateLimitingBehaviour, ct).ConfigureAwait(false);
|
||||
return await CheckGuardsAsync(guards, logger, itemId, type, definition, host, apiKey, requestWeight, rateLimitingBehaviour, keySuffix, ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
// Apply the weight on each guard
|
||||
foreach (var guard in guards)
|
||||
{
|
||||
var result = guard.ApplyWeight(type, definition, host, apiKey, requestWeight);
|
||||
var result = guard.ApplyWeight(type, definition, host, apiKey, requestWeight, keySuffix);
|
||||
if (result.IsApplied)
|
||||
{
|
||||
RateLimitUpdated?.Invoke(new RateLimitUpdateEvent(itemId, _name, guard.Description, result.Current, result.Limit, result.Period));
|
||||
|
||||
@@ -5,6 +5,7 @@ using CryptoExchange.Net.Objects.Sockets;
|
||||
using CryptoExchange.Net.RateLimiting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using System;
|
||||
using System.Buffers;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
@@ -32,6 +33,7 @@ namespace CryptoExchange.Net.Sockets
|
||||
|
||||
internal static int _lastStreamId;
|
||||
private static readonly object _streamIdLock = new();
|
||||
private static readonly ArrayPool<byte> _receiveBufferPool = ArrayPool<byte>.Shared;
|
||||
|
||||
private readonly AsyncResetEvent _sendEvent;
|
||||
private readonly ConcurrentQueue<SendItem> _sendBuffer;
|
||||
@@ -48,8 +50,9 @@ namespace CryptoExchange.Net.Sockets
|
||||
private DateTime _lastReconnectTime;
|
||||
private readonly string _baseAddress;
|
||||
private int _reconnectAttempt;
|
||||
private readonly int _receiveBufferSize;
|
||||
|
||||
private const int _receiveBufferSize = 1048576;
|
||||
private const int _defaultReceiveBufferSize = 1048576;
|
||||
private const int _sendBufferSize = 4096;
|
||||
|
||||
/// <summary>
|
||||
@@ -149,6 +152,7 @@ namespace CryptoExchange.Net.Sockets
|
||||
_sendBuffer = new ConcurrentQueue<SendItem>();
|
||||
_ctsSource = new CancellationTokenSource();
|
||||
_receivedMessagesLock = new object();
|
||||
_receiveBufferSize = websocketParameters.ReceiveBufferSize ?? _defaultReceiveBufferSize;
|
||||
|
||||
_closeSem = new SemaphoreSlim(1, 1);
|
||||
_socket = CreateSocket();
|
||||
@@ -219,7 +223,7 @@ namespace CryptoExchange.Net.Sockets
|
||||
if (Parameters.RateLimiter != null)
|
||||
{
|
||||
var definition = new RequestDefinition(Uri.AbsolutePath, HttpMethod.Get) { ConnectionId = Id };
|
||||
var limitResult = await Parameters.RateLimiter.ProcessAsync(_logger, Id, RateLimitItemType.Connection, definition, _baseAddress, null, 1, Parameters.RateLimitingBehavior, _ctsSource.Token).ConfigureAwait(false);
|
||||
var limitResult = await Parameters.RateLimiter.ProcessAsync(_logger, Id, RateLimitItemType.Connection, definition, _baseAddress, null, 1, Parameters.RateLimitingBehavior, null, _ctsSource.Token).ConfigureAwait(false);
|
||||
if (!limitResult)
|
||||
return new CallResult(new ClientRateLimitError("Connection limit reached"));
|
||||
}
|
||||
@@ -508,7 +512,7 @@ namespace CryptoExchange.Net.Sockets
|
||||
{
|
||||
try
|
||||
{
|
||||
var limitResult = await Parameters.RateLimiter.ProcessAsync(_logger, data.Id, RateLimitItemType.Request, requestDefinition, _baseAddress, null, data.Weight, Parameters.RateLimitingBehavior, _ctsSource.Token).ConfigureAwait(false);
|
||||
var limitResult = await Parameters.RateLimiter.ProcessAsync(_logger, data.Id, RateLimitItemType.Request, requestDefinition, _baseAddress, null, data.Weight, Parameters.RateLimitingBehavior, null, _ctsSource.Token).ConfigureAwait(false);
|
||||
if (!limitResult)
|
||||
{
|
||||
await (OnRequestRateLimited?.Invoke(data.Id) ?? Task.CompletedTask).ConfigureAwait(false);
|
||||
@@ -566,8 +570,8 @@ namespace CryptoExchange.Net.Sockets
|
||||
/// <returns></returns>
|
||||
private async Task ReceiveLoopAsync()
|
||||
{
|
||||
var buffer = new ArraySegment<byte>(new byte[_receiveBufferSize]);
|
||||
var received = 0;
|
||||
byte[] rentedBuffer = _receiveBufferPool.Rent(_receiveBufferSize);
|
||||
var buffer = new ArraySegment<byte>(rentedBuffer);
|
||||
try
|
||||
{
|
||||
while (true)
|
||||
@@ -583,13 +587,12 @@ namespace CryptoExchange.Net.Sockets
|
||||
try
|
||||
{
|
||||
receiveResult = await _socket.ReceiveAsync(buffer, _ctsSource.Token).ConfigureAwait(false);
|
||||
received += receiveResult.Count;
|
||||
lock (_receivedMessagesLock)
|
||||
_receivedMessages.Add(new ReceiveItem(DateTime.UtcNow, receiveResult.Count));
|
||||
}
|
||||
catch (OperationCanceledException ex)
|
||||
{
|
||||
if (ex.InnerException?.InnerException?.Message.Equals("The WebSocket didn't receive a Pong frame in response to a Ping frame within the configured KeepAliveTimeout.") == true)
|
||||
if (ex.InnerException?.InnerException?.Message.Contains("KeepAliveTimeout") == true)
|
||||
{
|
||||
// Specific case that the websocket connection got closed because of a ping frame timeout
|
||||
// Unfortunately doesn't seem to be a nicer way to catch
|
||||
@@ -649,7 +652,7 @@ namespace CryptoExchange.Net.Sockets
|
||||
{
|
||||
// Received a complete message and it's not multi part
|
||||
_logger.SocketReceivedSingleMessage(Id, receiveResult.Count);
|
||||
await ProcessData(receiveResult.MessageType, new ReadOnlyMemory<byte>(buffer.Array, buffer.Offset, receiveResult.Count)).ConfigureAwait(false);
|
||||
await ProcessData(receiveResult.MessageType, new ReadOnlyMemory<byte>(buffer.Array!, buffer.Offset, receiveResult.Count)).ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
{
|
||||
@@ -705,6 +708,7 @@ namespace CryptoExchange.Net.Sockets
|
||||
}
|
||||
finally
|
||||
{
|
||||
_receiveBufferPool.Return(rentedBuffer, true);
|
||||
_logger.SocketReceiveLoopFinished(Id);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,25 +5,26 @@
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Binance.Net" Version="10.16.2" />
|
||||
<PackageReference Include="Bitfinex.Net" Version="8.0.2" />
|
||||
<PackageReference Include="BitMart.Net" Version="1.12.2" />
|
||||
<PackageReference Include="Bybit.Net" Version="4.0.2" />
|
||||
<PackageReference Include="CoinEx.Net" Version="7.14.0" />
|
||||
<PackageReference Include="CryptoCom.Net" Version="1.5.1" />
|
||||
<PackageReference Include="GateIo.Net" Version="1.18.0" />
|
||||
<PackageReference Include="HyperLiquid.Net" Version="1.0.1" />
|
||||
<PackageReference Include="JK.BingX.Net" Version="1.20.1" />
|
||||
<PackageReference Include="JK.Bitget.Net" Version="1.20.0" />
|
||||
<PackageReference Include="JK.Mexc.Net" Version="2.0.0" />
|
||||
<PackageReference Include="JK.OKX.Net" Version="2.14.2" />
|
||||
<PackageReference Include="JKorf.BitMEX.Net" Version="1.0.0" />
|
||||
<PackageReference Include="JKorf.Coinbase.Net" Version="1.7.2" />
|
||||
<PackageReference Include="JKorf.HTX.Net" Version="6.8.2" />
|
||||
<PackageReference Include="KrakenExchange.Net" Version="5.6.0" />
|
||||
<PackageReference Include="Kucoin.Net" Version="5.23.5" />
|
||||
<PackageReference Include="Binance.Net" Version="10.18.0" />
|
||||
<PackageReference Include="Bitfinex.Net" Version="8.1.1" />
|
||||
<PackageReference Include="BitMart.Net" Version="1.14.0" />
|
||||
<PackageReference Include="Bybit.Net" Version="4.3.2" />
|
||||
<PackageReference Include="CoinEx.Net" Version="8.0.1" />
|
||||
<PackageReference Include="CryptoCom.Net" Version="1.6.0" />
|
||||
<PackageReference Include="DeepCoin.Net" Version="1.0.0" />
|
||||
<PackageReference Include="GateIo.Net" Version="1.20.1" />
|
||||
<PackageReference Include="HyperLiquid.Net" Version="1.1.0" />
|
||||
<PackageReference Include="JK.BingX.Net" Version="1.21.0" />
|
||||
<PackageReference Include="JK.Bitget.Net" Version="1.21.0" />
|
||||
<PackageReference Include="JK.Mexc.Net" Version="2.1.0" />
|
||||
<PackageReference Include="JK.OKX.Net" Version="2.15.0" />
|
||||
<PackageReference Include="JKorf.BitMEX.Net" Version="1.1.0" />
|
||||
<PackageReference Include="JKorf.Coinbase.Net" Version="1.8.1" />
|
||||
<PackageReference Include="JKorf.HTX.Net" Version="6.9.0" />
|
||||
<PackageReference Include="KrakenExchange.Net" Version="5.7.1" />
|
||||
<PackageReference Include="Kucoin.Net" Version="6.0.0" />
|
||||
<PackageReference Include="Serilog.AspNetCore" Version="9.0.0" />
|
||||
<PackageReference Include="WhiteBit.Net" Version="1.3.2" />
|
||||
<PackageReference Include="WhiteBit.Net" Version="1.4.0" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
||||
@@ -9,6 +9,7 @@
|
||||
@inject ICoinbaseRestClient coinbaseClient
|
||||
@inject ICoinExRestClient coinexClient
|
||||
@inject ICryptoComRestClient cryptocomClient
|
||||
@inject IDeepCoinRestClient deepCoinClient
|
||||
@inject IGateIoRestClient gateioClient
|
||||
@inject IHTXRestClient htxClient
|
||||
@inject IHyperLiquidRestClient hyperLiquidClient
|
||||
@@ -37,8 +38,9 @@
|
||||
var bitmexTask = bitmexClient.ExchangeApi.ExchangeData.GetSymbolsAsync("XBT_USDT");
|
||||
var bybitTask = bybitClient.V5Api.ExchangeData.GetSpotTickersAsync("BTCUSDT");
|
||||
var coinbaseTask = coinbaseClient.AdvancedTradeApi.ExchangeData.GetSymbolAsync("BTC-USDT");
|
||||
var coinexTask = coinexClient.SpotApi.ExchangeData.GetTickerAsync("BTCUSDT");
|
||||
var coinexTask = coinexClient.SpotApiV2.ExchangeData.GetTickersAsync(["BTCUSDT"]);
|
||||
var cryptocomTask = cryptocomClient.ExchangeApi.ExchangeData.GetTickersAsync("BTC_USDT");
|
||||
var deepCoinTask = deepCoinClient.ExchangeApi.ExchangeData.GetTickersAsync(DeepCoin.Net.Enums.SymbolType.Spot);
|
||||
var gateioTask = gateioClient.SpotApi.ExchangeData.GetTickersAsync("BTC_USDT");
|
||||
var htxTask = htxClient.SpotApi.ExchangeData.GetTickerAsync("btcusdt");
|
||||
var hyperLiquidTask = hyperLiquidClient.FuturesApi.ExchangeData.GetExchangeInfoAndTickersAsync(); // HyperLiquid does not have BTC spot trading
|
||||
@@ -48,7 +50,7 @@
|
||||
var okxTask = okxClient.UnifiedApi.ExchangeData.GetTickerAsync("BTCUSDT");
|
||||
var whitebitTask = whitebitClient.V4Api.ExchangeData.GetTickersAsync();
|
||||
|
||||
await Task.WhenAll(binanceTask, bingXTask, bitfinexTask, bitgetTask, bitmartTask, bybitTask, coinexTask, gateioTask, htxTask, krakenTask, kucoinTask, mexcTask, okxTask);
|
||||
await Task.WhenAll(binanceTask, bingXTask, bitfinexTask, bitgetTask, bitmartTask, bybitTask, coinexTask, deepCoinTask, gateioTask, htxTask, krakenTask, kucoinTask, mexcTask, okxTask);
|
||||
|
||||
if (binanceTask.Result.Success)
|
||||
_prices.Add("Binance", binanceTask.Result.Data.LastPrice);
|
||||
@@ -75,11 +77,18 @@
|
||||
_prices.Add("Coinbase", coinbaseTask.Result.Data.LastPrice ?? 0);
|
||||
|
||||
if (coinexTask.Result.Success)
|
||||
_prices.Add("CoinEx", coinexTask.Result.Data.Ticker.LastPrice);
|
||||
_prices.Add("CoinEx", coinexTask.Result.Data.Single().LastPrice);
|
||||
|
||||
if (cryptocomTask.Result.Success)
|
||||
_prices.Add("CryptoCom", cryptocomTask.Result.Data.First().LastPrice ?? 0);
|
||||
|
||||
if (deepCoinTask.Result.Success)
|
||||
{
|
||||
// DeepCoin API doesn't offer an endpoint to filter for a specific ticker, so we have to filter client side
|
||||
var tickers = deepCoinTask.Result.Data;
|
||||
_prices.Add("DeepCoin", tickers.Single(x => x.Symbol == "BTC-USDT").LastPrice ?? 0);
|
||||
}
|
||||
|
||||
if (gateioTask.Result.Success)
|
||||
_prices.Add("GateIo", gateioTask.Result.Data.First().LastPrice);
|
||||
|
||||
|
||||
@@ -9,6 +9,7 @@
|
||||
@inject ICoinbaseSocketClient coinbaseSocketClient
|
||||
@inject ICoinExSocketClient coinExSocketClient
|
||||
@inject ICryptoComSocketClient cryptocomSocketClient
|
||||
@inject IDeepCoinSocketClient deepCoinSocketClient
|
||||
@inject IGateIoSocketClient gateioSocketClient
|
||||
@inject IHTXSocketClient htxSocketClient
|
||||
@inject IHyperLiquidSocketClient hyperLiquidSocketClient
|
||||
@@ -47,6 +48,7 @@
|
||||
coinExSocketClient.SpotApiV2.SubscribeToTickerUpdatesAsync(["ETHBTC"], data => UpdateData("CoinEx", data.Data.First().LastPrice)),
|
||||
coinbaseSocketClient.AdvancedTradeApi.SubscribeToTickerUpdatesAsync("ETH-BTC", data => UpdateData("Coinbase", data.Data.LastPrice ?? 0)),
|
||||
cryptocomSocketClient.ExchangeApi.SubscribeToTickerUpdatesAsync("ETH_BTC", data => UpdateData("CryptoCom", data.Data.LastPrice ?? 0)),
|
||||
deepCoinSocketClient.ExchangeApi.SubscribeToSymbolUpdatesAsync("ETH-BTC", data => UpdateData("DeepCoin", data.Data.LastPrice ?? 0)),
|
||||
gateioSocketClient.SpotApi.SubscribeToTickerUpdatesAsync("ETH_BTC", data => UpdateData("GateIo", data.Data.LastPrice)),
|
||||
htxSocketClient.SpotApi.SubscribeToTickerUpdatesAsync("ethbtc", data => UpdateData("HTX", data.Data.ClosePrice ?? 0)),
|
||||
// HyperLiquid doesn't support the ETH/BTC pair
|
||||
|
||||
@@ -12,6 +12,7 @@
|
||||
@using Coinbase.Net.Interfaces
|
||||
@using CryptoExchange.Net.Interfaces
|
||||
@using CryptoCom.Net.Interfaces
|
||||
@using DeepCoin.Net.Interfaces
|
||||
@using GateIo.Net.Interfaces
|
||||
@using HTX.Net.Interfaces
|
||||
@using HyperLiquid.Net.Interfaces
|
||||
@@ -31,6 +32,7 @@
|
||||
@inject ICoinbaseOrderBookFactory coinbaseFactory
|
||||
@inject ICoinExOrderBookFactory coinExFactory
|
||||
@inject ICryptoComOrderBookFactory cryptocomFactory
|
||||
@inject IDeepCoinOrderBookFactory deepCoinFactory
|
||||
@inject IGateIoOrderBookFactory gateioFactory
|
||||
@inject IHTXOrderBookFactory htxFactory
|
||||
@inject IHyperLiquidOrderBookFactory hyperLiquidFactory
|
||||
@@ -83,6 +85,7 @@
|
||||
{ "CoinEx", coinExFactory.CreateSpot("ETHBTC") },
|
||||
{ "CryptoCom", cryptocomFactory.Create("ETH_BTC") },
|
||||
{ "GateIo", gateioFactory.CreateSpot("ETH_BTC") },
|
||||
{ "DeepCoin", deepCoinFactory.Create("ETH-BTC") },
|
||||
{ "HTX", htxFactory.CreateSpot("ethbtc") },
|
||||
// HyperLiquid does not support the ETH/BTC pair
|
||||
//{ "HyperLiquid", hyperLiquidFactory.Create("ETH/BTC") },
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
@page "/SpotClient"
|
||||
@using CryptoExchange.Net.SharedApis
|
||||
@using System.Diagnostics
|
||||
@inject IEnumerable<ISpotTickerRestClient> restClients
|
||||
|
||||
<h3>ETH-BTC prices:</h3>
|
||||
@@ -20,6 +21,8 @@
|
||||
{
|
||||
if (ticker.Success)
|
||||
_prices.Add(ticker.Exchange, ticker.Data.LastPrice);
|
||||
else
|
||||
Debug.WriteLine($"{ticker.Exchange} failed: {ticker.Error}");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -14,6 +14,7 @@
|
||||
@using CryptoCom.Net.Interfaces
|
||||
@using CryptoExchange.Net.SharedApis
|
||||
@using CryptoExchange.Net.Trackers.Trades
|
||||
@using DeepCoin.Net.Interfaces
|
||||
@using GateIo.Net.Interfaces
|
||||
@using HTX.Net.Interfaces
|
||||
@using HyperLiquid.Net.Interfaces
|
||||
@@ -33,6 +34,7 @@
|
||||
@inject ICoinbaseTrackerFactory coinbaseFactory
|
||||
@inject ICoinExTrackerFactory coinExFactory
|
||||
@inject ICryptoComTrackerFactory cryptocomFactory
|
||||
@inject IDeepCoinTrackerFactory deepCoinFactory
|
||||
@inject IGateIoTrackerFactory gateioFactory
|
||||
@inject IHTXTrackerFactory htxFactory
|
||||
@inject IHyperLiquidTrackerFactory hyperLiquidFactory
|
||||
@@ -77,6 +79,7 @@
|
||||
{ coinbaseFactory.CreateTradeTracker(usdtSymbol, period: TimeSpan.FromMinutes(5)) },
|
||||
{ coinExFactory.CreateTradeTracker(usdtSymbol, period: TimeSpan.FromMinutes(5)) },
|
||||
{ cryptocomFactory.CreateTradeTracker(usdtSymbol, period: TimeSpan.FromMinutes(5)) },
|
||||
{ deepCoinFactory.CreateTradeTracker(usdtSymbol, period: TimeSpan.FromMinutes(5)) },
|
||||
{ gateioFactory.CreateTradeTracker(usdtSymbol, period: TimeSpan.FromMinutes(5)) },
|
||||
{ htxFactory.CreateTradeTracker(usdtSymbol, period: TimeSpan.FromMinutes(5)) },
|
||||
// HyperLiquid doesn't support spot pair, but does have a futures BTC/USDC pair
|
||||
|
||||
@@ -45,6 +45,7 @@ namespace BlazorClient
|
||||
services.AddCoinbase();
|
||||
services.AddCoinEx();
|
||||
services.AddCryptoCom();
|
||||
services.AddDeepCoin();
|
||||
services.AddGateIo();
|
||||
services.AddHyperLiquid();
|
||||
services.AddHTX();
|
||||
|
||||
@@ -18,6 +18,7 @@
|
||||
@using Coinbase.Net.Interfaces.Clients;
|
||||
@using CoinEx.Net.Interfaces.Clients;
|
||||
@using CryptoCom.Net.Interfaces.Clients;
|
||||
@using DeepCoin.Net.Interfaces.Clients;
|
||||
@using GateIo.Net.Interfaces.Clients;
|
||||
@using HTX.Net.Interfaces.Clients;
|
||||
@using HyperLiquid.Net.Interfaces.Clients;
|
||||
|
||||
@@ -23,6 +23,7 @@ The following API's are directly supported. Note that there are 3rd party implem
|
||||
|CoinEx|[JKorf/CoinEx.Net](https://github.com/JKorf/CoinEx.Net)|[](https://www.nuget.org/packages/CoinEx.Net)|
|
||||
|CoinGecko|[JKorf/CoinGecko.Net](https://github.com/JKorf/CoinGecko.Net)|[](https://www.nuget.org/packages/CoinGecko.Net)|
|
||||
|Crypto.com|[JKorf/CryptoCom.Net](https://github.com/JKorf/CryptoCom.Net)|[](https://www.nuget.org/packages/CryptoCom.Net)|
|
||||
|DeepCoin|[JKorf/DeepCoin.Net](https://github.com/JKorf/DeepCoin.Net)|[](https://www.nuget.org/packages/DeepCoin.Net)|
|
||||
|Gate.io|[JKorf/GateIo.Net](https://github.com/JKorf/GateIo.Net)|[](https://www.nuget.org/packages/GateIo.Net)|
|
||||
|HTX|[JKorf/HTX.Net](https://github.com/JKorf/HTX.Net)|[](https://www.nuget.org/packages/JKorf.HTX.Net)|
|
||||
|HyperLiquid|[JKorf/HyperLiquid.Net](https://github.com/JKorf/HyperLiquid.Net)|[](https://www.nuget.org/packages/HyperLiquid.Net)|
|
||||
@@ -51,6 +52,7 @@ When creating an account on new exchanges please consider using a referral link
|
||||
|Coinbase|[https://advanced.coinbase.com/join/T6H54H8](https://advanced.coinbase.com/join/T6H54H8)|
|
||||
|CoinEx|[https://www.coinex.com/register?refer_code=hd6gn](https://www.coinex.com/register?refer_code=hd6gn)|
|
||||
|Crypto.com|[https://crypto.com/exch/26ge92xbkn](https://crypto.com/exch/26ge92xbkn)|
|
||||
|DeepCoin|[https://s.deepcoin.com/jddhfca)|
|
||||
|HTX|[https://www.htx.com/invite/en-us/1f?invite_code=fxp9](https://www.htx.com/invite/en-us/1f?invite_code=fxp9)|
|
||||
|HyperLiquid|[https://app.hyperliquid.xyz/join/JKORF](https://app.hyperliquid.xyz/join/JKORF)|
|
||||
|Kucoin|[https://www.kucoin.com/r/rf/QBS4FPED](https://www.kucoin.com/r/rf/QBS4FPED)|
|
||||
|
||||
+3
-1
@@ -157,6 +157,7 @@
|
||||
<tr><td>CoinEx</td><td><a href="https://github.com/JKorf/CoinEx.Net">JKorf/CoinEx.Net</a></td><td><a href="https://www.nuget.org/packages/CoinEx.Net" target="_blank"><img src="https://img.shields.io/nuget/v/CoinEx.net.svg?style=flat-square" /></a></td></tr>
|
||||
<tr><td>CoinGecko</td><td><a href="https://github.com/JKorf/CoinGecko.Net">JKorf/CoinGecko.Net</a></td><td><a href="https://www.nuget.org/packages/CoinGecko.Net" target="_blank"><img src="https://img.shields.io/nuget/v/CoinGecko.net.svg?style=flat-square" /></a></td></tr>
|
||||
<tr><td>Crypto.com</td><td><a href="https://github.com/JKorf/CryptoCom.Net">JKorf/CryptoCom.Net</a></td><td><a href="https://www.nuget.org/packages/CryptoCom.Net" target="_blank"><img src="https://img.shields.io/nuget/v/CryptoCom.net.svg?style=flat-square" /></a></td></tr>
|
||||
<tr><td>DeepCoin</td><td><a href="https://github.com/JKorf/DeepCoin.Net">JKorf/DeepCoin.Net</a></td><td><a href="https://www.nuget.org/packages/DeepCoin.Net" target="_blank"><img src="https://img.shields.io/nuget/v/DeepCoin.net.svg?style=flat-square" /></a></td></tr>
|
||||
<tr><td>Gate.io</td><td><a href="https://github.com/JKorf/GateIo.Net">JKorf/GateIo.Net</a></td><td><a href="https://www.nuget.org/packages/GateIo.Net" target="_blank"><img src="https://img.shields.io/nuget/v/GateIo.net.svg?style=flat-square" /></a></td></tr>
|
||||
<tr><td>HTX</td><td><a href="https://github.com/JKorf/HTX.Net">JKorf/HTX.Net</a></td><td><a href="https://www.nuget.org/packages/JKorf.HTX.Net" target="_blank"><img src="https://img.shields.io/nuget/v/JKorf.HTX.net.svg?style=flat-square" /></a></td></tr>
|
||||
<tr><td>HyperLiquid</td><td><a href="https://github.com/JKorf/HyperLiquid.Net">JKorf/HyperLiquid.Net</a></td><td><a href="https://www.nuget.org/packages/HyperLiquid.Net" target="_blank"><img src="https://img.shields.io/nuget/v/HyperLiquid.net.svg?style=flat-square" /></a></td></tr>
|
||||
@@ -204,7 +205,8 @@
|
||||
<tr><td>Bybit</td><td>https://partner.bybit.com/b/jkorf</td></tr>
|
||||
<tr><td>Coinbase</td><td>https://advanced.coinbase.com/join/T6H54H8</td></tr>
|
||||
<tr><td>CoinEx</td><td>https://www.coinex.com/register?refer_code=hd6gn</td></tr>
|
||||
<tr><td>Crypto.com</td><td>https://crypto.com/exch/26ge92xbkn</td></tr>
|
||||
<tr><td>Crypto.com</td><td>https://crypto.com/exch/26ge92xbkn</td></tr>
|
||||
<tr><td>DeepCoin</td><td>https://s.deepcoin.com/jddhfca</td></tr>
|
||||
<tr><td>HTX</td><td>https://www.htx.com/invite/en-us/1f?invite_code=fxp9</td></tr>
|
||||
<tr><td>HyperLiquid</td><td>https://app.hyperliquid.xyz/join/JKORF</td></tr>
|
||||
<tr><td>Kucoin</td><td>https://www.kucoin.com/r/rf/QBS4FPED</td></tr>
|
||||
|
||||
Reference in New Issue
Block a user