mirror of
https://github.com/JKorf/CryptoExchange.Net.git
synced 2026-08-12 08:53:01 +00:00
Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Jkorf
|
33c0fb26a7 | ||
|
Jkorf
|
dabbb5c868 | ||
|
Jkorf
|
204bda8622 | ||
|
Jkorf
|
78e3523a4f | ||
|
Jkorf
|
89a73747b0 | ||
|
Jkorf
|
02b70398b3 |
@@ -6,9 +6,9 @@
|
||||
<PackageId>CryptoExchange.Net</PackageId>
|
||||
<Authors>JKorf</Authors>
|
||||
<Description>CryptoExchange.Net is a base library which is used to implement different cryptocurrency (exchange) API's. It provides a standardized way of implementing different API's, which results in a very similar experience for users of the API implementations.</Description>
|
||||
<PackageVersion>10.7.0</PackageVersion>
|
||||
<AssemblyVersion>10.7.0</AssemblyVersion>
|
||||
<FileVersion>10.7.0</FileVersion>
|
||||
<PackageVersion>10.7.2</PackageVersion>
|
||||
<AssemblyVersion>10.7.2</AssemblyVersion>
|
||||
<FileVersion>10.7.2</FileVersion>
|
||||
<PackageRequireLicenseAcceptance>false</PackageRequireLicenseAcceptance>
|
||||
<PackageTags>OKX;OKX.Net;Mexc;Mexc.Net;Kucoin;Kucoin.Net;Kraken;Kraken.Net;Huobi;Huobi.Net;CoinEx;CoinEx.Net;Bybit;Bybit.Net;Bitget;Bitget.Net;Bitfinex;Bitfinex.Net;Binance;Binance.Net;CryptoCurrency;CryptoCurrency Exchange;CryptoExchange.Net</PackageTags>
|
||||
<RepositoryType>git</RepositoryType>
|
||||
|
||||
@@ -8,7 +8,9 @@ using CryptoExchange.Net.Sockets.Default.Interfaces;
|
||||
using CryptoExchange.Net.Sockets.Interfaces;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using System;
|
||||
using System.Collections;
|
||||
using System.Collections.Generic;
|
||||
using System.Collections.ObjectModel;
|
||||
using System.Diagnostics;
|
||||
using System.Linq;
|
||||
using System.Net.WebSockets;
|
||||
@@ -123,15 +125,7 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
{
|
||||
get
|
||||
{
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterReadLock();
|
||||
return _listeners.OfType<Subscription>().Count(h => h.UserSubscription);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitReadLock();
|
||||
}
|
||||
return _listeners.OfType<Subscription>().Count(h => h.UserSubscription);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -142,16 +136,7 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
{
|
||||
get
|
||||
{
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterReadLock();
|
||||
return _listeners.OfType<Subscription>().Where(h => h.UserSubscription).ToArray();
|
||||
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitReadLock();
|
||||
}
|
||||
return _listeners.OfType<Subscription>().Where(h => h.UserSubscription).ToArray();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -255,15 +240,7 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
{
|
||||
get
|
||||
{
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterReadLock();
|
||||
return _listeners.OfType<Subscription>().Select(x => x.Topic).Where(t => t != null).ToArray()!;
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitReadLock();
|
||||
}
|
||||
return _listeners.OfType<Subscription>().Select(x => x.Topic).Where(t => t != null).ToArray()!;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -274,22 +251,18 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
{
|
||||
get
|
||||
{
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterReadLock();
|
||||
return _listeners.OfType<Query>().Where(x => !x.Completed).Count();
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitReadLock();
|
||||
}
|
||||
return _listeners.OfType<Query>().Where(x => !x.Completed).Count();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private bool _pausedActivity;
|
||||
private readonly ReaderWriterLockSlim _listenersLock = new ReaderWriterLockSlim();
|
||||
private readonly List<IMessageProcessor> _listeners;
|
||||
#if NET9_0_OR_GREATER
|
||||
private readonly Lock _listenersLock = new Lock();
|
||||
#else
|
||||
private readonly object _listenersLock = new object();
|
||||
#endif
|
||||
private ReadOnlyCollection<IMessageProcessor> _listeners;
|
||||
private readonly ILogger _logger;
|
||||
private SocketStatus _status;
|
||||
|
||||
@@ -338,7 +311,7 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
_socket.OnError += HandleErrorAsync;
|
||||
_socket.GetReconnectionUrl = GetReconnectionUrlAsync;
|
||||
|
||||
_listeners = new List<IMessageProcessor>();
|
||||
_listeners = new ReadOnlyCollection<IMessageProcessor>([]);
|
||||
|
||||
_serializer = apiClient.CreateSerializer();
|
||||
}
|
||||
@@ -365,25 +338,17 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
if (ApiClient._socketConnections.ContainsKey(SocketId))
|
||||
ApiClient._socketConnections.TryRemove(SocketId, out _);
|
||||
|
||||
try
|
||||
foreach (var subscription in _listeners.OfType<Subscription>().Where(l => l.UserSubscription && !l.IsClosingConnection))
|
||||
{
|
||||
_listenersLock.EnterWriteLock();
|
||||
foreach (var subscription in _listeners.OfType<Subscription>().Where(l => l.UserSubscription && !l.IsClosingConnection))
|
||||
{
|
||||
subscription.IsClosingConnection = true;
|
||||
subscription.Reset();
|
||||
}
|
||||
subscription.IsClosingConnection = true;
|
||||
subscription.Reset();
|
||||
}
|
||||
|
||||
foreach (var query in _listeners.OfType<Query>().ToList())
|
||||
{
|
||||
query.Fail(new WebError("Connection interrupted"));
|
||||
_listeners.Remove(query);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitWriteLock();
|
||||
}
|
||||
var queryList = _listeners.OfType<Query>().ToList();
|
||||
foreach (var query in queryList)
|
||||
query.Fail(new WebError("Connection interrupted"));
|
||||
|
||||
RemoveMessageProcessors(queryList);
|
||||
|
||||
_ = Task.Run(() => ConnectionClosed?.Invoke());
|
||||
return Task.CompletedTask;
|
||||
@@ -399,22 +364,14 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
Authenticated = false;
|
||||
_lastSequenceNumber = 0;
|
||||
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterWriteLock();
|
||||
foreach (var subscription in _listeners.OfType<Subscription>().Where(l => l.UserSubscription))
|
||||
subscription.Reset();
|
||||
foreach (var subscription in _listeners.OfType<Subscription>().Where(l => l.UserSubscription))
|
||||
subscription.Reset();
|
||||
|
||||
foreach (var query in _listeners.OfType<Query>().ToList())
|
||||
{
|
||||
query.Fail(new WebError("Connection interrupted"));
|
||||
_listeners.Remove(query);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitWriteLock();
|
||||
}
|
||||
var queryList = _listeners.OfType<Query>().ToList();
|
||||
foreach (var query in queryList)
|
||||
query.Fail(new WebError("Connection interrupted"));
|
||||
|
||||
RemoveMessageProcessors(queryList);
|
||||
|
||||
_ = Task.Run(() => ConnectionLost?.Invoke());
|
||||
return Task.CompletedTask;
|
||||
@@ -436,19 +393,11 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
{
|
||||
Status = SocketStatus.Resubscribing;
|
||||
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterWriteLock();
|
||||
foreach (var query in _listeners.OfType<Query>().ToList())
|
||||
{
|
||||
query.Fail(new WebError("Connection interrupted"));
|
||||
_listeners.Remove(query);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitWriteLock();
|
||||
}
|
||||
var queryList = _listeners.OfType<Query>().ToList();
|
||||
foreach (var query in queryList)
|
||||
query.Fail(new WebError("Connection interrupted"));
|
||||
|
||||
RemoveMessageProcessors(queryList);
|
||||
|
||||
// Can't wait for this as it would cause a deadlock
|
||||
_ = Task.Run(async () =>
|
||||
@@ -503,17 +452,7 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
/// <returns></returns>
|
||||
protected virtual Task HandleRequestRateLimitedAsync(int requestId)
|
||||
{
|
||||
Query? query;
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterReadLock();
|
||||
query = _listeners.OfType<Query>().FirstOrDefault(x => x.Id == requestId);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitReadLock();
|
||||
}
|
||||
|
||||
var query = _listeners.OfType<Query>().FirstOrDefault(x => x.Id == requestId);
|
||||
if (query == null)
|
||||
return Task.CompletedTask;
|
||||
|
||||
@@ -537,17 +476,7 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
/// <param name="requestId">Id of the request sent</param>
|
||||
protected virtual Task HandleRequestSentAsync(int requestId)
|
||||
{
|
||||
Query? query;
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterReadLock();
|
||||
query = _listeners.OfType<Query>().FirstOrDefault(x => x.Id == requestId);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitReadLock();
|
||||
}
|
||||
|
||||
var query = _listeners.OfType<Query>().FirstOrDefault(x => x.Id == requestId);
|
||||
if (query == null)
|
||||
return Task.CompletedTask;
|
||||
|
||||
@@ -593,27 +522,19 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
}
|
||||
|
||||
Type? deserializationType = null;
|
||||
try
|
||||
foreach (var subscription in _listeners)
|
||||
{
|
||||
_listenersLock.EnterReadLock();
|
||||
foreach (var subscription in _listeners)
|
||||
foreach (var route in subscription.MessageRouter.Routes)
|
||||
{
|
||||
foreach (var route in subscription.MessageRouter.Routes)
|
||||
{
|
||||
if (!route.TypeIdentifier.Equals(typeIdentifier, StringComparison.Ordinal))
|
||||
continue;
|
||||
if (!route.TypeIdentifier.Equals(typeIdentifier, StringComparison.Ordinal))
|
||||
continue;
|
||||
|
||||
deserializationType = route.DeserializationType;
|
||||
break;
|
||||
}
|
||||
|
||||
if (deserializationType != null)
|
||||
break;
|
||||
deserializationType = route.DeserializationType;
|
||||
break;
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitReadLock();
|
||||
|
||||
if (deserializationType != null)
|
||||
break;
|
||||
}
|
||||
|
||||
if (deserializationType == null)
|
||||
@@ -660,89 +581,69 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
var topicFilter = messageConverter.GetTopicFilter(result);
|
||||
|
||||
bool processed = false;
|
||||
try
|
||||
foreach (var processor in _listeners)
|
||||
{
|
||||
_listenersLock.EnterReadLock();
|
||||
var currentCount = _listeners.Count;
|
||||
for(var i = 0; i < _listeners.Count; i++)
|
||||
bool isQuery = false;
|
||||
Query? query = null;
|
||||
if (processor is Query cquery)
|
||||
{
|
||||
if (_listeners.Count != currentCount)
|
||||
{
|
||||
// Possible a query added or removed. If added it's not a problem, if removed it is
|
||||
if (_listeners.Count < currentCount)
|
||||
throw new Exception("Listeners list adjusted, can't continue processing");
|
||||
}
|
||||
|
||||
var processor = _listeners[i];
|
||||
bool isQuery = false;
|
||||
Query? query = null;
|
||||
if (processor is Query cquery)
|
||||
{
|
||||
isQuery = true;
|
||||
query = cquery;
|
||||
}
|
||||
|
||||
var complete = false;
|
||||
|
||||
foreach (var route in processor.MessageRouter.Routes)
|
||||
{
|
||||
if (route.TypeIdentifier != typeIdentifier)
|
||||
continue;
|
||||
|
||||
// Forward message rules:
|
||||
// | Message Topic | Route Topic Filter | Topics Match | Forward | Description
|
||||
// | N | N | - | Y | No topic filter applied
|
||||
// | N | Y | - | N | Route only listens to specific topic
|
||||
// | Y | N | - | Y | Route listens to all message regardless of topic
|
||||
// | Y | Y | Y | Y | Route listens to specific message topic
|
||||
// | Y | Y | N | N | Route listens to different topic
|
||||
if (topicFilter == null)
|
||||
{
|
||||
if (route.TopicFilter != null)
|
||||
// No topic on message, but route is filtering on topic
|
||||
continue;
|
||||
}
|
||||
else
|
||||
{
|
||||
if (route.TopicFilter != null && !route.TopicFilter.Equals(topicFilter, StringComparison.Ordinal))
|
||||
// Message has a topic, and the route has a filter for another topic
|
||||
continue;
|
||||
}
|
||||
|
||||
processed = true;
|
||||
|
||||
if (isQuery && query!.Completed)
|
||||
continue;
|
||||
|
||||
processor.Handle(this, receiveTime, originalData, result, route);
|
||||
if (isQuery && !route.MultipleReaders)
|
||||
{
|
||||
complete = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (complete)
|
||||
break;
|
||||
isQuery = true;
|
||||
query = cquery;
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitReadLock();
|
||||
|
||||
var complete = false;
|
||||
|
||||
foreach (var route in processor.MessageRouter.Routes)
|
||||
{
|
||||
if (route.TypeIdentifier != typeIdentifier)
|
||||
continue;
|
||||
|
||||
// Forward message rules:
|
||||
// | Message Topic | Route Topic Filter | Topics Match | Forward | Description
|
||||
// | N | N | - | Y | No topic filter applied
|
||||
// | N | Y | - | N | Route only listens to specific topic
|
||||
// | Y | N | - | Y | Route listens to all message regardless of topic
|
||||
// | Y | Y | Y | Y | Route listens to specific message topic
|
||||
// | Y | Y | N | N | Route listens to different topic
|
||||
if (topicFilter == null)
|
||||
{
|
||||
if (route.TopicFilter != null)
|
||||
// No topic on message, but route is filtering on topic
|
||||
continue;
|
||||
}
|
||||
else
|
||||
{
|
||||
if (route.TopicFilter != null && !route.TopicFilter.Equals(topicFilter, StringComparison.Ordinal))
|
||||
// Message has a topic, and the route has a filter for another topic
|
||||
continue;
|
||||
}
|
||||
|
||||
processed = true;
|
||||
|
||||
if (isQuery && query!.Completed)
|
||||
continue;
|
||||
|
||||
processor.Handle(this, receiveTime, originalData, result, route);
|
||||
if (isQuery && !route.MultipleReaders)
|
||||
{
|
||||
complete = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (complete)
|
||||
break;
|
||||
}
|
||||
|
||||
if (!processed)
|
||||
{
|
||||
if (!ApiClient.HandleUnhandledMessage(this, typeIdentifier, data))
|
||||
{
|
||||
lock (_listenersLock)
|
||||
{
|
||||
_logger.ReceivedMessageNotMatchedToAnyListener(
|
||||
SocketId,
|
||||
typeIdentifier,
|
||||
topicFilter!,
|
||||
string.Join(",", _listeners.Select(x => string.Join(",", x.MessageRouter.Routes.Where(x => x.TypeIdentifier == typeIdentifier).Select(x => x.TopicFilter != null ? string.Join(",", x.TopicFilter) : "[null]")))));
|
||||
}
|
||||
_logger.ReceivedMessageNotMatchedToAnyListener(
|
||||
SocketId,
|
||||
typeIdentifier,
|
||||
topicFilter!,
|
||||
string.Join(",", _listeners.Select(x => string.Join(",", x.MessageRouter.Routes.Where(x => x.TypeIdentifier == typeIdentifier).Select(x => x.TopicFilter != null ? string.Join(",", x.TopicFilter) : "[null]")))));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -787,18 +688,10 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
if (ApiClient._socketConnections.ContainsKey(SocketId))
|
||||
ApiClient._socketConnections.TryRemove(SocketId, out _);
|
||||
|
||||
try
|
||||
foreach (var subscription in _listeners.OfType<Subscription>())
|
||||
{
|
||||
_listenersLock.EnterReadLock();
|
||||
foreach (var subscription in _listeners.OfType<Subscription>())
|
||||
{
|
||||
if (subscription.CancellationTokenRegistration.HasValue)
|
||||
subscription.CancellationTokenRegistration.Value.Dispose();
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitReadLock();
|
||||
if (subscription.CancellationTokenRegistration.HasValue)
|
||||
subscription.CancellationTokenRegistration.Value.Dispose();
|
||||
}
|
||||
|
||||
await _socket.CloseAsync().ConfigureAwait(false);
|
||||
@@ -828,32 +721,12 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
if (subscription.CancellationTokenRegistration.HasValue)
|
||||
subscription.CancellationTokenRegistration.Value.Dispose();
|
||||
|
||||
bool anyDuplicateSubscription;
|
||||
bool shouldCloseConnection;
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterReadLock();
|
||||
anyDuplicateSubscription = _listeners.OfType<Subscription>().Any(x => x != subscription && x.MessageRouter.Routes.All(l => subscription.MessageRouter.ContainsCheck(l)));
|
||||
shouldCloseConnection = _listeners.OfType<Subscription>().All(r => !r.UserSubscription || r.Status == SubscriptionStatus.Closing || r.Status == SubscriptionStatus.Closed) && !DedicatedRequestConnection.IsDedicatedRequestConnection;
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitReadLock();
|
||||
}
|
||||
bool anyDuplicateSubscription = _listeners.OfType<Subscription>().Any(x => x != subscription && x.MessageRouter.Routes.All(l => subscription.MessageRouter.ContainsCheck(l)));
|
||||
bool shouldCloseConnection = _listeners.OfType<Subscription>().All(r => !r.UserSubscription || r.Status == SubscriptionStatus.Closing || r.Status == SubscriptionStatus.Closed) && !DedicatedRequestConnection.IsDedicatedRequestConnection;
|
||||
|
||||
if (!anyDuplicateSubscription)
|
||||
{
|
||||
bool needUnsub;
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterReadLock();
|
||||
needUnsub = _listeners.Contains(subscription) && !shouldCloseConnection;
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitReadLock();
|
||||
}
|
||||
|
||||
var needUnsub = _listeners.Contains(subscription) && !shouldCloseConnection;
|
||||
if (needUnsub && _socket.IsOpen)
|
||||
await UnsubscribeAsync(subscription).ConfigureAwait(false);
|
||||
}
|
||||
@@ -877,15 +750,7 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
await CloseAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterWriteLock();
|
||||
_listeners.Remove(subscription);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitWriteLock();
|
||||
}
|
||||
RemoveMessageProcessor(subscription);
|
||||
|
||||
subscription.Status = SubscriptionStatus.Closed;
|
||||
}
|
||||
@@ -908,15 +773,8 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
{
|
||||
if (Status != SocketStatus.None && Status != SocketStatus.Connected)
|
||||
return false;
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterWriteLock();
|
||||
_listeners.Add(subscription);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitWriteLock();
|
||||
}
|
||||
|
||||
AddMessageProcessor(subscription);
|
||||
|
||||
if (subscription.UserSubscription)
|
||||
_logger.AddingNewSubscription(SocketId, subscription.Id, UserSubscriptionCount);
|
||||
@@ -929,15 +787,7 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
/// <param name="id"></param>
|
||||
public Subscription? GetSubscription(int id)
|
||||
{
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterReadLock();
|
||||
return _listeners.OfType<Subscription>().SingleOrDefault(s => s.Id == id);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitReadLock();
|
||||
}
|
||||
return _listeners.OfType<Subscription>().SingleOrDefault(s => s.Id == id);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -985,29 +835,12 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
|
||||
private async Task SendAndWaitIntAsync(Query query, CancellationToken ct = default)
|
||||
{
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterWriteLock();
|
||||
_listeners.Add(query);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitWriteLock();
|
||||
}
|
||||
|
||||
AddMessageProcessor(query);
|
||||
var sendResult = await SendAsync(query.Id, query.Request, query.Weight).ConfigureAwait(false);
|
||||
if (!sendResult)
|
||||
{
|
||||
query.Fail(sendResult.Error!);
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterWriteLock();
|
||||
_listeners.Remove(query);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitWriteLock();
|
||||
}
|
||||
RemoveMessageProcessor(query);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -1038,15 +871,7 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
}
|
||||
finally
|
||||
{
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterWriteLock();
|
||||
_listeners.Remove(query);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitWriteLock();
|
||||
}
|
||||
RemoveMessageProcessor(query);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1152,17 +977,7 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
|
||||
if (!DedicatedRequestConnection.IsDedicatedRequestConnection)
|
||||
{
|
||||
bool anySubscriptions;
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterReadLock();
|
||||
anySubscriptions = _listeners.OfType<Subscription>().Any(s => s.UserSubscription);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitReadLock();
|
||||
}
|
||||
|
||||
var anySubscriptions = _listeners.OfType<Subscription>().Any(s => s.UserSubscription);
|
||||
if (!anySubscriptions)
|
||||
{
|
||||
// No need to resubscribe anything
|
||||
@@ -1172,18 +987,8 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
}
|
||||
}
|
||||
|
||||
bool anyAuthenticated;
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterReadLock();
|
||||
anyAuthenticated = _listeners.OfType<Subscription>().Any(s => s.Authenticated)
|
||||
bool anyAuthenticated = _listeners.OfType<Subscription>().Any(s => s.Authenticated)
|
||||
|| DedicatedRequestConnection.IsDedicatedRequestConnection && DedicatedRequestConnection.Authenticated;
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitReadLock();
|
||||
}
|
||||
|
||||
if (anyAuthenticated)
|
||||
{
|
||||
// If we reconnected a authenticated connection we need to re-authenticate
|
||||
@@ -1206,17 +1011,7 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
if (!_socket.IsOpen)
|
||||
return new CallResult(new WebError("Socket not connected"));
|
||||
|
||||
List<Subscription> subList;
|
||||
try
|
||||
{
|
||||
_listenersLock.EnterReadLock();
|
||||
subList = _listeners.OfType<Subscription>().Where(x => x.Active).Skip(batch * batchSize).Take(batchSize).ToList();
|
||||
}
|
||||
finally
|
||||
{
|
||||
_listenersLock.ExitReadLock();
|
||||
}
|
||||
|
||||
var subList = _listeners.OfType<Subscription>().Where(x => x.Active).Skip(batch * batchSize).Take(batchSize).ToList();
|
||||
if (subList.Count == 0)
|
||||
break;
|
||||
|
||||
@@ -1398,6 +1193,37 @@ namespace CryptoExchange.Net.Sockets.Default
|
||||
});
|
||||
}
|
||||
|
||||
private void AddMessageProcessor(IMessageProcessor processor)
|
||||
{
|
||||
lock (_listenersLock)
|
||||
{
|
||||
var updatedList = new List<IMessageProcessor>(_listeners);
|
||||
updatedList.Add(processor);
|
||||
_listeners = updatedList.AsReadOnly();
|
||||
}
|
||||
}
|
||||
|
||||
private void RemoveMessageProcessor(IMessageProcessor processor)
|
||||
{
|
||||
lock (_listenersLock)
|
||||
{
|
||||
var updatedList = new List<IMessageProcessor>(_listeners);
|
||||
updatedList.Remove(processor);
|
||||
_listeners = updatedList.AsReadOnly();
|
||||
}
|
||||
}
|
||||
|
||||
private void RemoveMessageProcessors(IEnumerable<IMessageProcessor> processors)
|
||||
{
|
||||
lock (_listenersLock)
|
||||
{
|
||||
var updatedList = new List<IMessageProcessor>(_listeners);
|
||||
foreach (var processor in processors)
|
||||
updatedList.Remove(processor);
|
||||
_listeners = updatedList.AsReadOnly();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -112,6 +112,8 @@ namespace CryptoExchange.Net.Testing.Implementations
|
||||
|
||||
public async Task ReconnectAsync()
|
||||
{
|
||||
await Task.Delay(1).ConfigureAwait(false);
|
||||
|
||||
if (OnReconnecting != null)
|
||||
await OnReconnecting().ConfigureAwait(false);
|
||||
|
||||
|
||||
@@ -20,6 +20,7 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
|
||||
private readonly ExchangeParameters? _exchangeParameters;
|
||||
private readonly bool _requiresSymbolParameterOpenOrders;
|
||||
private readonly Dictionary<string, int> _openOrderNotReturnedTimes = new();
|
||||
private readonly TimeSpan _pollOverlapPeriod = TimeSpan.FromSeconds(3);
|
||||
|
||||
internal event Func<UpdateSource, SharedUserTrade[], Task>? OnTradeUpdate;
|
||||
|
||||
@@ -355,17 +356,17 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
|
||||
DateTime? fromTime = null;
|
||||
string? source = null;
|
||||
|
||||
// Use the last timestamp we we received data from the websocket as state should be correct at that time. 1 seconds buffer
|
||||
// Use the last timestamp we we received data from the websocket as state should be correct at that time
|
||||
if (_lastDataTimeBeforeDisconnect.HasValue && (fromTime == null || fromTime > _lastDataTimeBeforeDisconnect.Value))
|
||||
{
|
||||
fromTime = _lastDataTimeBeforeDisconnect.Value.AddSeconds(-1);
|
||||
fromTime = _lastDataTimeBeforeDisconnect.Value.Add(-_pollOverlapPeriod);
|
||||
source = "LastDataTimeBeforeDisconnect";
|
||||
}
|
||||
|
||||
// If we've previously polled use that timestamp to request data from
|
||||
if (_lastPollTime.HasValue && (fromTime == null || _lastPollTime.Value > fromTime))
|
||||
{
|
||||
fromTime = _lastPollTime;
|
||||
fromTime = _lastPollTime.Value.Add(-_pollOverlapPeriod);
|
||||
source = "LastPollTime";
|
||||
}
|
||||
|
||||
@@ -378,7 +379,7 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
|
||||
{
|
||||
// Could be improved by only requesting the specific open orders if there are only a few that would be better than trying to request a long
|
||||
// history if the open order is far back
|
||||
fromTime = trackedOrdersMinOpenTime.Value.AddMilliseconds(-1);
|
||||
fromTime = trackedOrdersMinOpenTime.Value.AddSeconds(-1);
|
||||
source = "OpenOrder";
|
||||
}
|
||||
|
||||
@@ -388,7 +389,7 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
|
||||
source = "StartTime";
|
||||
}
|
||||
|
||||
if (DateTime.UtcNow - fromTime < TimeSpan.FromSeconds(1))
|
||||
if (DateTime.UtcNow - fromTime < TimeSpan.FromSeconds(5))
|
||||
{
|
||||
// Set it to at least 5 seconds in the past to prevent issues when local time isn't in sync
|
||||
fromTime = DateTime.UtcNow.AddSeconds(-5);
|
||||
|
||||
@@ -18,6 +18,7 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
|
||||
private readonly IFuturesOrderRestClient _restClient;
|
||||
private readonly IUserTradeSocketClient? _socketClient;
|
||||
private readonly ExchangeParameters? _exchangeParameters;
|
||||
private readonly TimeSpan _pollOverlapPeriod = TimeSpan.FromSeconds(3);
|
||||
|
||||
internal Func<string[]>? GetTrackedOrderIds { get; set; }
|
||||
|
||||
@@ -106,22 +107,22 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
|
||||
|
||||
private DateTime? GetTradesRequestStartTime()
|
||||
{
|
||||
// Determine the timestamp from which we need to check order status
|
||||
// Determine the timestamp from which we need to request trades from
|
||||
// Use the timestamp we last know the correct state of the data
|
||||
DateTime? fromTime = null;
|
||||
string? source = null;
|
||||
|
||||
// Use the last timestamp we we received data from the websocket as state should be correct at that time. 1 seconds buffer
|
||||
// Use the last timestamp we we received data from the websocket as state should be correct at that time.
|
||||
if (_lastDataTimeBeforeDisconnect.HasValue && (fromTime == null || fromTime > _lastDataTimeBeforeDisconnect.Value))
|
||||
{
|
||||
fromTime = _lastDataTimeBeforeDisconnect.Value.AddSeconds(-1);
|
||||
fromTime = _lastDataTimeBeforeDisconnect.Value.Add(-_pollOverlapPeriod);
|
||||
source = "LastDataTimeBeforeDisconnect";
|
||||
}
|
||||
|
||||
// If we've previously polled use that timestamp to request data from
|
||||
if (_lastPollTime.HasValue && (fromTime == null || _lastPollTime.Value > fromTime))
|
||||
{
|
||||
fromTime = _lastPollTime;
|
||||
fromTime = _lastPollTime.Value.Add(-_pollOverlapPeriod);
|
||||
source = "LastPollTime";
|
||||
}
|
||||
|
||||
@@ -132,7 +133,7 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
|
||||
}
|
||||
|
||||
var now = DateTime.UtcNow;
|
||||
if (now - fromTime < TimeSpan.FromSeconds(1))
|
||||
if (now - fromTime < TimeSpan.FromSeconds(5))
|
||||
{
|
||||
// Set it to at least 5 seconds in the past to prevent issues when local time isn't in sync
|
||||
fromTime = DateTime.UtcNow.AddSeconds(-5);
|
||||
|
||||
@@ -20,6 +20,7 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
|
||||
private readonly ExchangeParameters? _exchangeParameters;
|
||||
private readonly bool _requiresSymbolParameterOpenOrders;
|
||||
private readonly Dictionary<string, int> _openOrderNotReturnedTimes = new();
|
||||
private readonly TimeSpan _pollOverlapPeriod = TimeSpan.FromSeconds(3);
|
||||
|
||||
internal event Func<UpdateSource, SharedUserTrade[], Task>? OnTradeUpdate;
|
||||
|
||||
@@ -366,17 +367,17 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
|
||||
DateTime? fromTime = null;
|
||||
string? source = null;
|
||||
|
||||
// Use the last timestamp we we received data from the websocket as state should be correct at that time. 1 seconds buffer
|
||||
// Use the last timestamp we we received data from the websocket as state should be correct at that time.
|
||||
if (_lastDataTimeBeforeDisconnect.HasValue && (fromTime == null || fromTime > _lastDataTimeBeforeDisconnect.Value))
|
||||
{
|
||||
fromTime = _lastDataTimeBeforeDisconnect.Value.AddSeconds(-1);
|
||||
fromTime = _lastDataTimeBeforeDisconnect.Value.Add(-_pollOverlapPeriod);
|
||||
source = "LastDataTimeBeforeDisconnect";
|
||||
}
|
||||
|
||||
// If we've previously polled use that timestamp to request data from
|
||||
if (_lastPollTime.HasValue && (fromTime == null || _lastPollTime.Value > fromTime))
|
||||
{
|
||||
fromTime = _lastPollTime;
|
||||
fromTime = _lastPollTime.Value.Add(-_pollOverlapPeriod);
|
||||
source = "LastPollTime";
|
||||
}
|
||||
|
||||
@@ -389,7 +390,7 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
|
||||
{
|
||||
// Could be improved by only requesting the specific open orders if there are only a few that would be better than trying to request a long
|
||||
// history if the open order is far back
|
||||
fromTime = trackedOrdersMinOpenTime.Value.AddMilliseconds(-1);
|
||||
fromTime = trackedOrdersMinOpenTime.Value.AddSeconds(-1);
|
||||
source = "OpenOrder";
|
||||
}
|
||||
|
||||
@@ -399,7 +400,7 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
|
||||
source = "StartTime";
|
||||
}
|
||||
|
||||
if (DateTime.UtcNow - fromTime < TimeSpan.FromSeconds(1))
|
||||
if (DateTime.UtcNow - fromTime < TimeSpan.FromSeconds(5))
|
||||
{
|
||||
// Set it to at least 5 seconds in the past to prevent issues when local time isn't in sync
|
||||
fromTime = DateTime.UtcNow.AddSeconds(-5);
|
||||
|
||||
@@ -18,6 +18,7 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
|
||||
private readonly ISpotOrderRestClient _restClient;
|
||||
private readonly IUserTradeSocketClient? _socketClient;
|
||||
private readonly ExchangeParameters? _exchangeParameters;
|
||||
private readonly TimeSpan _pollOverlapPeriod = TimeSpan.FromSeconds(3);
|
||||
|
||||
internal Func<string[]>? GetTrackedOrderIds { get; set; }
|
||||
|
||||
@@ -103,22 +104,22 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
|
||||
|
||||
private DateTime? GetTradesRequestStartTime()
|
||||
{
|
||||
// Determine the timestamp from which we need to check order status
|
||||
// Determine the timestamp from which we need to request trades from
|
||||
// Use the timestamp we last know the correct state of the data
|
||||
DateTime? fromTime = null;
|
||||
string? source = null;
|
||||
|
||||
// Use the last timestamp we we received data from the websocket as state should be correct at that time. 1 seconds buffer
|
||||
// Use the last timestamp we we received data from the websocket as state should be correct at that time.
|
||||
if (_lastDataTimeBeforeDisconnect.HasValue && (fromTime == null || fromTime > _lastDataTimeBeforeDisconnect.Value))
|
||||
{
|
||||
fromTime = _lastDataTimeBeforeDisconnect.Value.AddSeconds(-1);
|
||||
fromTime = _lastDataTimeBeforeDisconnect.Value.Add(-_pollOverlapPeriod);
|
||||
source = "LastDataTimeBeforeDisconnect";
|
||||
}
|
||||
|
||||
// If we've previously polled use that timestamp to request data from
|
||||
if (_lastPollTime.HasValue && (fromTime == null || _lastPollTime.Value > fromTime))
|
||||
{
|
||||
fromTime = _lastPollTime;
|
||||
fromTime = _lastPollTime.Value.Add(-_pollOverlapPeriod);
|
||||
source = "LastPollTime";
|
||||
}
|
||||
|
||||
@@ -128,7 +129,7 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
|
||||
source = "StartTime";
|
||||
}
|
||||
|
||||
if (DateTime.UtcNow - fromTime < TimeSpan.FromSeconds(1))
|
||||
if (DateTime.UtcNow - fromTime < TimeSpan.FromSeconds(5))
|
||||
{
|
||||
// Set it to at least 5 seconds in the past to prevent issues when local time isn't in sync
|
||||
fromTime = DateTime.UtcNow.AddSeconds(-5);
|
||||
|
||||
@@ -67,6 +67,12 @@ Make a one time donation in a crypto currency of your choice. If you prefer to d
|
||||
Alternatively, sponsor me on Github using [Github Sponsors](https://github.com/sponsors/JKorf).
|
||||
|
||||
## Release notes
|
||||
* Version 10.7.2 - 02 Mar 2026
|
||||
* Added small overlap in UserDataTracker polling logic to account for API endpoints not immediately having the data available
|
||||
|
||||
* Version 10.7.1 - 25 Feb 2026
|
||||
* Fixed deadlock scenario in websocket connection when subscribe and handling message concurrently
|
||||
|
||||
* Version 10.7.0 - 24 Feb 2026
|
||||
* Added parsing of REST response data up to 128 characters for error responses
|
||||
* Added check for invalid json in JsonSocketMessageHandler
|
||||
|
||||
Reference in New Issue
Block a user