WebSocket covered subscriptions

Updated WebSocket duplicate subscription handling
Fixed incorrect status keeping of WebSocket subscriptions which don't expect a response

CryptoExchange.Net/Clients/SocketApiClient.cs

@@ -533,9 +533,6 @@ namespace CryptoExchange.Net.Clients
             if (!connectResult.Success)
                 return connectResult;
 
-            if (ClientOptions.DelayAfterConnect != TimeSpan.Zero)
-                await Task.Delay(ClientOptions.DelayAfterConnect).ConfigureAwait(false);
-
             if (!authenticated || socket.Authenticated)
                 return CallResult.Ok();
 
@@ -862,7 +859,7 @@ namespace CryptoExchange.Net.Clients
                 RateLimitAdmissionCallbackRequest = () => AdmissionOverride.Value,
                 Proxy = ClientOptions.Proxy,
                 Timeout = ApiOptions.SocketNoDataTimeout ?? ClientOptions.SocketNoDataTimeout,
-                ReceiveBufferSize = ClientOptions.ReceiveBufferSize,
+                ReceiveBufferSize = ClientOptions.ReceiveBufferSize
             };
 
         /// <summary>

CryptoExchange.Net/Objects/Options/SocketApiOptions.cs

@@ -19,6 +19,11 @@ namespace CryptoExchange.Net.Objects.Options
         /// </summary>
         public int? MaxSocketConnections { get; set; }
 
+        /// <summary>
+        /// The time to wait after connecting a socket before sending messages. Can be used for API's which will rate limit if you subscribe directly after connecting.
+        /// </summary>
+        public TimeSpan? DelayAfterConnect { get; set; }
+
         /// <summary>
         /// Set the values of this options on the target options
         /// </summary>
@@ -28,6 +33,7 @@ namespace CryptoExchange.Net.Objects.Options
             item.SocketNoDataTimeout = SocketNoDataTimeout;
             item.AutoTimestamp = AutoTimestamp;
             item.MaxSocketConnections = MaxSocketConnections;
+            item.DelayAfterConnect = DelayAfterConnect;
             return item;
         }
     }

CryptoExchange.Net/SharedApis/CapabilityReferences/SharedCapabilities.Funding.cs

@@ -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();
         }
     }
 }

CryptoExchange.Net/SharedApis/Models/SharedIncrementalOrderBook.cs

@@ -0,0 +1,25 @@
+using CryptoExchange.Net.Interfaces;
+
+namespace CryptoExchange.Net.SharedApis
+{
+    /// <summary>
+    /// Order book info
+    /// </summary>
+    public record SharedIncrementalOrderBook : SharedOrderBook
+    {
+        /// <summary>
+        /// The sequence number of the first book update in this update.
+        /// </summary>
+        public long? StartSequenceNumber { get; set; }
+
+        /// <summary>
+        /// ctor
+        /// </summary>
+        public SharedIncrementalOrderBook(SharedQuantityType quantityType, long? startSequenceNumber, long? endSequenceNumber, ISymbolOrderBookEntry[] asks, ISymbolOrderBookEntry[] bids)
+            :base(quantityType, endSequenceNumber, asks, bids)
+        {
+            StartSequenceNumber = startSequenceNumber;
+        }
+    }
+
+}

CryptoExchange.Net/SharedApis/SharedUtils.cs

@@ -668,6 +668,20 @@ namespace CryptoExchange.Net.SharedApis
                 .ParallelEnumerateAsync();
         }
 
+        /// <summary>
+        /// Subscribe to funding info updates for all capabilities in parallel and return results as they arrive
+        /// </summary>
+        public static IAsyncEnumerable<WebSocketResult<UpdateSubscription>> SubscribeAllAsync(
+            this IEnumerable<SharedCapabilityResolution<ISubscribeFundingInfoSocket>> capabilities,
+            SubscribeFundingInfoRequest request,
+            Action<DataEvent<SharedFundingInfo>> onData,
+            CancellationToken ct = default)
+        {
+            return capabilities
+                .Select(x => x.Capability.SubscribeToFundingInfoUpdatesAsync(request, onData, ct))
+                .ParallelEnumerateAsync();
+        }
+
         /// <summary>
         /// Subscribe to kline updates for all capabilities in parallel and return results as they arrive
         /// </summary>
@@ -716,11 +730,11 @@ namespace CryptoExchange.Net.SharedApis
         public static IAsyncEnumerable<WebSocketResult<UpdateSubscription>> SubscribeAllAsync(
             this IEnumerable<SharedCapabilityResolution<ISubscribeIncrementalOrderBookSocket>> capabilities,
             SubscribeOrderBookRequest request,
-            Action<DataEvent<SharedOrderBook>> onData,
+            Action<DataEvent<SharedIncrementalOrderBook>> onData,
             CancellationToken ct = default)
         {
             return capabilities
-                .Select(x => x.Capability.SubscribeToOrderBookUpdatesAsync(request, onData, ct))
+                .Select(x => x.Capability.SubscribeToIncrementalOrderBookUpdatesAsync(request, onData, ct))
                 .ParallelEnumerateAsync();
         }
 

CryptoExchange.Net/SharedApis/V2/Funding/SubscribeFundingInfo/ISubscribeFundingInfoSocket.cs

@@ -0,0 +1,28 @@
+using CryptoExchange.Net.Objects;
+using CryptoExchange.Net.Objects.Sockets;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace CryptoExchange.Net.SharedApis
+{
+    /// <summary>
+    /// Operation for subscribing to funding info updates
+    /// </summary>
+    public interface ISubscribeFundingInfoSocket : ISharedSubscription
+    {
+        /// <summary>
+        /// Funding info subscription options
+        /// </summary>
+        SubscribeFundingInfoOptions SubscribeFundingInfoOptions { get; }
+
+        /// <summary>
+        /// Subscribe to funding info updates
+        /// </summary>
+        /// <param name="request">Request info</param>
+        /// <param name="handler">Update handler</param>
+        /// <param name="ct">Cancellation token, can be used to stop the updates</param>
+        /// <returns></returns>
+        Task<WebSocketResult<UpdateSubscription>> SubscribeToFundingInfoUpdatesAsync(SubscribeFundingInfoRequest request, Action<DataEvent<SharedFundingInfo>> handler, CancellationToken ct = default);
+    }
+}

CryptoExchange.Net/SharedApis/V2/Funding/SubscribeFundingInfo/SubscribeFundingInfoOptions.cs

@@ -0,0 +1,29 @@
+using CryptoExchange.Net.Objects;
+using System;
+using System.Linq;
+
+namespace CryptoExchange.Net.SharedApis
+{
+    /// <summary>
+    /// Options for subscribing to funding info updates
+    /// </summary>
+    public class SubscribeFundingInfoOptions : CapabilityOptions<SubscribeFundingInfoRequest, ISubscribeFundingInfoSocket>
+    {
+        /// <inheritdoc />
+        public override string Description => "Subscribe to funding info updates for a symbol";
+
+        private static readonly RequestParameterDescription[] _defaultParameterRules = new[]
+        {
+            RequestParameterRule<SubscribeFundingInfoRequest>.Optional(x => x.Symbol, "The symbol to subscribe to", new SharedSymbol(TradingMode.PerpetualLinear, "ETH", "USDT")),
+            RequestParameterRule<SubscribeFundingInfoRequest>.Optional(x => x.Symbols, "The symbols to subscribe to", new[] { new SharedSymbol(TradingMode.PerpetualLinear, "ETH", "USDT") }),
+        };
+
+        /// <summary>
+        /// ctor
+        /// </summary>
+        public SubscribeFundingInfoOptions(string exchange, bool needsAuthentication)
+            : base(exchange, needsAuthentication, nameof(ISubscribeFundingInfoSocket.SubscribeToFundingInfoUpdatesAsync), _defaultParameterRules, SharedTradingModeSets.Futures)
+        {
+        }
+    }
+}

CryptoExchange.Net/SharedApis/V2/Funding/SubscribeFundingInfo/SubscribeFundingInfoRequest.cs

@@ -0,0 +1,31 @@
+using System;
+using System.Collections.Generic;
+
+namespace CryptoExchange.Net.SharedApis
+{
+    /// <summary>
+    /// Request to subscribe to  funding info updates for a symbol
+    /// </summary>
+    public record SubscribeFundingInfoRequest : SharedSymbolRequest
+    {
+        /// <summary>
+        /// ctor
+        /// </summary>
+        /// <param name="symbol">The symbol to subscribe to</param>
+        /// <param name="exchangeParameters">Exchange specific parameters</param>
+        public SubscribeFundingInfoRequest(SharedSymbol symbol, ExchangeParameters? exchangeParameters = null)
+            : base(symbol, exchangeParameters)
+        {
+        }
+
+        /// <summary>
+        /// ctor
+        /// </summary>
+        /// <param name="symbols">The symbols to subscribe to</param>
+        /// <param name="exchangeParameters">Exchange specific parameters</param>
+        public SubscribeFundingInfoRequest(IEnumerable<SharedSymbol> symbols, ExchangeParameters? exchangeParameters = null)
+            : base(symbols, exchangeParameters)
+        {
+        }
+    }
+}

CryptoExchange.Net/SharedApis/V2/OrderBooks/SubscribeIncrementalOrderBook/ISubscribeIncrementalOrderBook.cs

@@ -14,7 +14,7 @@ namespace CryptoExchange.Net.SharedApis
         /// <summary>
         /// Order book subscription options
         /// </summary>
-        SubscribeOrderBookOptions SubscribeOrderBookOptions { get; }
+        SubscribeIncrementalOrderBookOptions SubscribeIncrementalOrderBookOptions { get; }
 
         /// <summary>
         /// Subscribe to incremental order book updates for a symbol
@@ -23,6 +23,6 @@ namespace CryptoExchange.Net.SharedApis
         /// <param name="handler">Update handler</param>
         /// <param name="ct">Cancellation token, can be used to stop the updates</param>
         /// <returns></returns>
-        Task<WebSocketResult<UpdateSubscription>> SubscribeToOrderBookUpdatesAsync(SubscribeOrderBookRequest request, Action<DataEvent<SharedOrderBook>> handler, CancellationToken ct = default);
+        Task<WebSocketResult<UpdateSubscription>> SubscribeToIncrementalOrderBookUpdatesAsync(SubscribeOrderBookRequest request, Action<DataEvent<SharedIncrementalOrderBook>> handler, CancellationToken ct = default);
     }
 }

CryptoExchange.Net/SharedApis/V2/OrderBooks/SubscribeIncrementalOrderBook/SubscribeIncrementalOrderBookOptions.cs

@@ -33,7 +33,7 @@ namespace CryptoExchange.Net.SharedApis
         /// ctor
         /// </summary>
         public SubscribeIncrementalOrderBookOptions(string exchange, bool needsAuthentication, int[] limits, SharedOrderBookSubscriptionType updateType)
-            : base(exchange, needsAuthentication, nameof(ISubscribeIncrementalOrderBookSocket.SubscribeToOrderBookUpdatesAsync), _defaultParameterRules)
+            : base(exchange, needsAuthentication, nameof(ISubscribeIncrementalOrderBookSocket.SubscribeToIncrementalOrderBookUpdatesAsync), _defaultParameterRules)
         {
             SupportedLimits = limits;
             UpdateType = updateType;

CryptoExchange.Net/Sockets/Default/SocketConnection.cs

@@ -424,6 +424,9 @@ namespace CryptoExchange.Net.Sockets.Default
             {
                 try
                 {
+                    if ((ApiClient.ApiOptions.DelayAfterConnect ?? ApiClient.ClientOptions.DelayAfterConnect) != TimeSpan.Zero)
+                        await Task.Delay(ApiClient.ApiOptions.DelayAfterConnect ?? ApiClient.ClientOptions.DelayAfterConnect).ConfigureAwait(false);
+
                     var reconnectSuccessful = await ProcessReconnectAsync().ConfigureAwait(false);
                     if (!reconnectSuccessful.Success)
                     {
@@ -616,7 +619,17 @@ namespace CryptoExchange.Net.Sockets.Default
         /// Connect the websocket
         /// </summary>
         /// <returns></returns>
-        public async Task<CallResult> ConnectAsync(CancellationToken ct) => await _socket.ConnectAsync(ct).ConfigureAwait(false);
+        public async Task<CallResult> ConnectAsync(CancellationToken ct)
+        {
+            var result = await _socket.ConnectAsync(ct).ConfigureAwait(false);
+            if (!result.Success)
+                return result;
+
+            if ((ApiClient.ApiOptions.DelayAfterConnect ?? ApiClient.ClientOptions.DelayAfterConnect) != TimeSpan.Zero)
+                await Task.Delay(ApiClient.ApiOptions.DelayAfterConnect ?? ApiClient.ClientOptions.DelayAfterConnect).ConfigureAwait(false);
+
+            return result;
+        }
 
         /// <summary>
         /// Retrieve the underlying socket
@@ -689,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)
@@ -1056,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
+                && x.MessageRouter.Routes.All(l => subscription.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 && other.MessageRouter.Routes.All(x => subscription.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>
@@ -1090,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)
@@ -1109,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)
             {
@@ -1119,9 +1166,6 @@ namespace CryptoExchange.Net.Sockets.Default
                 return WebSocketResult.Fail(ApiClient.Exchange, subQueryResult.Error!);
             }
 
-            if (!subQuery.ExpectsResponse)
-                subCompleteHandler();
-
             return subQueryResult;
         }
 

CryptoExchange.Net/Sockets/Query.cs

@@ -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;
+                Complete();
-                _event.Set();
             }
         }
 
@@ -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);
         }
     }
 }