From a0f3dcaa4c3ff9a6eb4b16285559dea63d9d58c1 Mon Sep 17 00:00:00 2001 From: Jan Korf Date: Mon, 12 Oct 2026 13:19:44 +0200 Subject: [PATCH] WebSocket covered subscriptions Updated WebSocket duplicate subscription handling Fixed incorrect status keeping of WebSocket subscriptions which don't expect a response --- .../SharedCapabilities.Funding.cs | 4 ++ .../Sockets/Default/SocketConnection.cs | 45 ++++++++++++++++--- CryptoExchange.Net/Sockets/Query.cs | 28 +++++++++--- 3 files changed, 64 insertions(+), 13 deletions(-) diff --git a/CryptoExchange.Net/SharedApis/CapabilityReferences/SharedCapabilities.Funding.cs b/CryptoExchange.Net/SharedApis/CapabilityReferences/SharedCapabilities.Funding.cs index e429c62a..f7b4d093 100644 --- a/CryptoExchange.Net/SharedApis/CapabilityReferences/SharedCapabilities.Funding.cs +++ b/CryptoExchange.Net/SharedApis/CapabilityReferences/SharedCapabilities.Funding.cs @@ -23,6 +23,10 @@ namespace CryptoExchange.Net.SharedApis /// Get user funding history capability. /// public static SharedRestCapability GetUserFundingHistory { get; } = new(); + /// + /// Subscribe to funding info updates capability. + /// + public static SharedCapabilityReference SubscribeFundingInfo { get; } = new(); } } } diff --git a/CryptoExchange.Net/Sockets/Default/SocketConnection.cs b/CryptoExchange.Net/Sockets/Default/SocketConnection.cs index 99b73736..36e33ad8 100644 --- a/CryptoExchange.Net/Sockets/Default/SocketConnection.cs +++ b/CryptoExchange.Net/Sockets/Default/SocketConnection.cs @@ -702,7 +702,7 @@ namespace CryptoExchange.Net.Sockets.Default if (subscription.CancellationTokenRegistration.HasValue) subscription.CancellationTokenRegistration.Value.Dispose(); - bool anyDuplicateSubscription = _listeners.OfType().Any(x => x != subscription && x.MessageRouter.Routes.All(l => subscription.MessageRouter.ContainsCheck(l))); + bool anyDuplicateSubscription = SubscriptionIsCovered(subscription); bool shouldCloseConnection = _listeners.OfType().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(); } + /// + /// Whether there is currently a subscription covering all the routes the subscription lists + /// + protected bool SubscriptionIsCovered(Subscription subscription) + => _listeners.OfType().Any( + x => x != subscription + && x.Active + && x.MessageRouter.Routes.All(l => subscription.MessageRouter.ContainsCheck(l))); + + /// + /// Whether there is a subscription earlier in the list covering all the routes the subscription lists + /// + protected bool IsCoveredByEarlierSubscription(Subscription subscription) + { + foreach (var other in _listeners.OfType()) + { + if (other == subscription) + return false; + + if (other.Active && other.MessageRouter.Routes.All(x => subscription.MessageRouter.ContainsCheck(x))) + return true; + } + + return false; + } + /// /// Try to subscribe a new subscription by sending the subscribe query and wait for the result as needed /// @@ -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; } diff --git a/CryptoExchange.Net/Sockets/Query.cs b/CryptoExchange.Net/Sockets/Query.cs index 8719c20d..da566594 100644 --- a/CryptoExchange.Net/Sockets/Query.cs +++ b/CryptoExchange.Net/Sockets/Query.cs @@ -124,7 +124,7 @@ namespace CryptoExchange.Net.Sockets /// /// On complete callback /// - public Action? OnComplete { get; set; } + public Action? OnComplete { get; set; } /// 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 /// public abstract void Timeout(); + /// + /// Mark request as successfully completed + /// + public abstract void Complete(); + /// /// Mark request as failed /// @@ -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); + } + + /// + public override void Complete() + { + if (Completed) + return; + + Result = CallResult.Ok(default!); + Completed = true; + _event.Set(); + OnComplete?.Invoke(Success); } /// @@ -280,7 +296,7 @@ namespace CryptoExchange.Net.Sockets Completed = true; _event.Set(); - OnComplete?.Invoke(); + OnComplete?.Invoke(Success); } } }