1
0
mirror of https://github.com/JKorf/CryptoExchange.Net.git synced 2026-10-14 07:01:46 +00:00

Compare commits

..

5 Commits

Author SHA1 Message Date
Jkorf f079e6f0f4 Updated to version 13.2.0 2026-10-13 11:41:36 +02:00
Jkorf c6647f6815 Fixed DelayAfterConnect applying on failed connection attemps on HighPerf connections 2026-10-13 10:38:21 +02:00
Jkorf 5135047097 Fix for duplicate subscription coverage 2026-10-13 10:37:49 +02:00
Jkorf 4f25a6e7d6 Added processor parameter for some methods in SocketApiClient to allow specific overriding behavior 2026-10-13 10:08:16 +02:00
Jan Korf a0f3dcaa4c WebSocket covered subscriptions
Updated WebSocket duplicate subscription handling
Fixed incorrect status keeping of WebSocket subscriptions which don't expect a response
2026-10-12 13:19:44 +02:00
9 changed files with 155 additions and 29 deletions
+71 -10
View File
@@ -286,7 +286,7 @@ namespace CryptoExchange.Net.Clients
while (true)
{
// Get a new or existing socket connection
var socketResult = await GetSocketConnection(url, subscription.Authenticated, false, ct, subscription.Topic, subscription.IndividualSubscriptionCount).ConfigureAwait(false);
var socketResult = await GetSocketConnection(url, subscription.Authenticated, false, ct, subscription, subscription.Topic, subscription.IndividualSubscriptionCount).ConfigureAwait(false);
if (!socketResult.Success)
return WebSocketResult.Fail<UpdateSubscription>(Exchange, socketResult.Error);
@@ -309,7 +309,7 @@ namespace CryptoExchange.Net.Clients
var needsConnecting = !socketConnection.Connected;
var connectResult = await ConnectIfNeededAsync(socketConnection, subscription.Authenticated, ct).ConfigureAwait(false);
var connectResult = await ConnectIfNeededAsync(socketConnection, subscription.Authenticated, ct, subscription).ConfigureAwait(false);
if (!connectResult.Success)
return WebSocketResult.Fail<UpdateSubscription>(Exchange, connectResult.Error!);
@@ -482,7 +482,7 @@ namespace CryptoExchange.Net.Clients
await semaphoreSlim.WaitAsync().ConfigureAwait(false);
try
{
var socketResult = await GetSocketConnection(url, query.Authenticated, true, ct).ConfigureAwait(false);
var socketResult = await GetSocketConnection(url, query.Authenticated, true, ct, processor: query).ConfigureAwait(false);
if (!socketResult.Success)
return QueryResult.Fail<THandlerResponse>(Exchange, socketResult.Error);
@@ -495,7 +495,7 @@ namespace CryptoExchange.Net.Clients
released = true;
}
var connectResult = await ConnectIfNeededAsync(socketConnection, query.Authenticated, ct).ConfigureAwait(false);
var connectResult = await ConnectIfNeededAsync(socketConnection, query.Authenticated, ct, query).ConfigureAwait(false);
if (!connectResult.Success)
return QueryResult.Fail<THandlerResponse>(Exchange, connectResult.Error!);
}
@@ -524,7 +524,18 @@ namespace CryptoExchange.Net.Clients
/// <param name="authenticated">Whether the socket should authenticated</param>
/// <param name="ct">Cancellation token</param>
/// <returns></returns>
protected virtual async Task<CallResult> ConnectIfNeededAsync(ISocketConnection socket, bool authenticated, CancellationToken ct)
protected virtual Task<CallResult> ConnectIfNeededAsync(ISocketConnection socket, bool authenticated, CancellationToken ct)
=> ConnectIfNeededAsync(socket, authenticated, ct, null);
/// <summary>
/// Checks if a socket needs to be connected and does so if needed. Also authenticates on the socket if needed
/// </summary>
/// <param name="socket">The connection to check</param>
/// <param name="authenticated">Whether the socket should authenticated</param>
/// <param name="processor">The message processor requesting a connection</param>
/// <param name="ct">Cancellation token</param>
/// <returns></returns>
protected virtual async Task<CallResult> ConnectIfNeededAsync(ISocketConnection socket, bool authenticated, CancellationToken ct, IMessageProcessor? processor)
{
if (socket.Connected)
return CallResult.Ok();
@@ -539,7 +550,7 @@ namespace CryptoExchange.Net.Clients
if (socket is not SocketConnection sc)
throw new InvalidOperationException("HighPerfSocketConnection not supported for authentication");
var result = await AuthenticateSocketAsync(sc).ConfigureAwait(false);
var result = await AuthenticateSocketAsync(sc, processor).ConfigureAwait(false);
if (!result.Success)
await socket.CloseAsync().ConfigureAwait(false);
@@ -551,13 +562,22 @@ namespace CryptoExchange.Net.Clients
/// </summary>
/// <param name="socket">Socket to authenticate</param>
/// <returns></returns>
public virtual async Task<CallResult> AuthenticateSocketAsync(SocketConnection socket)
public virtual Task<CallResult> AuthenticateSocketAsync(SocketConnection socket)
=> AuthenticateSocketAsync(socket, null);
/// <summary>
/// Authenticate a socket connection
/// </summary>
/// <param name="socket">Socket to authenticate</param>
/// <param name="processor">The message processor requesting authentication</param>
/// <returns></returns>
public virtual async Task<CallResult> AuthenticateSocketAsync(SocketConnection socket, IMessageProcessor? processor)
{
if (GetAuthenticationProvider() == null)
return CallResult.Fail(new NoApiCredentialsError());
_logger.AttemptingToAuthenticate(socket.SocketId);
var authRequest = await GetAuthenticationRequestAsync(socket).ConfigureAwait(false);
var authRequest = await GetAuthenticationRequestAsync(socket, processor).ConfigureAwait(false);
if (authRequest != null)
{
var result = await socket.SendAndWaitQueryAsync(authRequest).ConfigureAwait(false);
@@ -578,13 +598,21 @@ namespace CryptoExchange.Net.Clients
return CallResult.Ok();
}
/// <summary>
/// Should return the request which can be used to authenticate a socket connection
/// </summary>
/// <returns></returns>
protected internal virtual Task<Query?> GetAuthenticationRequestAsync(SocketConnection connection) =>
protected internal virtual Task<Query?> GetAuthenticationRequestAsync(SocketConnection connection) =>
Task.FromResult(GetAuthenticationProvider()!.GetAuthenticationQuery(this, connection));
/// <summary>
/// Should return the request which can be used to authenticate a socket connection
/// </summary>
/// <returns></returns>
protected internal virtual Task<Query?> GetAuthenticationRequestAsync(SocketConnection connection, IMessageProcessor? processor) =>
GetAuthenticationRequestAsync(connection);
/// <summary>
/// Adds a system subscription. Used for example to reply to ping requests
/// </summary>
@@ -642,6 +670,18 @@ namespace CryptoExchange.Net.Clients
&& (AllowTopicsOnTheSameConnection || !connection.Topics.Contains(topic));
}
/// <summary>
/// Whether the connection can be used for a new subscription or query with the provided parameters
/// </summary>
/// <param name="connection">The connection to check</param>
/// <param name="address">The address set by the request</param>
/// <param name="authenticated">Whether the request needs an authenticated connection</param>
/// <param name="processor">The processor requesting a connection</param>
/// <param name="topic">Topic of the request</param>
/// <returns>True if connection can be used</returns>
protected virtual bool ConnectionCanBeUsedFor(SocketConnection connection, string address, bool authenticated, IMessageProcessor? processor, string? topic = null)
=> ConnectionCanBeUsedFor(connection, address, authenticated, topic);
/// <summary>
/// Gets a connection for a new subscription or query. Can be an existing if there are open position or a new one.
/// </summary>
@@ -652,15 +692,36 @@ namespace CryptoExchange.Net.Clients
/// <param name="topic">The subscription topic, can be provided when multiple of the same topics are not allowed on a connection</param>
/// <param name="individualSubscriptionCount">The number of individual subscriptions in this subscribe request</param>
/// <returns></returns>
protected virtual Task<CallResult<SocketConnection>> GetSocketConnection(
string address,
bool authenticated,
bool dedicatedRequestConnection,
CancellationToken ct,
string? topic = null,
int individualSubscriptionCount = 1)
=> GetSocketConnection(address, authenticated, dedicatedRequestConnection, ct, null, topic, individualSubscriptionCount);
/// <summary>
/// Gets a connection for a new subscription or query. Can be an existing if there are open position or a new one.
/// </summary>
/// <param name="address">The address the socket is for</param>
/// <param name="authenticated">Whether the socket should be authenticated</param>
/// <param name="dedicatedRequestConnection">Whether a dedicated request connection should be returned</param>
/// <param name="ct">Cancellation token</param>
/// <param name="topic">The subscription topic, can be provided when multiple of the same topics are not allowed on a connection</param>
/// <param name="individualSubscriptionCount">The number of individual subscriptions in this subscribe request</param>
/// <param name="processor">The processor requesting a connection</param>
/// <returns></returns>
protected virtual async Task<CallResult<SocketConnection>> GetSocketConnection(
string address,
bool authenticated,
bool dedicatedRequestConnection,
CancellationToken ct,
IMessageProcessor? processor,
string? topic = null,
int individualSubscriptionCount = 1)
{
var socketQuery = _socketConnections.Where(s => ConnectionCanBeUsedFor(s.Value, address, authenticated, topic)).Select(x => x.Value); // Don't ToList this so the query is executed again when called
var socketQuery = _socketConnections.Where(s => ConnectionCanBeUsedFor(s.Value, address, authenticated, processor, topic)).Select(x => x.Value); // Don't ToList this so the query is executed again when called
// If all current socket connections are reconnecting or resubscribing wait for that to finish as we can probably use the existing connection
var delayStart = DateTime.UtcNow;
+3 -3
View File
@@ -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>13.1.0</PackageVersion>
<AssemblyVersion>13.1.0</AssemblyVersion>
<FileVersion>13.1.0</FileVersion>
<PackageVersion>13.2.0</PackageVersion>
<AssemblyVersion>13.2.0</AssemblyVersion>
<FileVersion>13.2.0</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>
@@ -23,6 +23,10 @@ namespace CryptoExchange.Net.SharedApis
/// Get user funding history capability.
/// </summary>
public static SharedRestCapability<IGetUserFundingHistory, IGetUserFundingHistoryRest> GetUserFundingHistory { get; } = new();
/// <summary>
/// Subscribe to funding info updates capability.
/// </summary>
public static SharedCapabilityReference<ISubscribeFundingInfoSocket> SubscribeFundingInfo { get; } = new();
}
}
}
@@ -702,7 +702,7 @@ namespace CryptoExchange.Net.Sockets.Default
if (subscription.CancellationTokenRegistration.HasValue)
subscription.CancellationTokenRegistration.Value.Dispose();
bool anyDuplicateSubscription = _listeners.OfType<Subscription>().Any(x => x != subscription && x.MessageRouter.Routes.All(l => subscription.MessageRouter.ContainsCheck(l)));
bool anyDuplicateSubscription = SubscriptionIsCovered(subscription);
bool shouldCloseConnection = _listeners.OfType<Subscription>().All(r => !r.UserSubscription || r.Status == SubscriptionStatus.Closing || r.Status == SubscriptionStatus.Closed) && !DedicatedRequestConnection.IsDedicatedRequestConnection;
if (!anyDuplicateSubscription)
@@ -1069,6 +1069,32 @@ namespace CryptoExchange.Net.Sockets.Default
return CallResult.Ok();
}
/// <summary>
/// Whether there is currently a subscription covering all the routes the subscription lists
/// </summary>
protected bool SubscriptionIsCovered(Subscription subscription)
=> _listeners.OfType<Subscription>().Any(
x => x != subscription
&& x.Active
&& subscription.MessageRouter.Routes.All(l => x.MessageRouter.ContainsCheck(l)));
/// <summary>
/// Whether there is a subscription earlier in the list covering all the routes the subscription lists
/// </summary>
protected bool IsCoveredByEarlierSubscription(Subscription subscription)
{
foreach (var other in _listeners.OfType<Subscription>())
{
if (other == subscription)
return false;
if (other.Active && subscription.MessageRouter.Routes.All(x => other.MessageRouter.ContainsCheck(x)))
return true;
}
return false;
}
/// <summary>
/// Try to subscribe a new subscription by sending the subscribe query and wait for the result as needed
/// </summary>
@@ -1103,11 +1129,11 @@ namespace CryptoExchange.Net.Sockets.Default
return WebSocketResult.Ok(ApiClient.Exchange, SocketId, default, 0, ConnectionUriString);
}
var subCompleteHandler = () =>
var subCompleteHandler = (bool success) =>
{
subscription.Status = subQuery.Success ? SubscriptionStatus.Subscribed : SubscriptionStatus.Pending;
subscription.Status = success ? SubscriptionStatus.Subscribed : SubscriptionStatus.Pending;
subscription.HandleSubQueryResponse(this, subQuery.Response);
if (newSubscription && subQuery.Success)
if (newSubscription && success)
{
subscription.TokenExpired += HandleTokenExpired;
if (subCancelToken != default)
@@ -1122,6 +1148,14 @@ namespace CryptoExchange.Net.Sockets.Default
};
subQuery.OnComplete = subCompleteHandler;
// For duplicate subscription don't send another subscribe request but just handle it internally
// Unsubscribe also doesn't unsubscribe when there is still a duplicate subscription running
if (IsCoveredByEarlierSubscription(subscription))
{
subQuery.Complete();
return WebSocketResult.Ok(ApiClient.Exchange, SocketId, default, subQuery.Id, ConnectionUriString);
}
var subQueryResult = await SendAndWaitQueryAsync(subQuery).ConfigureAwait(false);
if (!subQueryResult.Success)
{
@@ -1132,9 +1166,6 @@ namespace CryptoExchange.Net.Sockets.Default
return WebSocketResult.Fail(ApiClient.Exchange, subQueryResult.Error!);
}
if (!subQuery.ExpectsResponse)
subCompleteHandler();
return subQueryResult;
}
@@ -198,8 +198,13 @@ namespace CryptoExchange.Net.Sockets.HighPerf
{
var result = await _socket.ConnectAsync(ct).ConfigureAwait(false);
if (result.Success)
{
_processTask = ProcessAsync(_cts.Token);
if ((ApiClient.ApiOptions.DelayAfterConnect ?? ApiClient.ClientOptions.DelayAfterConnect) != TimeSpan.Zero)
await Task.Delay(ApiClient.ApiOptions.DelayAfterConnect ?? ApiClient.ClientOptions.DelayAfterConnect).ConfigureAwait(false);
}
return result;
}
+22 -6
View File
@@ -124,7 +124,7 @@ namespace CryptoExchange.Net.Sockets
/// <summary>
/// On complete callback
/// </summary>
public Action? OnComplete { get; set; }
public Action<bool>? OnComplete { get; set; }
/// <inheritdoc />
public event Action? OnMessageRouterUpdated;
@@ -157,8 +157,7 @@ namespace CryptoExchange.Net.Sockets
}
else
{
Completed = true;
_event.Set();
Complete();
}
}
@@ -175,6 +174,11 @@ namespace CryptoExchange.Net.Sockets
/// </summary>
public abstract void Timeout();
/// <summary>
/// Mark request as successfully completed
/// </summary>
public abstract void Complete();
/// <summary>
/// Mark request as failed
/// </summary>
@@ -248,7 +252,7 @@ namespace CryptoExchange.Net.Sockets
{
Completed = true;
_event.Set();
OnComplete?.Invoke();
OnComplete?.Invoke(Success);
}
return handled;
@@ -267,7 +271,19 @@ namespace CryptoExchange.Net.Sockets
Completed = true;
_event.Set();
OnComplete?.Invoke();
OnComplete?.Invoke(Success);
}
/// <inheritdoc />
public override void Complete()
{
if (Completed)
return;
Result = CallResult<THandlerResponse>.Ok(default!);
Completed = true;
_event.Set();
OnComplete?.Invoke(Success);
}
/// <inheritdoc />
@@ -280,7 +296,7 @@ namespace CryptoExchange.Net.Sockets
Completed = true;
_event.Set();
OnComplete?.Invoke();
OnComplete?.Invoke(Success);
}
}
}
+9
View File
@@ -127,6 +127,15 @@ Various:
* PlatformInfo now required support environment names in the constructor
## Release notes
* Version 13.2.0 - 13 Oct 2026
* Added Shared API ISubscribeFundingInfoSocket
* Added DelayAfterConnect to SocketApiOptions
* Added processor parameter for some methods in SocketApiClient to allow specific overriding behavior
* Updated WebSocket duplicate subscription handling
* Updated Shared APIs ISubscribeIncrementalOrderBook
* Fixed incorrect status keeping of WebSocket subscriptions which don't expect a response
* Fixed DelayAfterConnect not being respected on reconnection
* Version 13.1.0 - 28 Sep 2026
* Added UnsubscribeAllAsync method to Shared APIs ISharedSubscription and ISharedApiClientBase
* Added WaitAllAsync extension method for waiting on an IAsyncEnumerable to finish
+2 -2
View File
@@ -1,8 +1,8 @@
# CryptoExchange.Net — full AI context
Library: CryptoExchange.Net
Library version: 13.1.0
CryptoExchange.Net package version: 13.1.0
Library version: 13.2.0
CryptoExchange.Net package version: 13.2.0
Language: C#/.NET
Targets: netstandard2.0, netstandard2.1, net8.0, net9.0, net10.0
+1 -1
View File
@@ -2,7 +2,7 @@
> Base C#/.NET library used by 28+ cryptocurrency exchange clients. It standardizes REST, WebSocket, authentication, rate limiting, results, and exchange-agnostic Shared APIs.
CryptoExchange.Net is not an exchange client by itself. Install exchange-specific packages such as Binance.Net, Bybit.Net, JK.OKX.Net, Kraken.Net, or Coinbase.Net, or install `CryptoClients.Net` for the bundle. Current release: 13.1.0. Targets netstandard2.0, netstandard2.1, net8.0, net9.0, and net10.0; Native AOT is supported.
CryptoExchange.Net is not an exchange client by itself. Install exchange-specific packages such as Binance.Net, Bybit.Net, JK.OKX.Net, Kraken.Net, or Coinbase.Net, or install `CryptoClients.Net` for the bundle. Current release: 13.2.0. Targets netstandard2.0, netstandard2.1, net8.0, net9.0, and net10.0; Native AOT is supported.
## Shared API V2