mirror of
https://github.com/JKorf/CryptoExchange.Net.git
synced 2026-08-14 18:02:58 +00:00
Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| bdf7a07c6f | |||
| a8ffe90bf2 | |||
| 3372b9eb44 | |||
| df25221960 | |||
| 7c67a014f5 |
@@ -304,55 +304,9 @@ namespace CryptoExchange.Net.Clients
|
|||||||
return new CallResult<UpdateSubscription>(new ServerError(new ErrorInfo(ErrorType.WebsocketPaused, "Socket is paused")));
|
return new CallResult<UpdateSubscription>(new ServerError(new ErrorInfo(ErrorType.WebsocketPaused, "Socket is paused")));
|
||||||
}
|
}
|
||||||
|
|
||||||
void HandleSubscriptionComplete(bool success, object? response)
|
var subscribeResult = await socketConnection.TrySubscribeAsync(subscription, true, ct).ConfigureAwait(false);
|
||||||
{
|
if (!subscribeResult)
|
||||||
if (!success)
|
return new CallResult<UpdateSubscription>(subscribeResult.Error!);
|
||||||
return;
|
|
||||||
|
|
||||||
subscription.HandleSubQueryResponse(socketConnection, response);
|
|
||||||
subscription.Status = SubscriptionStatus.Subscribed;
|
|
||||||
if (ct != default)
|
|
||||||
{
|
|
||||||
subscription.CancellationTokenRegistration = ct.Register(async () =>
|
|
||||||
{
|
|
||||||
_logger.CancellationTokenSetClosingSubscription(socketConnection.SocketId, subscription.Id);
|
|
||||||
await socketConnection.CloseAsync(subscription).ConfigureAwait(false);
|
|
||||||
}, false);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
subscription.Status = SubscriptionStatus.Subscribing;
|
|
||||||
var subQuery = subscription.CreateSubscriptionQuery(socketConnection);
|
|
||||||
if (subQuery != null)
|
|
||||||
{
|
|
||||||
subQuery.OnComplete = () => HandleSubscriptionComplete(subQuery.Result?.Success ?? false, subQuery.Response);
|
|
||||||
|
|
||||||
// Send the request and wait for answer
|
|
||||||
var subResult = await socketConnection.SendAndWaitQueryAsync(subQuery, ct).ConfigureAwait(false);
|
|
||||||
if (!subResult)
|
|
||||||
{
|
|
||||||
var isTimeout = subResult.Error is CancellationRequestedError;
|
|
||||||
if (isTimeout && subscription.Status == SubscriptionStatus.Subscribed)
|
|
||||||
{
|
|
||||||
// No response received, but the subscription did receive updates. We'll assume success
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
_logger.FailedToSubscribe(socketConnection.SocketId, subResult.Error?.ToString());
|
|
||||||
// If this was a server process error we still might need to send an unsubscribe to prevent messages coming in later
|
|
||||||
subscription.Status = SubscriptionStatus.Pending;
|
|
||||||
await socketConnection.CloseAsync(subscription).ConfigureAwait(false);
|
|
||||||
return new CallResult<UpdateSubscription>(subResult.Error!);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if (!subQuery.ExpectsResponse)
|
|
||||||
HandleSubscriptionComplete(true, null);
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
HandleSubscriptionComplete(true, null);
|
|
||||||
}
|
|
||||||
|
|
||||||
_logger.SubscriptionCompletedSuccessfully(socketConnection.SocketId, subscription.Id);
|
_logger.SubscriptionCompletedSuccessfully(socketConnection.SocketId, subscription.Id);
|
||||||
return new CallResult<UpdateSubscription>(new UpdateSubscription(socketConnection, subscription));
|
return new CallResult<UpdateSubscription>(new UpdateSubscription(socketConnection, subscription));
|
||||||
|
|||||||
@@ -6,9 +6,9 @@
|
|||||||
<PackageId>CryptoExchange.Net</PackageId>
|
<PackageId>CryptoExchange.Net</PackageId>
|
||||||
<Authors>JKorf</Authors>
|
<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>
|
<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.5.1</PackageVersion>
|
<PackageVersion>10.5.2</PackageVersion>
|
||||||
<AssemblyVersion>10.5.1</AssemblyVersion>
|
<AssemblyVersion>10.5.2</AssemblyVersion>
|
||||||
<FileVersion>10.5.1</FileVersion>
|
<FileVersion>10.5.2</FileVersion>
|
||||||
<PackageRequireLicenseAcceptance>false</PackageRequireLicenseAcceptance>
|
<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>
|
<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>
|
<RepositoryType>git</RepositoryType>
|
||||||
|
|||||||
@@ -672,6 +672,8 @@ namespace CryptoExchange.Net.Sockets.Default
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (!processed)
|
if (!processed)
|
||||||
|
{
|
||||||
|
if (!ApiClient.HandleUnhandledMessage(this, typeIdentifier, data))
|
||||||
{
|
{
|
||||||
lock (_listenersLock)
|
lock (_listenersLock)
|
||||||
{
|
{
|
||||||
@@ -683,6 +685,7 @@ namespace CryptoExchange.Net.Sockets.Default
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Connect the websocket
|
/// Connect the websocket
|
||||||
@@ -1082,40 +1085,8 @@ namespace CryptoExchange.Net.Sockets.Default
|
|||||||
var taskList = new List<Task<CallResult>>();
|
var taskList = new List<Task<CallResult>>();
|
||||||
foreach (var subscription in subList)
|
foreach (var subscription in subList)
|
||||||
{
|
{
|
||||||
subscription.ConnectionInvocations = 0;
|
var subscribeTask = TrySubscribeAsync(subscription, false, default);
|
||||||
if (!subscription.Active)
|
taskList.Add(subscribeTask);
|
||||||
// Can be closed during resubscribing
|
|
||||||
continue;
|
|
||||||
|
|
||||||
subscription.Status = SubscriptionStatus.Subscribing;
|
|
||||||
var result = await ApiClient.RevitalizeRequestAsync(subscription).ConfigureAwait(false);
|
|
||||||
if (!result)
|
|
||||||
{
|
|
||||||
_logger.FailedRequestRevitalization(SocketId, result.Error?.ToString());
|
|
||||||
subscription.Status = SubscriptionStatus.Pending;
|
|
||||||
return result;
|
|
||||||
}
|
|
||||||
|
|
||||||
var subQuery = subscription.CreateSubscriptionQuery(this);
|
|
||||||
if (subQuery == null)
|
|
||||||
{
|
|
||||||
subscription.Status = SubscriptionStatus.Subscribed;
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
subQuery.OnComplete = () =>
|
|
||||||
{
|
|
||||||
subscription.Status = subQuery.Result!.Success ? SubscriptionStatus.Subscribed : SubscriptionStatus.Pending;
|
|
||||||
subscription.HandleSubQueryResponse(this, subQuery.Response);
|
|
||||||
};
|
|
||||||
|
|
||||||
taskList.Add(SendAndWaitQueryAsync(subQuery));
|
|
||||||
|
|
||||||
if (!subQuery.ExpectsResponse)
|
|
||||||
{
|
|
||||||
// If there won't be an answer we can immediately set this
|
|
||||||
subscription.Status = SubscriptionStatus.Subscribed;
|
|
||||||
subscription.HandleSubQueryResponse(this, null);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
await Task.WhenAll(taskList).ConfigureAwait(false);
|
await Task.WhenAll(taskList).ConfigureAwait(false);
|
||||||
@@ -1132,6 +1103,61 @@ namespace CryptoExchange.Net.Sockets.Default
|
|||||||
return CallResult.SuccessResult;
|
return CallResult.SuccessResult;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
protected internal async Task<CallResult> TrySubscribeAsync(Subscription subscription, bool newSubscription, CancellationToken subCancelToken)
|
||||||
|
{
|
||||||
|
subscription.ConnectionInvocations = 0;
|
||||||
|
|
||||||
|
if (!newSubscription)
|
||||||
|
{
|
||||||
|
if (!subscription.Active)
|
||||||
|
// Can be closed during resubscribing
|
||||||
|
return CallResult.SuccessResult;
|
||||||
|
|
||||||
|
var result = await ApiClient.RevitalizeRequestAsync(subscription).ConfigureAwait(false);
|
||||||
|
if (!result)
|
||||||
|
{
|
||||||
|
_logger.FailedRequestRevitalization(SocketId, result.Error?.ToString());
|
||||||
|
subscription.Status = SubscriptionStatus.Pending;
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
subscription.Status = SubscriptionStatus.Subscribing;
|
||||||
|
var subQuery = subscription.CreateSubscriptionQuery(this);
|
||||||
|
if (subQuery == null)
|
||||||
|
{
|
||||||
|
// No sub query, so successful
|
||||||
|
subscription.Status = SubscriptionStatus.Subscribed;
|
||||||
|
return CallResult.SuccessResult;
|
||||||
|
}
|
||||||
|
|
||||||
|
subQuery.OnComplete = () =>
|
||||||
|
{
|
||||||
|
subscription.Status = subQuery.Result!.Success ? SubscriptionStatus.Subscribed : SubscriptionStatus.Pending;
|
||||||
|
subscription.HandleSubQueryResponse(this, subQuery.Response);
|
||||||
|
if (newSubscription && subQuery.Result.Success && subCancelToken != default)
|
||||||
|
{
|
||||||
|
subscription.CancellationTokenRegistration = subCancelToken.Register(async () =>
|
||||||
|
{
|
||||||
|
_logger.CancellationTokenSetClosingSubscription(SocketId, subscription.Id);
|
||||||
|
await CloseAsync(subscription).ConfigureAwait(false);
|
||||||
|
}, false);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
var subQueryResult = await SendAndWaitQueryAsync(subQuery).ConfigureAwait(false);
|
||||||
|
if (!subQueryResult)
|
||||||
|
{
|
||||||
|
_logger.FailedToSubscribe(SocketId, subQueryResult.Error?.ToString());
|
||||||
|
// If this was a server process error or timeout we still send an unsubscribe to prevent messages coming in later
|
||||||
|
if (newSubscription)
|
||||||
|
await CloseAsync(subscription).ConfigureAwait(false);
|
||||||
|
return new CallResult<UpdateSubscription>(subQueryResult.Error!);
|
||||||
|
}
|
||||||
|
|
||||||
|
return subQueryResult;
|
||||||
|
}
|
||||||
|
|
||||||
internal async Task UnsubscribeAsync(Subscription subscription)
|
internal async Task UnsubscribeAsync(Subscription subscription)
|
||||||
{
|
{
|
||||||
var unsubscribeRequest = subscription.CreateUnsubscriptionQuery(this);
|
var unsubscribeRequest = subscription.CreateUnsubscriptionQuery(this);
|
||||||
|
|||||||
@@ -174,6 +174,14 @@ namespace CryptoExchange.Net.Sockets.Default
|
|||||||
{
|
{
|
||||||
ConnectionInvocations++;
|
ConnectionInvocations++;
|
||||||
TotalInvocations++;
|
TotalInvocations++;
|
||||||
|
if (SubscriptionQuery != null && !SubscriptionQuery.Completed && SubscriptionQuery.TimeoutBehavior == TimeoutBehavior.Succeed)
|
||||||
|
{
|
||||||
|
// The subscription query is one where it is successful if there is no error returned
|
||||||
|
// Since we've received a data update for the subscription we can assume the subscribe query was successful
|
||||||
|
// Call timeout to complete
|
||||||
|
SubscriptionQuery.Timeout();
|
||||||
|
}
|
||||||
|
|
||||||
return route.Handle(connection, receiveTime, originalData, data);
|
return route.Handle(connection, receiveTime, originalData, data);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -127,8 +127,8 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
Completed = true;
|
|
||||||
Result = CallResult.SuccessResult;
|
Result = CallResult.SuccessResult;
|
||||||
|
Completed = true;
|
||||||
_event.Set();
|
_event.Set();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -216,12 +216,12 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
if (Completed)
|
if (Completed)
|
||||||
return;
|
return;
|
||||||
|
|
||||||
Completed = true;
|
|
||||||
if (TimeoutBehavior == TimeoutBehavior.Fail)
|
if (TimeoutBehavior == TimeoutBehavior.Fail)
|
||||||
Result = new CallResult<THandlerResponse>(new TimeoutError());
|
Result = new CallResult<THandlerResponse>(new TimeoutError());
|
||||||
else
|
else
|
||||||
Result = new CallResult<THandlerResponse>(default, null, default);
|
Result = new CallResult<THandlerResponse>(default, null, default);
|
||||||
|
|
||||||
|
Completed = true;
|
||||||
_event.Set();
|
_event.Set();
|
||||||
OnComplete?.Invoke();
|
OnComplete?.Invoke();
|
||||||
}
|
}
|
||||||
@@ -234,6 +234,7 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
|
|
||||||
Result = new CallResult<THandlerResponse>(error);
|
Result = new CallResult<THandlerResponse>(error);
|
||||||
Completed = true;
|
Completed = true;
|
||||||
|
|
||||||
_event.Set();
|
_event.Set();
|
||||||
OnComplete?.Invoke();
|
OnComplete?.Invoke();
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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).
|
Alternatively, sponsor me on Github using [Github Sponsors](https://github.com/sponsors/JKorf).
|
||||||
|
|
||||||
## Release notes
|
## Release notes
|
||||||
|
* Version 10.5.2 - 10 Feb 2026
|
||||||
|
* Added check for subscribe queries with TimeoutBehavior.Success to complete when subscription has received update
|
||||||
|
* Added call to ApiClient.HandleUnhandledMessage when no websocket message processor is found based on topic to allow additional processing
|
||||||
|
* Combined websocket connection subscribe and re-subscribe logic
|
||||||
|
* Set websocket query completed after setting Result
|
||||||
|
|
||||||
* Version 10.5.1 - 10 Feb 2026
|
* Version 10.5.1 - 10 Feb 2026
|
||||||
* Fixed trading mode selection for futures listen key methods in FuturesUserDataTracker
|
* Fixed trading mode selection for futures listen key methods in FuturesUserDataTracker
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user