Updated version

change SetProxy method in web socket client to support socks5 proxy.

CryptoExchange.Net.UnitTests/RestClientTests.cs

@@ -144,7 +144,7 @@ namespace CryptoExchange.Net.UnitTests
                 }
             });
 
-            client.SetParameterPosition(new HttpMethod(method), pos);
+            client.Api1.SetParameterPosition(new HttpMethod(method), pos);
 
             client.SetResponse("{}", out var request);
 

CryptoExchange.Net.UnitTests/TestImplementations/TestRestClient.cs

@@ -31,11 +31,6 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
             RequestFactory = new Mock<IRequestFactory>().Object;
         }
 
-        public void SetParameterPosition(HttpMethod method, HttpMethodParameterPosition position)
-        {
-            ParameterPositions[method] = position;
-        }
-
         public void SetResponse(string responseData, out IRequest requestObj)
         {
             var expectedBytes = Encoding.UTF8.GetBytes(responseData);
@@ -120,7 +115,11 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
     {
         public TestRestApi1Client(TestClientOptions options): base(options, options.Api1Options)
         {
+        }
 
+        public void SetParameterPosition(HttpMethod method, HttpMethodParameterPosition position)
+        {
+            ParameterPositions[method] = position;
         }
 
         public override TimeSpan GetTimeOffset()

CryptoExchange.Net.UnitTests/TestImplementations/TestSocket.cs

@@ -38,6 +38,8 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
 
         public double IncomingKbps => throw new NotImplementedException();
 
+        public Uri Uri => new Uri("");
+
         public static int lastId = 0;
         public static object lastIdLock = new object();
 
@@ -111,5 +113,11 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
         {
             OnError?.Invoke(error);
         }
+
+        public async Task ProcessAsync()
+        {
+            while (Connected)
+                await Task.Delay(50);
+        }
     }
 }

CryptoExchange.Net/Clients/BaseApiClient.cs

@@ -1,4 +1,6 @@
 using System;
+using System.Collections.Generic;
+using System.Net.Http;
 using CryptoExchange.Net.Authentication;
 using CryptoExchange.Net.Objects;
 
@@ -31,6 +33,37 @@ namespace CryptoExchange.Net
             }
         }
 
+        /// <summary>
+        /// Where to put the parameters for requests with different Http methods
+        /// </summary>
+        public Dictionary<HttpMethod, HttpMethodParameterPosition> ParameterPositions { get; set; } = new Dictionary<HttpMethod, HttpMethodParameterPosition>
+        {
+            { HttpMethod.Get, HttpMethodParameterPosition.InUri },
+            { HttpMethod.Post, HttpMethodParameterPosition.InBody },
+            { HttpMethod.Delete, HttpMethodParameterPosition.InBody },
+            { HttpMethod.Put, HttpMethodParameterPosition.InBody }
+        };
+
+        /// <summary>
+        /// Request body content type
+        /// </summary>
+        public RequestBodyFormat requestBodyFormat = RequestBodyFormat.Json;
+
+        /// <summary>
+        /// Whether or not we need to manually parse an error instead of relying on the http status code
+        /// </summary>
+        public bool manualParseError = false;
+
+        /// <summary>
+        /// How to serialize array parameters when making requests
+        /// </summary>
+        public ArrayParametersSerialization arraySerialization = ArrayParametersSerialization.Array;
+
+        /// <summary>
+        /// What request body should be set when no data is send (only used in combination with postParametersPosition.InBody)
+        /// </summary>
+        public string requestBodyEmptyContent = "{}";
+
         /// <summary>
         /// The base address for this API client
         /// </summary>

CryptoExchange.Net/Clients/BaseRestClient.cs

@@ -26,38 +26,7 @@ namespace CryptoExchange.Net
         /// The factory for creating requests. Used for unit testing
         /// </summary>
         public IRequestFactory RequestFactory { get; set; } = new RequestFactory();
-
-        /// <summary>
-        /// Where to put the parameters for requests with different Http methods
-        /// </summary>
-        protected Dictionary<HttpMethod, HttpMethodParameterPosition> ParameterPositions { get; set; } = new Dictionary<HttpMethod, HttpMethodParameterPosition>
-        { 
-            { HttpMethod.Get, HttpMethodParameterPosition.InUri },
-            { HttpMethod.Post, HttpMethodParameterPosition.InBody },
-            { HttpMethod.Delete, HttpMethodParameterPosition.InBody },
-            { HttpMethod.Put, HttpMethodParameterPosition.InBody }
-        };
-
-        /// <summary>
-        /// Request body content type
-        /// </summary>
-        protected RequestBodyFormat requestBodyFormat = RequestBodyFormat.Json;
-
-        /// <summary>
-        /// Whether or not we need to manually parse an error instead of relying on the http status code
-        /// </summary>
-        protected bool manualParseError = false;
-
-        /// <summary>
-        /// How to serialize array parameters when making requests
-        /// </summary>
-        protected ArrayParametersSerialization arraySerialization = ArrayParametersSerialization.Array;
-
-        /// <summary>
-        /// What request body should be set when no data is send (only used in combination with postParametersPosition.InBody)
-        /// </summary>
-        protected string requestBodyEmptyContent = "{}";
-
+        
         /// <inheritdoc />
         public int TotalRequestsMade => ApiClients.OfType<RestApiClient>().Sum(s => s.TotalRequestsMade);
 
@@ -92,6 +61,44 @@ namespace CryptoExchange.Net
                 apiClient.SetApiCredentials(credentials);
         }
 
+        /// <summary>
+        /// Execute a request to the uri and returns if it was successful
+        /// </summary>
+        /// <param name="apiClient">The API client the request is for</param>
+        /// <param name="uri">The uri to send the request to</param>
+        /// <param name="method">The method of the request</param>
+        /// <param name="cancellationToken">Cancellation token</param>
+        /// <param name="parameters">The parameters of the request</param>
+        /// <param name="signed">Whether or not the request should be authenticated</param>
+        /// <param name="parameterPosition">Where the parameters should be placed, overwrites the value set in the client</param>
+        /// <param name="arraySerialization">How array parameters should be serialized, overwrites the value set in the client</param>
+        /// <param name="requestWeight">Credits used for the request</param>
+        /// <param name="deserializer">The JsonSerializer to use for deserialization</param>
+        /// <param name="additionalHeaders">Additional headers to send with the request</param>
+        /// <param name="ignoreRatelimit">Ignore rate limits for this request</param>
+        /// <returns></returns>
+        [return: NotNull]
+        protected virtual async Task<WebCallResult> SendRequestAsync(RestApiClient apiClient,
+            Uri uri,
+            HttpMethod method,
+            CancellationToken cancellationToken,
+            Dictionary<string, object>? parameters = null,
+            bool signed = false,
+            HttpMethodParameterPosition? parameterPosition = null,
+            ArrayParametersSerialization? arraySerialization = null,
+            int requestWeight = 1,
+            JsonSerializer? deserializer = null,
+            Dictionary<string, string>? additionalHeaders = null,
+            bool ignoreRatelimit = false)
+        {
+            var request = await PrepareRequestAsync(apiClient, uri, method, cancellationToken, parameters, signed, parameterPosition, arraySerialization, requestWeight, deserializer, additionalHeaders, ignoreRatelimit).ConfigureAwait(false);
+            if (!request)
+                return new WebCallResult(request.Error!);
+
+            var result = await GetResponseAsync<object>(apiClient, request.Data, deserializer, cancellationToken, true).ConfigureAwait(false);
+            return result.AsDataless();
+        }
+
         /// <summary>
         /// Execute a request to the uri and deserialize the response into the provided type parameter
         /// </summary>
@@ -124,6 +131,42 @@ namespace CryptoExchange.Net
             Dictionary<string, string>? additionalHeaders = null,
             bool ignoreRatelimit = false
             ) where T : class
+        {
+            var request = await PrepareRequestAsync(apiClient, uri, method, cancellationToken, parameters, signed, parameterPosition, arraySerialization, requestWeight, deserializer, additionalHeaders, ignoreRatelimit).ConfigureAwait(false);
+            if (!request)
+                return new WebCallResult<T>(request.Error!);
+
+            return await GetResponseAsync<T>(apiClient, request.Data, deserializer, cancellationToken, false).ConfigureAwait(false);
+        }
+
+        /// <summary>
+        /// Prepares a request to be sent to the server
+        /// </summary>
+        /// <param name="apiClient">The API client the request is for</param>
+        /// <param name="uri">The uri to send the request to</param>
+        /// <param name="method">The method of the request</param>
+        /// <param name="cancellationToken">Cancellation token</param>
+        /// <param name="parameters">The parameters of the request</param>
+        /// <param name="signed">Whether or not the request should be authenticated</param>
+        /// <param name="parameterPosition">Where the parameters should be placed, overwrites the value set in the client</param>
+        /// <param name="arraySerialization">How array parameters should be serialized, overwrites the value set in the client</param>
+        /// <param name="requestWeight">Credits used for the request</param>
+        /// <param name="deserializer">The JsonSerializer to use for deserialization</param>
+        /// <param name="additionalHeaders">Additional headers to send with the request</param>
+        /// <param name="ignoreRatelimit">Ignore rate limits for this request</param>
+        /// <returns></returns>
+        protected virtual async Task<CallResult<IRequest>> PrepareRequestAsync(RestApiClient apiClient,
+            Uri uri,
+            HttpMethod method,
+            CancellationToken cancellationToken,
+            Dictionary<string, object>? parameters = null,
+            bool signed = false,
+            HttpMethodParameterPosition? parameterPosition = null,
+            ArrayParametersSerialization? arraySerialization = null,
+            int requestWeight = 1,
+            JsonSerializer? deserializer = null,
+            Dictionary<string, string>? additionalHeaders = null,
+            bool ignoreRatelimit = false)
         {
             var requestId = NextId();
 
@@ -133,7 +176,7 @@ namespace CryptoExchange.Net
                 if (!syncTimeResult)
                 {
                     log.Write(LogLevel.Debug, $"[{requestId}] Failed to sync time, aborting request: " + syncTimeResult.Error);
-                    return syncTimeResult.As<T>(default);
+                    return syncTimeResult.As<IRequest>(default);
                 }
             }
 
@@ -143,20 +186,20 @@ namespace CryptoExchange.Net
                 {
                     var limitResult = await limiter.LimitRequestAsync(log, uri.AbsolutePath, method, signed, apiClient.Options.ApiCredentials?.Key, apiClient.Options.RateLimitingBehaviour, requestWeight, cancellationToken).ConfigureAwait(false);
                     if (!limitResult.Success)
-                        return new WebCallResult<T>(limitResult.Error!);
+                        return new CallResult<IRequest>(limitResult.Error!);
                 }
             }
 
             if (signed && apiClient.AuthenticationProvider == null)
             {
                 log.Write(LogLevel.Warning, $"[{requestId}] Request {uri.AbsolutePath} failed because no ApiCredentials were provided");
-                return new WebCallResult<T>(new NoApiCredentialsError());
+                return new CallResult<IRequest>(new NoApiCredentialsError());
             }
 
             log.Write(LogLevel.Information, $"[{requestId}] Creating request for " + uri);
-            var paramsPosition = parameterPosition ?? ParameterPositions[method];
-            var request = ConstructRequest(apiClient, uri, method, parameters, signed, paramsPosition, arraySerialization ?? this.arraySerialization, requestId, additionalHeaders);
-            
+            var paramsPosition = parameterPosition ?? apiClient.ParameterPositions[method];
+            var request = ConstructRequest(apiClient, uri, method, parameters, signed, paramsPosition, arraySerialization ?? apiClient.arraySerialization, requestId, additionalHeaders);
+
             string? paramString = "";
             if (paramsPosition == HttpMethodParameterPosition.InBody)
                 paramString = $" with request body '{request.Content}'";
@@ -164,20 +207,29 @@ namespace CryptoExchange.Net
             var headers = request.GetHeaders();
             if (headers.Any())
                 paramString += " with headers " + string.Join(", ", headers.Select(h => h.Key + $"=[{string.Join(",", h.Value)}]"));
-            
+
             apiClient.TotalRequestsMade++;
             log.Write(LogLevel.Trace, $"[{requestId}] Sending {method}{(signed ? " signed" : "")} request to {request.Uri}{paramString ?? " "}{(ClientOptions.Proxy == null ? "" : $" via proxy {ClientOptions.Proxy.Host}")}");
-            return await GetResponseAsync<T>(request, deserializer, cancellationToken).ConfigureAwait(false);
+            return new CallResult<IRequest>(request);
         }
 
+
+
         /// <summary>
         /// Executes the request and returns the result deserialized into the type parameter class
         /// </summary>
+        /// <param name="apiClient">The client making the request</param>
         /// <param name="request">The request object to execute</param>
         /// <param name="deserializer">The JsonSerializer to use for deserialization</param>
         /// <param name="cancellationToken">Cancellation token</param>
+        /// <param name="expectedEmptyResponse">If an empty response is expected</param>
         /// <returns></returns>
-        protected virtual async Task<WebCallResult<T>> GetResponseAsync<T>(IRequest request, JsonSerializer? deserializer, CancellationToken cancellationToken)
+        protected virtual async Task<WebCallResult<T>> GetResponseAsync<T>(
+            BaseApiClient apiClient, 
+            IRequest request, 
+            JsonSerializer? deserializer, 
+            CancellationToken cancellationToken,
+            bool expectedEmptyResponse)
         {
             try
             {
@@ -191,7 +243,7 @@ namespace CryptoExchange.Net
                 {
                     // If we have to manually parse error responses (can't rely on HttpStatusCode) we'll need to read the full
                     // response before being able to deserialize it into the resulting type since we don't know if it an error response or data
-                    if (manualParseError)
+                    if (apiClient.manualParseError)
                     {
                         using var reader = new StreamReader(responseStream);
                         var data = await reader.ReadToEndAsync().ConfigureAwait(false);
@@ -199,22 +251,52 @@ namespace CryptoExchange.Net
                         response.Close();
                         log.Write(LogLevel.Debug, $"[{request.RequestId}] Response received in {sw.ElapsedMilliseconds}ms{(log.Level == LogLevel.Trace ? (": "+data): "")}");
 
-                        // Validate if it is valid json. Sometimes other data will be returned, 502 error html pages for example
-                        var parseResult = ValidateJson(data);
-                        if (!parseResult.Success)
-                            return new WebCallResult<T>(response.StatusCode, response.ResponseHeaders, sw.Elapsed, ClientOptions.OutputOriginalData ? data : null, request.Uri.ToString(), request.Content, request.Method, request.GetHeaders(), default, parseResult.Error!);
+                        if (!expectedEmptyResponse)
+                        {
+                            // Validate if it is valid json. Sometimes other data will be returned, 502 error html pages for example
+                            var parseResult = ValidateJson(data);
+                            if (!parseResult.Success)
+                                return new WebCallResult<T>(response.StatusCode, response.ResponseHeaders, sw.Elapsed, ClientOptions.OutputOriginalData ? data : null, request.Uri.ToString(), request.Content, request.Method, request.GetHeaders(), default, parseResult.Error!);
 
-                        // Let the library implementation see if it is an error response, and if so parse the error
-                        var error = await TryParseErrorAsync(parseResult.Data).ConfigureAwait(false);
-                        if (error != null)
-                            return new WebCallResult<T>(response.StatusCode, response.ResponseHeaders, sw.Elapsed, ClientOptions.OutputOriginalData ? data : null, request.Uri.ToString(), request.Content, request.Method, request.GetHeaders(), default, error!);
+                            // Let the library implementation see if it is an error response, and if so parse the error
+                            var error = await TryParseErrorAsync(parseResult.Data).ConfigureAwait(false);
+                            if (error != null)
+                                return new WebCallResult<T>(response.StatusCode, response.ResponseHeaders, sw.Elapsed, ClientOptions.OutputOriginalData ? data : null, request.Uri.ToString(), request.Content, request.Method, request.GetHeaders(), default, error!);
 
-                        // Not an error, so continue deserializing
-                        var deserializeResult = Deserialize<T>(parseResult.Data, deserializer, request.RequestId);
-                        return new WebCallResult<T>(response.StatusCode, response.ResponseHeaders, sw.Elapsed, ClientOptions.OutputOriginalData ? data: null, request.Uri.ToString(), request.Content, request.Method, request.GetHeaders(), deserializeResult.Data, deserializeResult.Error);
+                            // Not an error, so continue deserializing
+                            var deserializeResult = Deserialize<T>(parseResult.Data, deserializer, request.RequestId);
+                            return new WebCallResult<T>(response.StatusCode, response.ResponseHeaders, sw.Elapsed, ClientOptions.OutputOriginalData ? data : null, request.Uri.ToString(), request.Content, request.Method, request.GetHeaders(), deserializeResult.Data, deserializeResult.Error);
+                        }
+                        else
+                        {
+                            if (!string.IsNullOrEmpty(data))
+                            {
+                                var parseResult = ValidateJson(data);
+                                if (!parseResult.Success)
+                                    // Not empty, and not json
+                                    return new WebCallResult<T>(response.StatusCode, response.ResponseHeaders, sw.Elapsed, ClientOptions.OutputOriginalData ? data : null, request.Uri.ToString(), request.Content, request.Method, request.GetHeaders(), default, parseResult.Error!);
+
+                                var error = await TryParseErrorAsync(parseResult.Data).ConfigureAwait(false);
+                                if (error != null)
+                                    // Error response
+                                    return new WebCallResult<T>(response.StatusCode, response.ResponseHeaders, sw.Elapsed, ClientOptions.OutputOriginalData ? data : null, request.Uri.ToString(), request.Content, request.Method, request.GetHeaders(), default, error!);
+                            }
+
+                            // Empty success response; okay
+                            return new WebCallResult<T>(response.StatusCode, response.ResponseHeaders, sw.Elapsed, ClientOptions.OutputOriginalData ? data : null, request.Uri.ToString(), request.Content, request.Method, request.GetHeaders(), default, default);
+                        }
                     }
                     else
                     {
+                        if (expectedEmptyResponse)
+                        {
+                            // We expected an empty response and the request is successful and don't manually parse errors, so assume it's correct
+                            responseStream.Close();
+                            response.Close();
+
+                            return new WebCallResult<T>(statusCode, headers, sw.Elapsed, null, request.Uri.ToString(), request.Content, request.Method, request.GetHeaders(), default, null);
+                        }
+
                         // Success status code, and we don't have to check for errors. Continue deserializing directly from the stream
                         var desResult = await DeserializeAsync<T>(responseStream, deserializer, request.RequestId, sw.ElapsedMilliseconds).ConfigureAwait(false);
                         responseStream.Close();
@@ -362,11 +444,11 @@ namespace CryptoExchange.Net
 
             if (parameterPosition == HttpMethodParameterPosition.InBody)
             {
-                var contentType = requestBodyFormat == RequestBodyFormat.Json ? Constants.JsonContentHeader : Constants.FormContentHeader;
+                var contentType = apiClient.requestBodyFormat == RequestBodyFormat.Json ? Constants.JsonContentHeader : Constants.FormContentHeader;
                 if (bodyParameters.Any())
-                    WriteParamBody(request, bodyParameters, contentType);
+                    WriteParamBody(apiClient, request, bodyParameters, contentType);
                 else
-                    request.SetContent(requestBodyEmptyContent, contentType);
+                    request.SetContent(apiClient.requestBodyEmptyContent, contentType);
             }
 
             return request;
@@ -375,18 +457,19 @@ namespace CryptoExchange.Net
         /// <summary>
         /// Writes the parameters of the request to the request object body
         /// </summary>
+        /// <param name="apiClient">The client making the request</param>
         /// <param name="request">The request to set the parameters on</param>
         /// <param name="parameters">The parameters to set</param>
         /// <param name="contentType">The content type of the data</param>
-        protected virtual void WriteParamBody(IRequest request, SortedDictionary<string, object> parameters, string contentType)
+        protected virtual void WriteParamBody(BaseApiClient apiClient, IRequest request, SortedDictionary<string, object> parameters, string contentType)
         {
-            if (requestBodyFormat == RequestBodyFormat.Json)
+            if (apiClient.requestBodyFormat == RequestBodyFormat.Json)
             {
                 // Write the parameters as json in the body
                 var stringData = JsonConvert.SerializeObject(parameters);
                 request.SetContent(stringData, contentType);
             }
-            else if (requestBodyFormat == RequestBodyFormat.FormData)
+            else if (apiClient.requestBodyFormat == RequestBodyFormat.FormData)
             {
                 // Write the parameters as form data in the body
                 var stringData = parameters.ToFormData();

CryptoExchange.Net/Clients/BaseSocketClient.cs

@@ -6,6 +6,7 @@ using System.Linq;
 using System.Net.WebSockets;
 using System.Threading;
 using System.Threading.Tasks;
+using CryptoExchange.Net.Authentication;
 using CryptoExchange.Net.Interfaces;
 using CryptoExchange.Net.Objects;
 using CryptoExchange.Net.Sockets;
@@ -29,7 +30,7 @@ namespace CryptoExchange.Net
         /// <summary>
         /// List of socket connections currently connecting/connected
         /// </summary>
-        protected internal ConcurrentDictionary<int, SocketConnection> sockets = new();
+        protected internal ConcurrentDictionary<int, SocketConnection> socketConnections = new();
         /// <summary>
         /// Semaphore used while creating sockets
         /// </summary>
@@ -84,10 +85,10 @@ namespace CryptoExchange.Net
         {
             get
             {
-                if (!sockets.Any())
+                if (!socketConnections.Any())
                     return 0;
 
-                return sockets.Sum(s => s.Value.Socket.IncomingKbps);
+                return socketConnections.Sum(s => s.Value.IncomingKbps);
             }
         }
 
@@ -111,6 +112,13 @@ namespace CryptoExchange.Net
             ClientOptions = options;
         }
 
+        /// <inheritdoc />
+        public void SetApiCredentials(ApiCredentials credentials)
+        {
+            foreach (var apiClient in ApiClients)
+                apiClient.SetApiCredentials(credentials);
+        }
+
         /// <summary>
         /// Set a delegate to be used for processing data received from socket connections before it is processed by handlers
         /// </summary>
@@ -197,7 +205,7 @@ namespace CryptoExchange.Net
 
             if (socketConnection.PausedActivity)
             {
-                log.Write(LogLevel.Warning, $"Socket {socketConnection.Socket.Id} has been paused, can't subscribe at this moment");
+                log.Write(LogLevel.Warning, $"Socket {socketConnection.SocketId} has been paused, can't subscribe at this moment");
                 return new CallResult<UpdateSubscription>( new ServerError("Socket is paused"));
             }
 
@@ -222,12 +230,12 @@ namespace CryptoExchange.Net
             {
                 subscription.CancellationTokenRegistration = ct.Register(async () =>
                 {
-                    log.Write(LogLevel.Information, $"Socket {socketConnection.Socket.Id} Cancellation token set, closing subscription");
+                    log.Write(LogLevel.Information, $"Socket {socketConnection.SocketId} Cancellation token set, closing subscription");
                     await socketConnection.CloseAsync(subscription).ConfigureAwait(false);
                 }, false);
             }
 
-            log.Write(LogLevel.Information, $"Socket {socketConnection.Socket.Id} subscription completed");
+            log.Write(LogLevel.Information, $"Socket {socketConnection.SocketId} subscription completed");
             return new CallResult<UpdateSubscription>(new UpdateSubscription(socketConnection, subscription));
         }
 
@@ -309,7 +317,7 @@ namespace CryptoExchange.Net
 
             if (socketConnection.PausedActivity)
             {
-                log.Write(LogLevel.Warning, $"Socket {socketConnection.Socket.Id} has been paused, can't send query at this moment");
+                log.Write(LogLevel.Warning, $"Socket {socketConnection.SocketId} has been paused, can't send query at this moment");
                 return new CallResult<T>(new ServerError("Socket is paused"));
             }
 
@@ -360,7 +368,7 @@ namespace CryptoExchange.Net
             if (!result)
             {
                 await socket.CloseAsync().ConfigureAwait(false);
-                log.Write(LogLevel.Warning, $"Socket {socket.Socket.Id} authentication failed");
+                log.Write(LogLevel.Warning, $"Socket {socket.SocketId} authentication failed");
                 result.Error!.Message = "Authentication failed: " + result.Error.Message;
                 return new CallResult<bool>(result.Error);
             }
@@ -435,6 +443,9 @@ namespace CryptoExchange.Net
         /// <param name="message"></param>
         /// <returns></returns>
         protected internal virtual JToken ProcessTokenData(JToken message)
+
+
+
         {
             return message;
         }
@@ -463,7 +474,7 @@ namespace CryptoExchange.Net
                 var desResult = Deserialize<T>(messageEvent.JsonData);
                 if (!desResult)
                 {
-                    log.Write(LogLevel.Warning, $"Socket {connection.Socket.Id} Failed to deserialize data into type {typeof(T)}: {desResult.Error}");
+                    log.Write(LogLevel.Warning, $"Socket {connection.SocketId} Failed to deserialize data into type {typeof(T)}: {desResult.Error}");
                     return;
                 }
 
@@ -486,7 +497,7 @@ namespace CryptoExchange.Net
         {
             genericHandlers.Add(identifier, action);
             var subscription = SocketSubscription.CreateForIdentifier(NextId(), identifier, false, action);
-            foreach (var connection in sockets.Values)
+            foreach (var connection in socketConnections.Values)
                 connection.AddSubscription(subscription);
         }
 
@@ -499,13 +510,13 @@ namespace CryptoExchange.Net
         /// <returns></returns>
         protected virtual SocketConnection GetSocketConnection(SocketApiClient apiClient, string address, bool authenticated)
         {
-            var socketResult = sockets.Where(s => s.Value.Socket.Url.TrimEnd('/') == address.TrimEnd('/')
+            var socketResult = socketConnections.Where(s => s.Value.Uri.ToString().TrimEnd('/') == address.TrimEnd('/')
                                                   && (s.Value.ApiClient.GetType() == apiClient.GetType())
                                                   && (s.Value.Authenticated == authenticated || !authenticated) && s.Value.Connected).OrderBy(s => s.Value.SubscriptionCount).FirstOrDefault();
             var result = socketResult.Equals(default(KeyValuePair<int, SocketConnection>)) ? null : socketResult.Value;
             if (result != null)
             {
-                if (result.SubscriptionCount < ClientOptions.SocketSubscriptionsCombineTarget || (sockets.Count >= MaxSocketConnections && sockets.All(s => s.Value.SubscriptionCount >= ClientOptions.SocketSubscriptionsCombineTarget)))
+                if (result.SubscriptionCount < ClientOptions.SocketSubscriptionsCombineTarget || (socketConnections.Count >= MaxSocketConnections && socketConnections.All(s => s.Value.SubscriptionCount >= ClientOptions.SocketSubscriptionsCombineTarget)))
                 {
                     // Use existing socket if it has less than target connections OR it has the least connections and we can't make new
                     return result;
@@ -540,13 +551,13 @@ namespace CryptoExchange.Net
         /// <returns></returns>
         protected virtual async Task<CallResult<bool>> ConnectSocketAsync(SocketConnection socketConnection)
         {
-            if (await socketConnection.Socket.ConnectAsync().ConfigureAwait(false))
+            if (await socketConnection.ConnectAsync().ConfigureAwait(false))
             {
-                sockets.TryAdd(socketConnection.Socket.Id, socketConnection);
+                socketConnections.TryAdd(socketConnection.SocketId, socketConnection);
                 return new CallResult<bool>(true);
             }
 
-            socketConnection.Socket.Dispose();
+            socketConnection.Dispose();
             return new CallResult<bool>(new CantConnectError());
         }
 
@@ -597,27 +608,27 @@ namespace CryptoExchange.Net
                     if (disposing)
                         break;
                     
-                    foreach (var socket in sockets.Values)
+                    foreach (var socketConnection in socketConnections.Values)
                     {
                         if (disposing)
                             break;
 
-                        if (!socket.Socket.IsOpen)
+                        if (!socketConnection.Connected)
                             continue;
 
-                        var obj = objGetter(socket);
+                        var obj = objGetter(socketConnection);
                         if (obj == null)
                             continue;
 
-                        log.Write(LogLevel.Trace, $"Socket {socket.Socket.Id} sending periodic {identifier}");
+                        log.Write(LogLevel.Trace, $"Socket {socketConnection.SocketId} sending periodic {identifier}");
 
                         try
                         {
-                            socket.Send(obj);
+                            socketConnection.Send(obj);
                         }
                         catch (Exception ex)
                         {
-                            log.Write(LogLevel.Warning, $"Socket {socket.Socket.Id} Periodic send {identifier} failed: " + ex.ToLogString());
+                            log.Write(LogLevel.Warning, $"Socket {socketConnection.SocketId} Periodic send {identifier} failed: " + ex.ToLogString());
                         }
                     }
                 }
@@ -634,7 +645,7 @@ namespace CryptoExchange.Net
 
             SocketSubscription? subscription = null;
             SocketConnection? connection = null;
-            foreach(var socket in sockets.Values.ToList())
+            foreach(var socket in socketConnections.Values.ToList())
             {
                 subscription = socket.GetSubscription(subscriptionId);
                 if (subscription != null)
@@ -671,19 +682,15 @@ namespace CryptoExchange.Net
         /// <returns></returns>
         public virtual async Task UnsubscribeAllAsync()
         {
-            log.Write(LogLevel.Information, $"Closing all {sockets.Sum(s => s.Value.SubscriptionCount)} subscriptions");
-
-            await Task.Run(async () =>
+            log.Write(LogLevel.Information, $"Closing all {socketConnections.Sum(s => s.Value.SubscriptionCount)} subscriptions");
+            var tasks = new List<Task>();
             {
-                var tasks = new List<Task>();
-                {
-                    var socketList = sockets.Values;
-                    foreach (var sub in socketList)
-                        tasks.Add(sub.CloseAsync());
-                }
+                var socketList = socketConnections.Values;
+                foreach (var sub in socketList)
+                    tasks.Add(sub.CloseAsync());
+            }
 
-                await Task.WhenAll(tasks.ToArray()).ConfigureAwait(false);
-            }).ConfigureAwait(false);
+            await Task.WhenAll(tasks.ToArray()).ConfigureAwait(false);
         }
 
         /// <summary>
@@ -695,7 +702,7 @@ namespace CryptoExchange.Net
             periodicEvent?.Set();
             periodicEvent?.Dispose();
             log.Write(LogLevel.Debug, "Disposing socket client, closing all subscriptions");
-            Task.Run(UnsubscribeAllAsync).ConfigureAwait(false).GetAwaiter().GetResult();
+            _ = UnsubscribeAllAsync();
             semaphoreSlim?.Dispose();
             base.Dispose();
         }

CryptoExchange.Net/Clients/RestApiClient.cs

@@ -92,7 +92,7 @@ namespace CryptoExchange.Net
                 }
 
                 // Calculate time offset between local and server
-                var offset = result.Data - localTime;
+                var offset = result.Data - (localTime.AddMilliseconds(result.ResponseTime!.Value.TotalMilliseconds / 2));
                 timeSyncParams.UpdateTimeOffset(offset);
                 timeSyncParams.TimeSyncState.Semaphore.Release();                
             }

CryptoExchange.Net/Converters/DateTimeConverter.cs

@@ -1,5 +1,4 @@
 using Newtonsoft.Json;
-using Newtonsoft.Json.Linq;
 using System;
 using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
@@ -34,11 +33,11 @@ namespace CryptoExchange.Net.Converters
                 var longValue = (long)reader.Value;
                 if (longValue == 0)
                     return objectType == typeof(DateTime) ? default(DateTime): null;
-                if (longValue < 1999999999)
+                if (longValue < 19999999999)
                     return ConvertFromSeconds(longValue);
-                if (longValue < 1999999999999)
+                if (longValue < 19999999999999)
                     return ConvertFromMilliseconds(longValue);
-                if (longValue < 1999999999999999)
+                if (longValue < 19999999999999999)
                     return ConvertFromMicroseconds(longValue);
                 
                 return ConvertFromNanoseconds(longValue);
@@ -46,7 +45,7 @@ namespace CryptoExchange.Net.Converters
             else if (reader.TokenType is JsonToken.Float)
             {
                 var doubleValue = (double)reader.Value;
-                if (doubleValue < 1999999999)
+                if (doubleValue < 19999999999)
                     return ConvertFromSeconds(doubleValue);
                 
                 return ConvertFromMilliseconds(doubleValue);
@@ -86,11 +85,11 @@ namespace CryptoExchange.Net.Converters
                 if (double.TryParse(stringValue, NumberStyles.Float, CultureInfo.InvariantCulture, out var doubleValue))
                 {
                     // Parse 1637745563.000 format
-                    if (doubleValue < 1999999999)
+                    if (doubleValue < 19999999999)
                         return ConvertFromSeconds(doubleValue);
-                    if (doubleValue < 1999999999999)
+                    if (doubleValue < 19999999999999)
                         return ConvertFromMilliseconds((long)doubleValue);
-                    if (doubleValue < 1999999999999999)
+                    if (doubleValue < 19999999999999999)
                         return ConvertFromMicroseconds((long)doubleValue);
 
                     return ConvertFromNanoseconds((long)doubleValue);

CryptoExchange.Net/CryptoExchange.Net.csproj

@@ -6,16 +6,16 @@
     <PackageId>CryptoExchange.Net</PackageId>
     <Authors>JKorf</Authors>
     <Description>A base package for implementing cryptocurrency API's</Description>
-    <PackageVersion>5.1.6</PackageVersion>
-    <AssemblyVersion>5.1.6</AssemblyVersion>
-    <FileVersion>5.1.6</FileVersion>
+    <PackageVersion>5.1.9</PackageVersion>
+    <AssemblyVersion>5.1.9</AssemblyVersion>
+    <FileVersion>5.1.9</FileVersion>
     <PackageRequireLicenseAcceptance>false</PackageRequireLicenseAcceptance>
     <RepositoryType>git</RepositoryType>
     <RepositoryUrl>https://github.com/JKorf/CryptoExchange.Net.git</RepositoryUrl>
     <PackageProjectUrl>https://github.com/JKorf/CryptoExchange.Net</PackageProjectUrl>
     <NeutralLanguage>en</NeutralLanguage>
     <GeneratePackageOnBuild>true</GeneratePackageOnBuild>
-    <PackageReleaseNotes>5.1.6 - Updated EnumConverter to properly handle emtpy/null and default values</PackageReleaseNotes>
+    <PackageReleaseNotes>5.1.9 - Added latency to the timesync calculation, Small fix for exception in socket close handling</PackageReleaseNotes>
     <Nullable>enable</Nullable>
     <LangVersion>9.0</LangVersion>
     <PackageLicenseExpression>MIT</PackageLicenseExpression>

CryptoExchange.Net/ExtensionMethods.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
@@ -491,7 +490,6 @@ namespace CryptoExchange.Net
 
             return ub.Uri;
         }
-
     }
 }
 

CryptoExchange.Net/Interfaces/ISocketClient.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Threading.Tasks;
+using CryptoExchange.Net.Authentication;
 using CryptoExchange.Net.Objects;
 using CryptoExchange.Net.Sockets;
 
@@ -15,6 +16,12 @@ namespace CryptoExchange.Net.Interfaces
         /// </summary>
         BaseSocketClientOptions ClientOptions { get; }
 
+        /// <summary>
+        /// Set the API credentials for this client. All Api clients in this client will use the new credentials, regardless of earlier set options.
+        /// </summary>
+        /// <param name="credentials">The credentials to set</param>
+        void SetApiCredentials(ApiCredentials credentials);
+
         /// <summary>
         /// Incoming kilobytes per second of data
         /// </summary>

CryptoExchange.Net/Interfaces/IWebsocket.cs

@@ -41,10 +41,6 @@ namespace CryptoExchange.Net.Interfaces
         /// </summary>
         Encoding? Encoding { get; set; }
         /// <summary>
-        /// Whether socket is in the process of reconnecting
-        /// </summary>
-        bool Reconnecting { get; set; }
-        /// <summary>
         /// The max amount of outgoing messages per second
         /// </summary>
         int? RatelimitPerSecond { get; set; }
@@ -61,9 +57,9 @@ namespace CryptoExchange.Net.Interfaces
         /// </summary>
         Func<string, string>? DataInterpreterString { get; set; }
         /// <summary>
-        /// The url the socket connects to
+        /// The uri the socket connects to
         /// </summary>
-        string Url { get; }
+        Uri Uri { get; }
         /// <summary>
         /// Whether the socket connection is closed
         /// </summary>
@@ -89,7 +85,12 @@ namespace CryptoExchange.Net.Interfaces
         /// Connect the socket
         /// </summary>
         /// <returns></returns>
-        Task<bool> ConnectAsync();
+        Task<bool> ConnectAsync();        
+        /// <summary>
+        /// Receive and send messages over the connection. Resulting task should complete when closing the socket.
+        /// </summary>
+        /// <returns></returns>
+        Task ProcessAsync();
         /// <summary>
         /// Send data
         /// </summary>

CryptoExchange.Net/OrderBook/SymbolOrderBook.cs

@@ -483,7 +483,7 @@ namespace CryptoExchange.Net.OrderBook
         /// <param name="timeout">Max wait time</param>
         /// <param name="ct">Cancellation token</param>
         /// <returns></returns>
-        protected async Task<CallResult<bool>> WaitForSetOrderBookAsync(int timeout, CancellationToken ct)
+        protected async Task<CallResult<bool>> WaitForSetOrderBookAsync(TimeSpan timeout, CancellationToken ct)
         {
             var startWait = DateTime.UtcNow;
             while (!bookSet && Status == OrderBookStatus.Syncing)
@@ -491,12 +491,12 @@ namespace CryptoExchange.Net.OrderBook
                 if(ct.IsCancellationRequested)
                     return new CallResult<bool>(new CancellationRequestedError());
 
-                if ((DateTime.UtcNow - startWait).TotalMilliseconds > timeout)
+                if (DateTime.UtcNow - startWait > timeout)
                     return new CallResult<bool>(new ServerError("Timeout while waiting for data"));
 
                 try
                 {
-                    await Task.Delay(10, ct).ConfigureAwait(false);
+                    await Task.Delay(50, ct).ConfigureAwait(false);
                 }
                 catch (OperationCanceledException)
                 { }

CryptoExchange.Net/Sockets/CryptoExchangeWebSocketClient.cs

@@ -23,20 +23,14 @@ namespace CryptoExchange.Net.Sockets
     public class CryptoExchangeWebSocketClient : IWebsocket
     {
         internal static int lastStreamId;
-        private static readonly object streamIdLock = new object();
+        private static readonly object streamIdLock = new();
 
         private ClientWebSocket _socket;
-        private Task? _sendTask;
-        private Task? _receiveTask;
-        private Task? _timeoutTask;
         private readonly AsyncResetEvent _sendEvent;
         private readonly ConcurrentQueue<byte[]> _sendBuffer;
         private readonly IDictionary<string, string> cookies;
         private readonly IDictionary<string, string> headers;
         private CancellationTokenSource _ctsSource;
-        private bool _closing;
-        private bool _startedSent;
-        private bool _startedReceive;
 
         private readonly List<DateTime> _outgoingMessages;
         private DateTime _lastReceivedMessagesUpdate;
@@ -45,6 +39,7 @@ namespace CryptoExchange.Net.Sockets
         /// Received messages, the size and the timstamp
         /// </summary>
         protected readonly List<ReceiveItem> _receivedMessages;
+
         /// <summary>
         /// Received messages lock
         /// </summary>
@@ -58,19 +53,19 @@ namespace CryptoExchange.Net.Sockets
         /// <summary>
         /// Handlers for when an error happens on the socket
         /// </summary>
-        protected readonly List<Action<Exception>> errorHandlers = new List<Action<Exception>>();
+        protected readonly List<Action<Exception>> errorHandlers = new();
         /// <summary>
         /// Handlers for when the socket connection is opened
         /// </summary>
-        protected readonly List<Action> openHandlers = new List<Action>();
+        protected readonly List<Action> openHandlers = new();
         /// <summary>
         /// Handlers for when the connection is closed
         /// </summary>
-        protected readonly List<Action> closeHandlers = new List<Action>();
+        protected readonly List<Action> closeHandlers = new();
         /// <summary>
         /// Handlers for when a message is received
         /// </summary>
-        protected readonly List<Action<string>> messageHandlers = new List<Action<string>>();
+        protected readonly List<Action<string>> messageHandlers = new();
 
         /// <inheritdoc />
         public int Id { get; }
@@ -78,9 +73,6 @@ namespace CryptoExchange.Net.Sockets
         /// <inheritdoc />
         public string? Origin { get; set; }
 
-        /// <inheritdoc />
-        public bool Reconnecting { get; set; }
-
         /// <summary>
         /// The timestamp this socket has been active for the last time
         /// </summary>
@@ -97,13 +89,13 @@ namespace CryptoExchange.Net.Sockets
         public Func<string, string>? DataInterpreterString { get; set; }
 
         /// <inheritdoc />
-        public string Url { get; }
+        public Uri Uri { get; }
 
         /// <inheritdoc />
         public bool IsClosed => _socket.State == WebSocketState.Closed;
 
         /// <inheritdoc />
-        public bool IsOpen => _socket.State == WebSocketState.Open && !_closing;
+        public bool IsOpen => _socket.State == WebSocketState.Open && !_ctsSource.IsCancellationRequested;
 
         /// <summary>
         /// Ssl protocols supported. NOT USED BY THIS IMPLEMENTATION
@@ -179,8 +171,8 @@ namespace CryptoExchange.Net.Sockets
         /// ctor
         /// </summary>
         /// <param name="log">The log object to use</param>
-        /// <param name="url">The url the socket should connect to</param>
-        public CryptoExchangeWebSocketClient(Log log, string url) : this(log, url, new Dictionary<string, string>(), new Dictionary<string, string>())
+        /// <param name="uri">The uri the socket should connect to</param>
+        public CryptoExchangeWebSocketClient(Log log, Uri uri) : this(log, uri, new Dictionary<string, string>(), new Dictionary<string, string>())
         {
         }
 
@@ -188,14 +180,14 @@ namespace CryptoExchange.Net.Sockets
         /// ctor
         /// </summary>
         /// <param name="log">The log object to use</param>
-        /// <param name="url">The url the socket should connect to</param>
+        /// <param name="uri">The uri the socket should connect to</param>
         /// <param name="cookies">Cookies to sent in the socket connection request</param>
         /// <param name="headers">Headers to sent in the socket connection request</param>
-        public CryptoExchangeWebSocketClient(Log log, string url, IDictionary<string, string> cookies, IDictionary<string, string> headers)
+        public CryptoExchangeWebSocketClient(Log log, Uri uri, IDictionary<string, string> cookies, IDictionary<string, string> headers)
         {
             Id = NextStreamId();
             this.log = log;
-            Url = url;
+            Uri = uri;
             this.cookies = cookies;
             this.headers = headers;
 
@@ -212,7 +204,16 @@ namespace CryptoExchange.Net.Sockets
         /// <inheritdoc />
         public virtual void SetProxy(ApiProxy proxy)
         {
-            _socket.Options.Proxy = new WebProxy(proxy.Host, proxy.Port);
+            if (!Uri.TryCreate($"{proxy.Host}:{proxy.Port}", UriKind.Absolute, out var uri))
+                throw new ArgumentException("Proxy settings invalid, {proxy.Host}:{proxy.Port} not a valid URI", nameof(proxy));
+
+            _socket.Options.Proxy = uri?.Scheme == null
+                ? _socket.Options.Proxy = new WebProxy(proxy.Host, proxy.Port)
+                : _socket.Options.Proxy = new WebProxy
+                {
+                    Address = uri
+                };
+
             if (proxy.Login != null)
                 _socket.Options.Proxy.Credentials = new NetworkCredential(proxy.Login, proxy.Password);
         }
@@ -223,8 +224,8 @@ namespace CryptoExchange.Net.Sockets
             log.Write(LogLevel.Debug, $"Socket {Id} connecting");
             try
             {
-                using CancellationTokenSource tcs = new CancellationTokenSource(TimeSpan.FromSeconds(10));                
-                await _socket.ConnectAsync(new Uri(Url), tcs.Token).ConfigureAwait(false);
+                using CancellationTokenSource tcs = new(TimeSpan.FromSeconds(10));                
+                await _socket.ConnectAsync(Uri, tcs.Token).ConfigureAwait(false);
                 
                 Handle(openHandlers);
             }
@@ -233,35 +234,27 @@ namespace CryptoExchange.Net.Sockets
                 log.Write(LogLevel.Debug, $"Socket {Id} connection failed: " + e.ToLogString());
                 return false;
             }
-
-            log.Write(LogLevel.Trace, $"Socket {Id} connection succeeded, starting communication");
-            _sendTask = Task.Factory.StartNew(SendLoopAsync, default, TaskCreationOptions.LongRunning | TaskCreationOptions.DenyChildAttach, TaskScheduler.Default).Unwrap();
-            _receiveTask = Task.Factory.StartNew(ReceiveLoopAsync, default, TaskCreationOptions.LongRunning | TaskCreationOptions.DenyChildAttach, TaskScheduler.Default).Unwrap();
-            if (Timeout != default)
-                _timeoutTask = Task.Run(CheckTimeoutAsync);
-
-            var sw = Stopwatch.StartNew();
-            while (!_startedSent || !_startedReceive)
-            {
-                // Wait for the tasks to have actually started
-                await Task.Delay(10).ConfigureAwait(false);
-
-                if(sw.ElapsedMilliseconds > 5000)
-                {
-                    _ = _socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "", default);
-                    log.Write(LogLevel.Debug, $"Socket {Id} startup interupted");
-                    return false;
-                }    
-            }
-
-            log.Write(LogLevel.Debug, $"Socket {Id} connected to {Url}");
+    
+            log.Write(LogLevel.Debug, $"Socket {Id} connected to {Uri}");
             return true;
         }
 
+        /// <inheritdoc />
+        public virtual async Task ProcessAsync()
+        {
+            log.Write(LogLevel.Trace, $"Socket {Id} ProcessAsync started");
+            var sendTask = SendLoopAsync();
+            var receiveTask = ReceiveLoopAsync();
+            var timeoutTask = Timeout != default ? CheckTimeoutAsync() : Task.CompletedTask;
+            log.Write(LogLevel.Trace, $"Socket {Id} processing startup completed");
+            await Task.WhenAll(sendTask, receiveTask, timeoutTask).ConfigureAwait(false);
+            log.Write(LogLevel.Trace, $"Socket {Id} ProcessAsync finished");
+        }
+
         /// <inheritdoc />
         public virtual void Send(string data)
         {
-            if (_closing)
+            if (_ctsSource.IsCancellationRequested)
                 throw new InvalidOperationException($"Socket {Id} Can't send data when socket is not connected");
 
             var bytes = _encoding.GetBytes(data);
@@ -274,36 +267,27 @@ namespace CryptoExchange.Net.Sockets
         public virtual async Task CloseAsync()
         {
             log.Write(LogLevel.Debug, $"Socket {Id} closing");
-            await CloseInternalAsync(true, true).ConfigureAwait(false);
+            await CloseInternalAsync().ConfigureAwait(false);
         }
         
         /// <summary>
-        /// Internal close method, will wait for each task to complete to gracefully close
+        /// Internal close method
         /// </summary>
-        /// <param name="waitSend"></param>
-        /// <param name="waitReceive"></param>
         /// <returns></returns>
-        private async Task CloseInternalAsync(bool waitSend, bool waitReceive)
+        private async Task CloseInternalAsync()
         {
-            if (_closing)
-                return;
-
-            _closing = true;
-            var tasksToAwait = new List<Task>();
-            if (_socket.State == WebSocketState.Open)
-                tasksToAwait.Add(_socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "Closing", default));
-
             _ctsSource.Cancel();
             _sendEvent.Set();
-            if (waitSend)
-                tasksToAwait.Add(_sendTask!);
-            if (waitReceive)
-                tasksToAwait.Add(_receiveTask!);
-            if (_timeoutTask != null)
-                tasksToAwait.Add(_timeoutTask);
 
-            log.Write(LogLevel.Trace, $"Socket {Id} waiting for communication loops to finish");
-            await Task.WhenAll(tasksToAwait).ConfigureAwait(false);
+            if (_socket.State == WebSocketState.Open)
+            {
+                try
+                {
+                    await _socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "Closing", default).ConfigureAwait(false);
+                }
+                catch(Exception)
+                { } // Can sometimes throw an exception when socket is in aborted state due to timing
+            }
             log.Write(LogLevel.Debug, $"Socket {Id} closed");
             Handle(closeHandlers);
         }
@@ -329,7 +313,6 @@ namespace CryptoExchange.Net.Sockets
         {
             log.Write(LogLevel.Debug, $"Socket {Id} resetting");
             _ctsSource = new CancellationTokenSource();
-            _closing = false;
 
             while (_sendBuffer.TryDequeue(out _)) { } // Clear send buffer
 
@@ -360,17 +343,16 @@ namespace CryptoExchange.Net.Sockets
         /// <returns></returns>
         private async Task SendLoopAsync()
         {
-            _startedSent = true;
             try
             {
                 while (true)
                 {
-                    if (_closing)
+                    if (_ctsSource.IsCancellationRequested)
                         break;
 
                     await _sendEvent.WaitAsync().ConfigureAwait(false);
 
-                    if (_closing)
+                    if (_ctsSource.IsCancellationRequested)
                         break;
 
                     while (_sendBuffer.TryDequeue(out var data))
@@ -382,7 +364,7 @@ namespace CryptoExchange.Net.Sockets
                             while (MessagesSentLastSecond() >= RatelimitPerSecond)
                             {
                                 start ??= DateTime.UtcNow;
-                                await Task.Delay(10).ConfigureAwait(false);
+                                await Task.Delay(50).ConfigureAwait(false);
                             }
 
                             if (start != null)
@@ -404,7 +386,7 @@ namespace CryptoExchange.Net.Sockets
                         {
                             // Connection closed unexpectedly, .NET framework                      
                             Handle(errorHandlers, ioe);
-                            _ = Task.Run(async () => await CloseInternalAsync(false, true).ConfigureAwait(false));
+                            await CloseInternalAsync().ConfigureAwait(false);
                             break;
                         }
                     }
@@ -421,7 +403,6 @@ namespace CryptoExchange.Net.Sockets
             finally
             {
                 log.Write(LogLevel.Trace, $"Socket {Id} Send loop finished");
-                _startedSent = false;
             }
         }
 
@@ -431,14 +412,13 @@ namespace CryptoExchange.Net.Sockets
         /// <returns></returns>
         private async Task ReceiveLoopAsync()
         {
-            _startedReceive = true;
             var buffer = new ArraySegment<byte>(new byte[65536]);
             var received = 0;
             try
             {
                 while (true)
                 {
-                    if (_closing)
+                    if (_ctsSource.IsCancellationRequested)
                         break;
 
                     MemoryStream? memoryStream = null;
@@ -462,7 +442,7 @@ namespace CryptoExchange.Net.Sockets
                         {
                             // Connection closed unexpectedly        
                             Handle(errorHandlers, wse);
-                            _ = Task.Run(async () => await CloseInternalAsync(true, true).ConfigureAwait(false));
+                            await CloseInternalAsync().ConfigureAwait(false);
                             break;
                         }
 
@@ -470,7 +450,7 @@ namespace CryptoExchange.Net.Sockets
                         {
                             // Connection closed unexpectedly        
                             log.Write(LogLevel.Debug, $"Socket {Id} received `Close` message");
-                            _ = Task.Run(async () => await CloseInternalAsync(true, true).ConfigureAwait(false));
+                            await CloseInternalAsync().ConfigureAwait(false);
                             break;
                         }
 
@@ -509,7 +489,7 @@ namespace CryptoExchange.Net.Sockets
                         break;
                     }
 
-                    if (receiveResult == null || _closing)
+                    if (receiveResult == null || _ctsSource.IsCancellationRequested)
                     {
                         // Error during receiving or cancellation requested, stop.
                         break;
@@ -541,7 +521,6 @@ namespace CryptoExchange.Net.Sockets
             finally
             {
                 log.Write(LogLevel.Trace, $"Socket {Id} Receive loop finished");
-                _startedReceive = false;
             }
         }
 
@@ -609,7 +588,7 @@ namespace CryptoExchange.Net.Sockets
             { 
                 while (true)
                 {
-                    if (_closing)
+                    if (_ctsSource.IsCancellationRequested)
                         return;
 
                     if (DateTime.UtcNow - LastActionTime > Timeout)

CryptoExchange.Net/Sockets/PendingRequest.cs

@@ -0,0 +1,47 @@
+using CryptoExchange.Net.Objects;
+using Newtonsoft.Json.Linq;
+using System;
+using System.Threading;
+
+namespace CryptoExchange.Net.Sockets
+{
+    internal class PendingRequest
+    {
+        public Func<JToken, bool> Handler { get; }
+        public JToken? Result { get; private set; }
+        public bool Completed { get; private set; }
+        public AsyncResetEvent Event { get; }
+        public TimeSpan Timeout { get; }
+
+        private CancellationTokenSource cts;
+
+        public PendingRequest(Func<JToken, bool> handler, TimeSpan timeout)
+        {
+            Handler = handler;
+            Event = new AsyncResetEvent(false, false);
+            Timeout = timeout;
+
+            cts = new CancellationTokenSource(timeout);
+            cts.Token.Register(Fail, false);
+        }
+
+        public bool CheckData(JToken data)
+        {
+            if (Handler(data))
+            {
+                Result = data;
+                Completed = true;
+                Event.Set();
+                return true;
+            }
+
+            return false;
+        }
+
+        public void Fail()
+        {
+            Completed = true;
+            Event.Set();
+        }
+    }
+}

CryptoExchange.Net/Sockets/SocketConnection.cs

@@ -65,12 +65,22 @@ namespace CryptoExchange.Net.Sockets
         /// <summary>
         /// If connection is made
         /// </summary>
-        public bool Connected { get; private set; }
+        public bool Connected => _socket.IsOpen;
 
         /// <summary>
-        /// The underlying websocket
+        /// The unique ID of the socket
         /// </summary>
-        public IWebsocket Socket { get; set; }
+        public int SocketId => _socket.Id;
+
+        /// <summary>
+        /// The current kilobytes per second of data being received, averaged over the last 3 seconds
+        /// </summary>
+        public double IncomingKbps => _socket.IncomingKbps;
+
+        /// <summary>
+        /// The connection uri
+        /// </summary>
+        public Uri Uri => _socket.Uri;
 
         /// <summary>
         /// The API client the connection is for
@@ -113,7 +123,7 @@ namespace CryptoExchange.Net.Sockets
                 if (pausedActivity != value)
                 {
                     pausedActivity = value;
-                    log.Write(LogLevel.Information, $"Socket {Socket.Id} Paused activity: " + value);
+                    log.Write(LogLevel.Information, $"Socket {SocketId} Paused activity: " + value);
                     if(pausedActivity) ActivityPaused?.Invoke();
                     else ActivityUnpaused?.Invoke();
                 }
@@ -122,13 +132,19 @@ namespace CryptoExchange.Net.Sockets
 
         private bool pausedActivity;
         private readonly List<SocketSubscription> subscriptions;
-        private readonly object subscriptionLock = new object();
+        private readonly object subscriptionLock = new();
 
         private bool lostTriggered;
         private readonly Log log;
         private readonly BaseSocketClient socketClient;
 
         private readonly List<PendingRequest> pendingRequests;
+        private Task? _socketProcessReconnectTask;
+
+        /// <summary>
+        /// The underlying websocket
+        /// </summary>
+        private readonly IWebsocket _socket;
 
         /// <summary>
         /// New socket connection
@@ -145,14 +161,244 @@ namespace CryptoExchange.Net.Sockets
             pendingRequests = new List<PendingRequest>();
 
             subscriptions = new List<SocketSubscription>();
-            Socket = socket;
+            _socket = socket;
 
-            Socket.Timeout = client.ClientOptions.SocketNoDataTimeout;
-            Socket.OnMessage += ProcessMessage;
-            Socket.OnClose += SocketOnClose;
-            Socket.OnOpen += SocketOnOpen;
+            _socket.Timeout = client.ClientOptions.SocketNoDataTimeout;
+            _socket.OnMessage += ProcessMessage;
+            _socket.OnOpen += SocketOnOpen;
         }
         
+        /// <summary>
+        /// Connect the websocket and start processing
+        /// </summary>
+        /// <returns></returns>
+        public async Task<bool> ConnectAsync()
+        {
+            var connected = await _socket.ConnectAsync().ConfigureAwait(false);
+            if (connected)
+                StartProcessingTask();
+
+            return connected;
+        }
+
+        /// <summary>
+        /// Retrieve the underlying socket
+        /// </summary>
+        /// <returns></returns>
+        public IWebsocket GetSocket()
+        {
+            return _socket;
+        }
+
+        /// <summary>
+        /// Trigger a reconnect of the socket connection
+        /// </summary>
+        /// <returns></returns>
+        public async Task TriggerReconnectAsync()
+        {
+            await _socket.CloseAsync().ConfigureAwait(false);
+        }
+
+        /// <summary>
+        /// Close the connection
+        /// </summary>
+        /// <returns></returns>
+        public async Task CloseAsync()
+        {
+            ShouldReconnect = false;
+            if (socketClient.socketConnections.ContainsKey(SocketId))
+                socketClient.socketConnections.TryRemove(SocketId, out _);
+
+            lock (subscriptionLock)
+            {
+                foreach (var subscription in subscriptions)
+                {
+                    if (subscription.CancellationTokenRegistration.HasValue)
+                        subscription.CancellationTokenRegistration.Value.Dispose();
+                }
+            }
+            await _socket.CloseAsync().ConfigureAwait(false);
+
+            if (_socketProcessReconnectTask != null)
+                await _socketProcessReconnectTask.ConfigureAwait(false);
+
+            _socket.Dispose();
+        }
+
+        /// <summary>
+        /// Close a subscription on this connection. If all subscriptions on this connection are closed the connection gets closed as well
+        /// </summary>
+        /// <param name="subscription">Subscription to close</param>
+        /// <returns></returns>
+        public async Task CloseAsync(SocketSubscription subscription)
+        {
+            if (!_socket.IsOpen)
+                return;
+
+            if (subscription.CancellationTokenRegistration.HasValue)
+                subscription.CancellationTokenRegistration.Value.Dispose();
+
+            if (subscription.Confirmed)
+                await socketClient.UnsubscribeAsync(this, subscription).ConfigureAwait(false);
+
+            bool shouldCloseConnection;
+            lock (subscriptionLock)
+                shouldCloseConnection = !subscriptions.Any(r => r.UserSubscription && subscription != r);
+
+            if (shouldCloseConnection)
+                await CloseAsync().ConfigureAwait(false);
+
+            lock (subscriptionLock)
+                subscriptions.Remove(subscription);
+        }
+
+        private void StartProcessingTask()
+        {
+            log.Write(LogLevel.Trace, $"Starting {SocketId} process/reconnect task");
+            _socketProcessReconnectTask = Task.Run(async () =>
+            {
+                await _socket.ProcessAsync().ConfigureAwait(false);
+                await ReconnectAsync().ConfigureAwait(false);
+                log.Write(LogLevel.Trace, $"Process/reconnect {SocketId} task finished");
+            });
+        }
+
+        private async Task ReconnectAsync()
+        {
+            // Fail all pending requests
+            lock (pendingRequests)
+            {
+                foreach (var pendingRequest in pendingRequests.ToList())
+                {
+                    pendingRequest.Fail();
+                    pendingRequests.Remove(pendingRequest);
+                }
+            }
+
+            if (socketClient.ClientOptions.AutoReconnect && ShouldReconnect)
+            {
+                // Should reconnect
+                DisconnectTime = DateTime.UtcNow;
+                log.Write(LogLevel.Warning, $"Socket {SocketId} Connection lost, will try to reconnect");
+                if (!lostTriggered)
+                {
+                    lostTriggered = true;
+                    ConnectionLost?.Invoke();
+                }
+
+                while (ShouldReconnect)
+                {
+                    if (ReconnectTry > 0)
+                    {
+                        // Wait a bit before attempting reconnect
+                        await Task.Delay(socketClient.ClientOptions.ReconnectInterval).ConfigureAwait(false);
+                    }
+
+                    if (!ShouldReconnect)
+                    {
+                        // Should reconnect changed to false while waiting to reconnect
+                        return;
+                    }
+
+                    _socket.Reset();
+                    if (!await _socket.ConnectAsync().ConfigureAwait(false))
+                    {
+                        // Reconnect failed
+                        ReconnectTry++;
+                        ResubscribeTry = 0;
+                        if (socketClient.ClientOptions.MaxReconnectTries != null
+                        && ReconnectTry >= socketClient.ClientOptions.MaxReconnectTries)
+                        {
+                            log.Write(LogLevel.Warning, $"Socket {SocketId} failed to reconnect after {ReconnectTry} tries, closing");
+                            ShouldReconnect = false;
+
+                            if (socketClient.socketConnections.ContainsKey(SocketId))
+                                socketClient.socketConnections.TryRemove(SocketId, out _);
+
+                            _ = Task.Run(() => ConnectionClosed?.Invoke());
+                            // Reached max tries, break loop and leave connection closed
+                            break;
+                        }
+
+                        // Continue to try again
+                        log.Write(LogLevel.Debug, $"Socket {SocketId} failed to reconnect{(socketClient.ClientOptions.MaxReconnectTries != null ? $", try {ReconnectTry}/{socketClient.ClientOptions.MaxReconnectTries}" : "")}, will try again in {socketClient.ClientOptions.ReconnectInterval}");
+                        continue;
+                    }
+
+                    // Successfully reconnected, start processing
+                    StartProcessingTask();
+
+                    ReconnectTry = 0;
+                    var time = DisconnectTime;
+                    DisconnectTime = null;
+
+                    log.Write(LogLevel.Information, $"Socket {SocketId} reconnected after {DateTime.UtcNow - time}");
+
+                    var reconnectResult = await ProcessReconnectAsync().ConfigureAwait(false);
+                    if (!reconnectResult)
+                    {
+                        // Failed to resubscribe everything
+                        ResubscribeTry++;
+                        DisconnectTime = time;
+
+                        if (socketClient.ClientOptions.MaxResubscribeTries != null &&
+                        ResubscribeTry >= socketClient.ClientOptions.MaxResubscribeTries)
+                        {
+                            log.Write(LogLevel.Warning, $"Socket {SocketId} failed to resubscribe after {ResubscribeTry} tries, closing. Last resubscription error: {reconnectResult.Error}");
+                            ShouldReconnect = false;
+
+                            if (socketClient.socketConnections.ContainsKey(SocketId))
+                                socketClient.socketConnections.TryRemove(SocketId, out _);
+
+                            _ = Task.Run(() => ConnectionClosed?.Invoke());
+                        }
+                        else
+                            log.Write(LogLevel.Debug, $"Socket {SocketId} resubscribing all subscriptions failed on reconnected socket{(socketClient.ClientOptions.MaxResubscribeTries != null ? $", try {ResubscribeTry}/{socketClient.ClientOptions.MaxResubscribeTries}" : "")}. Error: {reconnectResult.Error}. Disconnecting and reconnecting.");
+
+                        // Failed resubscribe, close socket if it is still open
+                        if (_socket.IsOpen)
+                            await _socket.CloseAsync().ConfigureAwait(false);
+                        else
+                            DisconnectTime = DateTime.UtcNow;
+
+                        // Break out of the loop, the new processing task should reconnect again
+                        break;
+                    }
+                    else
+                    {
+                        // Succesfully reconnected
+                        log.Write(LogLevel.Information, $"Socket {SocketId} data connection restored.");
+                        ResubscribeTry = 0;
+                        if (lostTriggered)
+                        {
+                            lostTriggered = false;
+                            _ = Task.Run(() => ConnectionRestored?.Invoke(time.HasValue ? DateTime.UtcNow - time.Value : TimeSpan.FromSeconds(0))).ConfigureAwait(false);
+                        }
+
+                        break;
+                    }
+                }
+            }
+            else
+            {
+                if (!socketClient.ClientOptions.AutoReconnect && ShouldReconnect)
+                    _ = Task.Run(() => ConnectionClosed?.Invoke());
+
+                // No reconnecting needed
+                log.Write(LogLevel.Information, $"Socket {SocketId} closed");
+                if (socketClient.socketConnections.ContainsKey(SocketId))
+                    socketClient.socketConnections.TryRemove(SocketId, out _);
+            }
+        }
+
+        /// <summary>
+        /// Dispose the connection
+        /// </summary>
+        public void Dispose()
+        {
+            _socket.Dispose();
+        }
+
         /// <summary>
         /// Process a message received by the socket
         /// </summary>
@@ -160,7 +406,7 @@ namespace CryptoExchange.Net.Sockets
         private void ProcessMessage(string data)
         {
             var timestamp = DateTime.UtcNow;
-            log.Write(LogLevel.Trace, $"Socket {Socket.Id} received data: " + data);
+            log.Write(LogLevel.Trace, $"Socket {SocketId} received data: " + data);
             if (string.IsNullOrEmpty(data)) return;
 
             var tokenData = data.ToJToken(log);
@@ -202,12 +448,20 @@ namespace CryptoExchange.Net.Sockets
 
             // Message was not a request response, check data handlers
             var messageEvent = new MessageEvent(this, tokenData, socketClient.ClientOptions.OutputOriginalData ? data: null, timestamp);
-            if (!HandleData(messageEvent) && !handledResponse)
+            var (handled, userProcessTime) = HandleData(messageEvent);
+            if (!handled && !handledResponse)
             {
                 if (!socketClient.UnhandledMessageExpected)
-                    log.Write(LogLevel.Warning, $"Socket {Socket.Id} Message not handled: " + tokenData);
+                    log.Write(LogLevel.Warning, $"Socket {SocketId} Message not handled: " + tokenData);
                 UnhandledMessage?.Invoke(tokenData);
             }
+
+            var total = DateTime.UtcNow - timestamp;
+            if (userProcessTime.TotalMilliseconds > 500)
+                log.Write(LogLevel.Debug, $"Socket {SocketId} message processing slow ({(int)total.TotalMilliseconds}ms), consider offloading data handling to another thread. " +
+                                                "Data from this socket may arrive late or not at all if message processing is continuously slow.");
+            
+            log.Write(LogLevel.Trace, $"Socket {SocketId} message processed in {(int)total.TotalMilliseconds}ms, ({(int)userProcessTime.TotalMilliseconds}ms user code)");
         }
 
         /// <summary>
@@ -246,13 +500,13 @@ namespace CryptoExchange.Net.Sockets
         /// </summary>
         /// <param name="messageEvent"></param>
         /// <returns>True if the data was successfully handled</returns>
-        private bool HandleData(MessageEvent messageEvent)
+        private (bool, TimeSpan) HandleData(MessageEvent messageEvent)
         {
             SocketSubscription? currentSubscription = null;
             try
             { 
                 var handled = false;
-                var sw = Stopwatch.StartNew();
+                TimeSpan userCodeDuration = TimeSpan.Zero;
 
                 // Loop the subscriptions to check if any of them signal us that the message is for them
                 List<SocketSubscription> subscriptionsCopy;
@@ -267,7 +521,10 @@ namespace CryptoExchange.Net.Sockets
                         if (socketClient.MessageMatchesHandler(this, messageEvent.JsonData, subscription.Identifier!))
                         {
                             handled = true;
+                            var userSw = Stopwatch.StartNew();
                             subscription.MessageHandler(messageEvent);
+                            userSw.Stop();
+                            userCodeDuration = userSw.Elapsed;
                         }
                     }
                     else
@@ -276,24 +533,21 @@ namespace CryptoExchange.Net.Sockets
                         {
                             handled = true;
                             messageEvent.JsonData = socketClient.ProcessTokenData(messageEvent.JsonData);
+                            var userSw = Stopwatch.StartNew();
                             subscription.MessageHandler(messageEvent);
+                            userSw.Stop();
+                            userCodeDuration = userSw.Elapsed;
                         }
                     }
                 }
-                
-                sw.Stop();
-                if (sw.ElapsedMilliseconds > 500)
-                    log.Write(LogLevel.Debug, $"Socket {Socket.Id} message processing slow ({sw.ElapsedMilliseconds}ms), consider offloading data handling to another thread. " +
-                                                    "Data from this socket may arrive late or not at all if message processing is continuously slow.");
-                else
-                    log.Write(LogLevel.Trace, $"Socket {Socket.Id} message processed in {sw.ElapsedMilliseconds}ms");
-                return handled;
+                               
+                return (handled, userCodeDuration);
             }
             catch (Exception ex)
             {
-                log.Write(LogLevel.Error, $"Socket {Socket.Id} Exception during message processing\r\nException: {ex.ToLogString()}\r\nData: {messageEvent.JsonData}");
+                log.Write(LogLevel.Error, $"Socket {SocketId} Exception during message processing\r\nException: {ex.ToLogString()}\r\nData: {messageEvent.JsonData}");
                 currentSubscription?.InvokeExceptionHandler(ex);
-                return false;
+                return (false, TimeSpan.Zero);
             }
         }
 
@@ -312,7 +566,10 @@ namespace CryptoExchange.Net.Sockets
             {
                 pendingRequests.Add(pending);
             }
-            Send(obj);
+            var sendOk = Send(obj);
+            if(!sendOk)            
+                pending.Fail();            
+
             return pending.Event.WaitAsync(timeout);
         }
 
@@ -322,22 +579,30 @@ namespace CryptoExchange.Net.Sockets
         /// <typeparam name="T">The type of the object to send</typeparam>
         /// <param name="obj">The object to send</param>
         /// <param name="nullValueHandling">How null values should be serialized</param>
-        public virtual void Send<T>(T obj, NullValueHandling nullValueHandling = NullValueHandling.Ignore)
+        public virtual bool Send<T>(T obj, NullValueHandling nullValueHandling = NullValueHandling.Ignore)
         {
             if(obj is string str)
-                Send(str);
+                return Send(str);
             else
-                Send(JsonConvert.SerializeObject(obj, Formatting.None, new JsonSerializerSettings { NullValueHandling = nullValueHandling }));
+                return Send(JsonConvert.SerializeObject(obj, Formatting.None, new JsonSerializerSettings { NullValueHandling = nullValueHandling }));
         }
 
         /// <summary>
         /// Send string data over the websocket connection
         /// </summary>
         /// <param name="data">The data to send</param>
-        public virtual void Send(string data)
+        public virtual bool Send(string data)
         {
-            log.Write(LogLevel.Trace, $"Socket {Socket.Id} sending data: {data}");
-            Socket.Send(data);
+            log.Write(LogLevel.Trace, $"Socket {SocketId} sending data: {data}");
+            try
+            {
+                _socket.Send(data);
+                return true;
+            }
+            catch(Exception)
+            {
+                return false;
+            }
         }
 
         /// <summary>
@@ -347,140 +612,11 @@ namespace CryptoExchange.Net.Sockets
         {
             ReconnectTry = 0;
             PausedActivity = false;
-            Connected = true;
-        }
-
-        /// <summary>
-        /// Handler for a socket closing. Reconnects the socket if needed, or removes it from the active socket list if not
-        /// </summary>
-        protected virtual void SocketOnClose()
-        {
-            lock (pendingRequests)
-            {
-                foreach(var pendingRequest in pendingRequests.ToList())
-                {
-                    pendingRequest.Fail();
-                    pendingRequests.Remove(pendingRequest);
-                }
-            }
-
-            if (socketClient.ClientOptions.AutoReconnect && ShouldReconnect)
-            {
-                if (Socket.Reconnecting)
-                    return; // Already reconnecting
-
-                Socket.Reconnecting = true;
-
-                DisconnectTime = DateTime.UtcNow;
-                log.Write(LogLevel.Warning, $"Socket {Socket.Id} Connection lost, will try to reconnect");
-                if (!lostTriggered)
-                {
-                    lostTriggered = true;
-                    ConnectionLost?.Invoke();
-                }
-
-                Task.Run(async () =>
-                {
-                    while (ShouldReconnect)
-                    {
-                        if (ReconnectTry > 0)
-                        {
-                            // Wait a bit before attempting reconnect
-                            await Task.Delay(socketClient.ClientOptions.ReconnectInterval).ConfigureAwait(false);
-                        }
-
-                        if (!ShouldReconnect)
-                        {
-                            // Should reconnect changed to false while waiting to reconnect
-                            Socket.Reconnecting = false;
-                            return;
-                        }
-
-                        Socket.Reset();
-                        if (!await Socket.ConnectAsync().ConfigureAwait(false))
-                        {
-                            ReconnectTry++;
-                            ResubscribeTry = 0;
-                            if (socketClient.ClientOptions.MaxReconnectTries != null
-                            && ReconnectTry >= socketClient.ClientOptions.MaxReconnectTries)
-                            {
-                                log.Write(LogLevel.Warning, $"Socket {Socket.Id} failed to reconnect after {ReconnectTry} tries, closing");
-                                ShouldReconnect = false;
-
-                                if (socketClient.sockets.ContainsKey(Socket.Id))
-                                    socketClient.sockets.TryRemove(Socket.Id, out _);
-
-                                _ = Task.Run(() => ConnectionClosed?.Invoke());
-                                break;
-                            }
-
-                            log.Write(LogLevel.Debug, $"Socket {Socket.Id} failed to reconnect{(socketClient.ClientOptions.MaxReconnectTries != null ? $", try {ReconnectTry}/{socketClient.ClientOptions.MaxReconnectTries}": "")}, will try again in {socketClient.ClientOptions.ReconnectInterval}");
-                            continue;
-                        }
-
-                        // Successfully reconnected
-                        var time = DisconnectTime;
-                        DisconnectTime = null;
-
-                        log.Write(LogLevel.Information, $"Socket {Socket.Id} reconnected after {DateTime.UtcNow - time}");
-
-                        var reconnectResult = await ProcessReconnectAsync().ConfigureAwait(false);
-                        if (!reconnectResult)
-                        {
-                            ResubscribeTry++;
-                            DisconnectTime = time;
-
-                            if (socketClient.ClientOptions.MaxResubscribeTries != null &&
-                            ResubscribeTry >= socketClient.ClientOptions.MaxResubscribeTries)
-                            {
-                                log.Write(LogLevel.Warning, $"Socket {Socket.Id} failed to resubscribe after {ResubscribeTry} tries, closing. Last resubscription error: {reconnectResult.Error}");
-                                ShouldReconnect = false;
-
-                                if (socketClient.sockets.ContainsKey(Socket.Id))
-                                    socketClient.sockets.TryRemove(Socket.Id, out _);
-
-                                _ = Task.Run(() => ConnectionClosed?.Invoke());
-                            }
-                            else
-                                log.Write(LogLevel.Debug, $"Socket {Socket.Id} resubscribing all subscriptions failed on reconnected socket{(socketClient.ClientOptions.MaxResubscribeTries != null ? $", try {ResubscribeTry}/{socketClient.ClientOptions.MaxResubscribeTries}" : "")}. Error: {reconnectResult.Error}. Disconnecting and reconnecting.");
-
-                            if (Socket.IsOpen)                            
-                                await Socket.CloseAsync().ConfigureAwait(false);                            
-                            else
-                                DisconnectTime = DateTime.UtcNow;
-                        }
-                        else
-                        {
-                            log.Write(LogLevel.Information, $"Socket {Socket.Id} data connection restored.");
-                            ResubscribeTry = 0;
-                            if (lostTriggered)
-                            {
-                                lostTriggered = false;
-                                _ = Task.Run(() => ConnectionRestored?.Invoke(time.HasValue ? DateTime.UtcNow - time.Value : TimeSpan.FromSeconds(0))).ConfigureAwait(false);
-                            }
-
-                            break;
-                        }
-                    }
-
-                    Socket.Reconnecting = false;
-                });
-            }
-            else
-            {
-                if (!socketClient.ClientOptions.AutoReconnect && ShouldReconnect)
-                    _ = Task.Run(() => ConnectionClosed?.Invoke());
-
-                // No reconnecting needed
-                log.Write(LogLevel.Information, $"Socket {Socket.Id} closed");
-                if (socketClient.sockets.ContainsKey(Socket.Id))
-                    socketClient.sockets.TryRemove(Socket.Id, out _);
-            }
         }
 
         private async Task<CallResult<bool>> ProcessReconnectAsync()
         {
-            if (!Socket.IsOpen)
+            if (!_socket.IsOpen)
                 return new CallResult<bool>(new WebError("Socket not connected"));
 
             if (Authenticated)
@@ -489,11 +625,11 @@ namespace CryptoExchange.Net.Sockets
                 var authResult = await socketClient.AuthenticateSocketAsync(this).ConfigureAwait(false);
                 if (!authResult)
                 {
-                    log.Write(LogLevel.Warning, $"Socket {Socket.Id} authentication failed on reconnected socket. Disconnecting and reconnecting.");
+                    log.Write(LogLevel.Warning, $"Socket {SocketId} authentication failed on reconnected socket. Disconnecting and reconnecting.");
                     return authResult;
                 }
 
-                log.Write(LogLevel.Debug, $"Socket {Socket.Id} authentication succeeded on reconnected socket.");
+                log.Write(LogLevel.Debug, $"Socket {SocketId} authentication succeeded on reconnected socket.");
             }
 
             // Get a list of all subscriptions on the socket
@@ -504,7 +640,7 @@ namespace CryptoExchange.Net.Sockets
             // Foreach subscription which is subscribed by a subscription request we will need to resend that request to resubscribe
             for (var i = 0; i < subscriptionList.Count; i += socketClient.ClientOptions.MaxConcurrentResubscriptionsPerSocket)
             {
-                if (!Socket.IsOpen)
+                if (!_socket.IsOpen)
                     return new CallResult<bool>(new WebError("Socket not connected"));
 
                 var taskList = new List<Task<CallResult<bool>>>();
@@ -516,10 +652,10 @@ namespace CryptoExchange.Net.Sockets
                     return taskList.First(t => !t.Result.Success).Result;
             }
 
-            if (!Socket.IsOpen)
+            if (!_socket.IsOpen)
                 return new CallResult<bool>(new WebError("Socket not connected"));
 
-            log.Write(LogLevel.Debug, $"Socket {Socket.Id} all subscription successfully resubscribed on reconnected socket.");
+            log.Write(LogLevel.Debug, $"Socket {SocketId} all subscription successfully resubscribed on reconnected socket.");
             return new CallResult<bool>(true);
         }
 
@@ -530,100 +666,10 @@ namespace CryptoExchange.Net.Sockets
 
         internal async Task<CallResult<bool>> ResubscribeAsync(SocketSubscription socketSubscription)
         {
-            if (!Socket.IsOpen)
+            if (!_socket.IsOpen)
                 return new CallResult<bool>(new UnknownError("Socket is not connected"));
 
             return await socketClient.SubscribeAndWaitAsync(this, socketSubscription.Request!, socketSubscription).ConfigureAwait(false);
         }
-        
-        /// <summary>
-        /// Close the connection
-        /// </summary>
-        /// <returns></returns>
-        public async Task CloseAsync()
-        {
-            Connected = false;
-            ShouldReconnect = false;
-            if (socketClient.sockets.ContainsKey(Socket.Id))
-                socketClient.sockets.TryRemove(Socket.Id, out _);
-
-            lock (subscriptionLock) 
-            {
-                foreach (var subscription in subscriptions)
-                {
-                    if (subscription.CancellationTokenRegistration.HasValue)
-                        subscription.CancellationTokenRegistration.Value.Dispose();
-                }
-            }
-            await Socket.CloseAsync().ConfigureAwait(false);
-            Socket.Dispose();
-        }
-
-        /// <summary>
-        /// Close a subscription on this connection. If all subscriptions on this connection are closed the connection gets closed as well
-        /// </summary>
-        /// <param name="subscription">Subscription to close</param>
-        /// <returns></returns>
-        public async Task CloseAsync(SocketSubscription subscription)
-        {
-            if (!Socket.IsOpen)
-                return;
-
-            if (subscription.CancellationTokenRegistration.HasValue)
-                subscription.CancellationTokenRegistration.Value.Dispose();
-
-            if (subscription.Confirmed)
-                await socketClient.UnsubscribeAsync(this, subscription).ConfigureAwait(false);
-
-            bool shouldCloseConnection;
-            lock (subscriptionLock)
-                shouldCloseConnection = !subscriptions.Any(r => r.UserSubscription && subscription != r);
-
-            if (shouldCloseConnection)
-                await CloseAsync().ConfigureAwait(false);
-
-            lock (subscriptionLock)
-                subscriptions.Remove(subscription);            
-        }
-    }
-
-    internal class PendingRequest
-    {
-        public Func<JToken, bool> Handler { get; }
-        public JToken? Result { get; private set; }
-        public bool Completed { get; private set; }
-        public AsyncResetEvent Event { get; }
-        public TimeSpan Timeout { get; }
-
-        private CancellationTokenSource cts;
-
-        public PendingRequest(Func<JToken, bool> handler, TimeSpan timeout)
-        {
-            Handler = handler;
-            Event = new AsyncResetEvent(false, false);
-            Timeout = timeout;
-
-            cts = new CancellationTokenSource(timeout);
-            cts.Token.Register(Fail, false);
-        }
-
-        public bool CheckData(JToken data)
-        {
-            if (Handler(data))
-            {
-                Result = data;
-                Completed = true;
-                Event.Set();
-                return true;
-            }            
-
-            return false;
-        }
-
-        public void Fail()
-        {
-            Completed = true;
-            Event.Set();
-        }
     }
 }

CryptoExchange.Net/Sockets/UpdateSubscription.cs

@@ -72,7 +72,7 @@ namespace CryptoExchange.Net.Sockets
         /// <summary>
         /// The id of the socket
         /// </summary>
-        public int SocketId => connection.Socket.Id;
+        public int SocketId => connection.SocketId;
 
         /// <summary>
         /// The id of the subscription
@@ -103,9 +103,9 @@ namespace CryptoExchange.Net.Sockets
         /// Close the socket to cause a reconnect
         /// </summary>
         /// <returns></returns>
-        internal Task ReconnectAsync()
+        public Task ReconnectAsync()
         {
-            return connection.Socket.CloseAsync();
+            return connection.TriggerReconnectAsync();
         }
 
         /// <summary>

CryptoExchange.Net/Sockets/WebsocketFactory.cs

@@ -1,24 +1,25 @@
-using System.Collections.Generic;
+using System;
+using System.Collections.Generic;
 using CryptoExchange.Net.Interfaces;
 using CryptoExchange.Net.Logging;
 
 namespace CryptoExchange.Net.Sockets
 {
     /// <summary>
-    /// Default weboscket factory implementation
+    /// Default websocket factory implementation
     /// </summary>
     public class WebsocketFactory : IWebsocketFactory
     {
         /// <inheritdoc />
         public IWebsocket CreateWebsocket(Log log, string url)
         {
-            return new CryptoExchangeWebSocketClient(log, url);
+            return new CryptoExchangeWebSocketClient(log, new Uri(url));
         }
 
         /// <inheritdoc />
         public IWebsocket CreateWebsocket(Log log, string url, IDictionary<string, string> cookies, IDictionary<string, string> headers)
         {
-            return new CryptoExchangeWebSocketClient(log, url, cookies, headers);
+            return new CryptoExchangeWebSocketClient(log, new Uri(url), cookies, headers);
         }
     }
 }

README.md

@@ -18,6 +18,18 @@ I develop and maintain this package on my own for free in my spare time. Donatio
 Alternatively, sponsor me on Github using [Github Sponsors](https://github.com/sponsors/JKorf)  
 
 ## Release notes
+* Version 5.1.9 - 08 May 2022
+    * Added latency to the timesync calculation
+    * Small fix for exception in socket close handling
+
+* Version 5.1.8 - 01 May 2022
+    * Cleanup socket code, fixed an issue which could cause connections to never reconnect when connection was lost
+    * Added support for sending requests which expect an empty response
+    * Fixed issue with the DateTimeConverter date interpretation
+
+* Version 5.1.7 - 14 Apr 2022
+    * Moved some Rest parameters from BaseRestClient to RestApiClient to allow different implementations for sub clients
+
 * Version 5.1.6 - 10 Mar 2022
     * Updated EnumConverter to properly handle emtpy/null and default values