Fixed socket client options setting, added automatic unsubscribe if the subscription confirmation comes in after request timeout

CryptoExchange.Net/Clients/SocketApiClient.cs

@@ -265,7 +265,7 @@ namespace CryptoExchange.Net
         protected internal virtual async Task<CallResult<bool>> SubscribeAndWaitAsync(SocketConnection socketConnection, object request, SocketSubscription subscription)
         {
             CallResult<object>? callResult = null;
-            await socketConnection.SendAndWaitAsync(request, Options.SocketResponseTimeout, data => HandleSubscriptionResponse(socketConnection, subscription, request, data, out callResult)).ConfigureAwait(false);
+            await socketConnection.SendAndWaitAsync(request, Options.SocketResponseTimeout, subscription, data => HandleSubscriptionResponse(socketConnection, subscription, request, data, out callResult)).ConfigureAwait(false);
 
             if (callResult?.Success == true)
             {
@@ -351,7 +351,7 @@ namespace CryptoExchange.Net
         protected virtual async Task<CallResult<T>> QueryAndWaitAsync<T>(SocketConnection socket, object request)
         {
             var dataResult = new CallResult<T>(new ServerError("No response on query received"));
-            await socket.SendAndWaitAsync(request, Options.SocketResponseTimeout, data =>
+            await socket.SendAndWaitAsync(request, Options.SocketResponseTimeout, null, data =>
             {
                 if (!HandleQueryResponse<T>(socket, request, data, out var callResult))
                     return false;

CryptoExchange.Net/Objects/Options.cs

@@ -298,14 +298,14 @@ namespace CryptoExchange.Net.Objects
             if (baseOptions == null)
                 return;
 
-            AutoReconnect = baseOptions.AutoReconnect;
+            AutoReconnect = newValues?.AutoReconnect ?? baseOptions.AutoReconnect;
-            ReconnectInterval = baseOptions.ReconnectInterval;
+            ReconnectInterval = newValues?.ReconnectInterval ?? baseOptions.ReconnectInterval;
-            MaxConcurrentResubscriptionsPerSocket = baseOptions.MaxConcurrentResubscriptionsPerSocket;
+            MaxConcurrentResubscriptionsPerSocket = newValues?.MaxConcurrentResubscriptionsPerSocket ?? baseOptions.MaxConcurrentResubscriptionsPerSocket;
-            SocketResponseTimeout = baseOptions.SocketResponseTimeout;
+            SocketResponseTimeout = newValues?.SocketResponseTimeout ?? baseOptions.SocketResponseTimeout;
-            SocketNoDataTimeout = baseOptions.SocketNoDataTimeout;
+            SocketNoDataTimeout = newValues?.SocketNoDataTimeout ?? baseOptions.SocketNoDataTimeout;
-            SocketSubscriptionsCombineTarget = baseOptions.SocketSubscriptionsCombineTarget;
+            SocketSubscriptionsCombineTarget = newValues?.SocketSubscriptionsCombineTarget ?? baseOptions.SocketSubscriptionsCombineTarget;
-            MaxSocketConnections = baseOptions.MaxSocketConnections;
+            MaxSocketConnections = newValues?.MaxSocketConnections ?? baseOptions.MaxSocketConnections;
-            DelayAfterConnect = baseOptions.DelayAfterConnect;
+            DelayAfterConnect = newValues?.DelayAfterConnect ?? baseOptions.DelayAfterConnect;
         }
 
         /// <inheritdoc />

CryptoExchange.Net/Sockets/PendingRequest.cs

@@ -11,15 +11,19 @@ namespace CryptoExchange.Net.Sockets
         public JToken? Result { get; private set; }
         public bool Completed { get; private set; }
         public AsyncResetEvent Event { get; }
+        public DateTime RequestTimestamp { get; set; }
         public TimeSpan Timeout { get; }
+        public SocketSubscription? Subscription { get; }
 
         private CancellationTokenSource cts;
 
-        public PendingRequest(Func<JToken, bool> handler, TimeSpan timeout)
+        public PendingRequest(Func<JToken, bool> handler, TimeSpan timeout, SocketSubscription? subscription)
         {
             Handler = handler;
             Event = new AsyncResetEvent(false, false);
             Timeout = timeout;
+            RequestTimestamp = DateTime.UtcNow;
+            Subscription = subscription;
 
             cts = new CancellationTokenSource(timeout);
             cts.Token.Register(Fail, false);
@@ -27,7 +31,10 @@ namespace CryptoExchange.Net.Sockets
 
         public bool CheckData(JToken data)
         {
-            if (Handler(data))
+            return Handler(data);
+        }
+
+        public bool Succeed(JToken data)
         {
             Result = data;
             Completed = true;
@@ -35,9 +42,6 @@ namespace CryptoExchange.Net.Sockets
             return true;
         }
 
-            return false;
-        }
-
         public void Fail()
         {
             Completed = true;

CryptoExchange.Net/Sockets/SocketConnection.cs

@@ -304,18 +304,35 @@ namespace CryptoExchange.Net.Sockets
             PendingRequest[] requests;
             lock (_pendingRequests)
             {
-                _pendingRequests.RemoveAll(r => r.Completed);
+                // Remove only timed out requests after 5 minutes have passed so we can still process any 
+                // message coming in after the request timeout
+                _pendingRequests.RemoveAll(r => r.Completed && DateTime.UtcNow - r.RequestTimestamp > TimeSpan.FromMinutes(5));
                 requests = _pendingRequests.ToArray();
             }
 
             // Check if this message is an answer on any pending requests
             foreach (var pendingRequest in requests)
             {
+
                 if (pendingRequest.CheckData(tokenData))
                 {
                     lock (_pendingRequests)
                         _pendingRequests.Remove(pendingRequest);
 
+                    if (pendingRequest.Completed)
+                    {
+                        // Answer to a timed out request, unsub if it is a subscription request
+                        if (pendingRequest.Subscription != null)
+                        {
+                            _log.Write(LogLevel.Warning, "Received subscription info after request timed out; unsubscribing. Consider increasing the SocketResponseTimout");
+                            _ = ApiClient.UnsubscribeAsync(this, pendingRequest.Subscription).ConfigureAwait(false);
+                        }
+                    }
+                    else
+                    {
+                        pendingRequest.Succeed(tokenData);
+                    }
+
                     if (!ApiClient.ContinueOnQueryResponse)
                         return;
 
@@ -546,11 +563,12 @@ namespace CryptoExchange.Net.Sockets
         /// <typeparam name="T">The data type expected in response</typeparam>
         /// <param name="obj">The object to send</param>
         /// <param name="timeout">The timeout for response</param>
+        /// <param name="subscription">Subscription if this is a subscribe request</param>
         /// <param name="handler">The response handler, should return true if the received JToken was the response to the request</param>
         /// <returns></returns>
-        public virtual Task SendAndWaitAsync<T>(T obj, TimeSpan timeout, Func<JToken, bool> handler)
+        public virtual Task SendAndWaitAsync<T>(T obj, TimeSpan timeout, SocketSubscription? subscription, Func<JToken, bool> handler)
         {
-            var pending = new PendingRequest(handler, timeout);
+            var pending = new PendingRequest(handler, timeout, subscription);
             lock (_pendingRequests)
             {
                 _pendingRequests.Add(pending);