Updated version

commit 0571ed17a0
Author: JKorf <jankorf91@gmail.com>
Date:   Sun Jul 10 19:56:27 2022 +0200

    Fixed tests

commit 99c331b389
Author: JKorf <jankorf91@gmail.com>
Date:   Sun Jul 10 16:41:14 2022 +0200

    Updated version

commit 70f8bd203a
Author: JKorf <jankorf91@gmail.com>
Date:   Sun Jul 10 16:36:00 2022 +0200

    Finished up websocket refactoring

commit 89b517c936
Author: JKorf <jankorf91@gmail.com>
Date:   Fri Jul 8 20:24:58 2022 +0200

    wip

commit 91e33cc42c
Author: JKorf <jankorf91@gmail.com>
Date:   Thu Jul 7 22:17:55 2022 +0200

    wip

CryptoExchange.Net.UnitTests/CryptoExchange.Net.UnitTests.csproj

@@ -6,10 +6,10 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <packagereference Include="Microsoft.NET.Test.Sdk" Version="17.1.0-preview-20211130-02"></packagereference>
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.1.0-preview-20211130-02"></PackageReference>
     <PackageReference Include="Moq" Version="4.16.1" />
-    <packagereference Include="NUnit" Version="3.13.2"></packagereference>
-    <packagereference Include="NUnit3TestAdapter" Version="4.2.0"></packagereference>
+    <PackageReference Include="NUnit" Version="3.13.2"></PackageReference>
+    <PackageReference Include="NUnit3TestAdapter" Version="4.2.0"></PackageReference>
   </ItemGroup>
 
   <ItemGroup>

CryptoExchange.Net.UnitTests/SocketClientTests.cs

@@ -42,7 +42,7 @@ namespace CryptoExchange.Net.UnitTests
             socket.CanConnect = canConnect;
 
             //act
-            var connectResult = client.ConnectSocketSub(new SocketConnection(client, null, socket));
+            var connectResult = client.ConnectSocketSub(new SocketConnection(client, null, socket, null));
 
             //assert
             Assert.IsTrue(connectResult.Success == canConnect);
@@ -57,10 +57,10 @@ namespace CryptoExchange.Net.UnitTests
             socket.ShouldReconnect = true;
             socket.CanConnect = true;
             socket.DisconnectTime = DateTime.UtcNow;
-            var sub = new SocketConnection(client, null, socket);
+            var sub = new SocketConnection(client, null, socket, null);
             var rstEvent = new ManualResetEvent(false);
             JToken result = null;
-            sub.AddSubscription(SocketSubscription.CreateForIdentifier(10, "TestHandler", true, (messageEvent) =>
+            sub.AddSubscription(SocketSubscription.CreateForIdentifier(10, "TestHandler", true, false, (messageEvent) =>
             {
                 result = messageEvent.JsonData;
                 rstEvent.Set();
@@ -85,10 +85,10 @@ namespace CryptoExchange.Net.UnitTests
             socket.ShouldReconnect = true;
             socket.CanConnect = true;
             socket.DisconnectTime = DateTime.UtcNow;
-            var sub = new SocketConnection(client, null, socket);
+            var sub = new SocketConnection(client, null, socket, null);
             var rstEvent = new ManualResetEvent(false);
             string original = null;
-            sub.AddSubscription(SocketSubscription.CreateForIdentifier(10, "TestHandler", true, (messageEvent) =>
+            sub.AddSubscription(SocketSubscription.CreateForIdentifier(10, "TestHandler", true, false, (messageEvent) =>
             {
                 original = messageEvent.OriginalData;
                 rstEvent.Set();
@@ -103,34 +103,6 @@ namespace CryptoExchange.Net.UnitTests
             Assert.IsTrue(original == (enabled ? "{\"property\": 123}" : null));
         }
 
-        [TestCase]
-        public void DisconnectedSocket_Should_Reconnect()
-        {
-            // arrange
-            bool reconnected = false;
-            var client = new TestSocketClient(new TestOptions() { ReconnectInterval = TimeSpan.Zero, LogLevel = LogLevel.Debug });
-            var socket = client.CreateSocket();
-            socket.ShouldReconnect = true;
-            socket.CanConnect = true;
-            socket.DisconnectTime = DateTime.UtcNow;
-            var sub = new SocketConnection(client, null, socket);
-            sub.ShouldReconnect = true;
-            client.ConnectSocketSub(sub);
-            var rstEvent = new ManualResetEvent(false);
-            sub.ConnectionRestored += (a) =>
-            {
-                reconnected = true;
-                rstEvent.Set();
-            };
-
-            // act
-            socket.InvokeClose();
-            rstEvent.WaitOne(1000);
-
-            // assert
-            Assert.IsTrue(reconnected);
-        }
-
         [TestCase()]
         public void UnsubscribingStream_Should_CloseTheSocket()
         {
@@ -138,9 +110,11 @@ namespace CryptoExchange.Net.UnitTests
             var client = new TestSocketClient(new TestOptions() { ReconnectInterval = TimeSpan.Zero, LogLevel = LogLevel.Debug });
             var socket = client.CreateSocket();
             socket.CanConnect = true;
-            var sub = new SocketConnection(client, null, socket);
+            var sub = new SocketConnection(client, null, socket, null);
             client.ConnectSocketSub(sub);
-            var ups = new UpdateSubscription(sub, SocketSubscription.CreateForIdentifier(10, "Test", true, (e) => {}));
+            var us = SocketSubscription.CreateForIdentifier(10, "Test", true, false, (e) => { });
+            var ups = new UpdateSubscription(sub, us);
+            sub.AddSubscription(us);
 
             // act
             client.UnsubscribeAsync(ups).Wait();
@@ -158,8 +132,8 @@ namespace CryptoExchange.Net.UnitTests
             var socket2 = client.CreateSocket();
             socket1.CanConnect = true;
             socket2.CanConnect = true;
-            var sub1 = new SocketConnection(client, null, socket1);
-            var sub2 = new SocketConnection(client, null, socket2);
+            var sub1 = new SocketConnection(client, null, socket1, null);
+            var sub2 = new SocketConnection(client, null, socket2, null);
             client.ConnectSocketSub(sub1);
             client.ConnectSocketSub(sub2);
 
@@ -178,7 +152,7 @@ namespace CryptoExchange.Net.UnitTests
             var client = new TestSocketClient(new TestOptions() { ReconnectInterval = TimeSpan.Zero, LogLevel = LogLevel.Debug });
             var socket = client.CreateSocket();
             socket.CanConnect = false;
-            var sub = new SocketConnection(client, null, socket);
+            var sub = new SocketConnection(client, null, socket, null);
 
             // act
             var connectResult = client.ConnectSocketSub(sub);

CryptoExchange.Net.UnitTests/TestImplementations/TestRestClient.cs

@@ -135,7 +135,7 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
             throw new NotImplementedException();
         }
 
-        protected override TimeSyncInfo GetTimeSyncInfo()
+        public override TimeSyncInfo GetTimeSyncInfo()
         {
             throw new NotImplementedException();
         }
@@ -161,7 +161,7 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
             throw new NotImplementedException();
         }
 
-        protected override TimeSyncInfo GetTimeSyncInfo()
+        public override TimeSyncInfo GetTimeSyncInfo()
         {
             throw new NotImplementedException();
         }

CryptoExchange.Net.UnitTests/TestImplementations/TestSocket.cs

@@ -13,9 +13,15 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
         public bool Connected { get; set; }
 
         public event Action OnClose;
+
+#pragma warning disable 0067
+        public event Action OnReconnected;
+        public event Action OnReconnecting;
+#pragma warning restore 0067
         public event Action<string> OnMessage;
         public event Action<Exception> OnError;
         public event Action OnOpen;
+        public Func<Task<Uri>> GetReconnectionUrl { get; set; }
 
         public int Id { get; }
         public bool ShouldReconnect { get; set; }
@@ -40,6 +46,8 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
 
         public Uri Uri => new Uri("");
 
+        public TimeSpan KeepAliveInterval { get; set; }
+
         public static int lastId = 0;
         public static object lastIdLock = new object();
 
@@ -91,6 +99,7 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
         {
             Connected = false;
             DisconnectTime = DateTime.UtcNow;
+            Reconnecting = true;
             OnClose?.Invoke();
         }
 
@@ -113,11 +122,6 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
         {
             OnError?.Invoke(error);
         }
-
-        public async Task ProcessAsync()
-        {
-            while (Connected)
-                await Task.Delay(50);
-        }
+        public Task ReconnectAsync() => Task.CompletedTask;
     }
 }

CryptoExchange.Net.UnitTests/TestImplementations/TestSocketClient.cs

@@ -22,13 +22,13 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
         {
             SubClient = new TestSubSocketClient(exchangeOptions, exchangeOptions.SubOptions);
             SocketFactory = new Mock<IWebsocketFactory>().Object;
-            Mock.Get(SocketFactory).Setup(f => f.CreateWebsocket(It.IsAny<Log>(), It.IsAny<string>())).Returns(new TestSocket());
+            Mock.Get(SocketFactory).Setup(f => f.CreateWebsocket(It.IsAny<Log>(), It.IsAny<WebSocketParameters>())).Returns(new TestSocket());
         }
 
         public TestSocket CreateSocket()
         {
-            Mock.Get(SocketFactory).Setup(f => f.CreateWebsocket(It.IsAny<Log>(), It.IsAny<string>())).Returns(new TestSocket());
-            return (TestSocket)CreateSocket("123");
+            Mock.Get(SocketFactory).Setup(f => f.CreateWebsocket(It.IsAny<Log>(), It.IsAny<WebSocketParameters>())).Returns(new TestSocket());
+            return (TestSocket)CreateSocket("https://localhost:123/");
         }
 
         public CallResult<bool> ConnectSocketSub(SocketConnection sub)

CryptoExchange.Net/Clients/BaseClient.cs

@@ -63,6 +63,7 @@ namespace CryptoExchange.Net
             log = new Log(name);
             log.UpdateWriters(options.LogWriters);
             log.Level = options.LogLevel;
+            options.OnLoggingChanged += HandleLogConfigChange;
 
             ClientOptions = options;
 
@@ -282,12 +283,22 @@ namespace CryptoExchange.Net
             }
         }
 
+        /// <summary>
+        /// Handle a change in the client options log config
+        /// </summary>
+        private void HandleLogConfigChange()
+        {
+            log.UpdateWriters(ClientOptions.LogWriters);
+            log.Level = ClientOptions.LogLevel;
+        }
+
         /// <summary>
         /// Dispose
         /// </summary>
         public virtual void Dispose()
         {
             log.Write(LogLevel.Debug, "Disposing client");
+            ClientOptions.OnLoggingChanged -= HandleLogConfigChange;
             foreach (var client in ApiClients)
                 client.Dispose();
         }

CryptoExchange.Net/Clients/BaseRestClient.cs

@@ -172,11 +172,17 @@ namespace CryptoExchange.Net
 
             if (signed)
             {
-                var syncTimeResult = await apiClient.SyncTimeAsync().ConfigureAwait(false);
-                if (!syncTimeResult)
+                var syncTask = apiClient.SyncTimeAsync();
+                var timeSyncInfo = apiClient.GetTimeSyncInfo();
+                if (timeSyncInfo.TimeSyncState.LastSyncTime == default)
                 {
-                    log.Write(LogLevel.Debug, $"[{requestId}] Failed to sync time, aborting request: " + syncTimeResult.Error);
-                    return syncTimeResult.As<IRequest>(default);
+                    // Initially with first request we'll need to wait for the time syncing, if it's not the first request we can just continue
+                    var syncTimeResult = await syncTask.ConfigureAwait(false);
+                    if (!syncTimeResult)
+                    {
+                        log.Write(LogLevel.Debug, $"[{requestId}] Failed to sync time, aborting request: " + syncTimeResult.Error);
+                        return syncTimeResult.As<IRequest>(default);
+                    }
                 }
             }
 
@@ -314,7 +320,7 @@ namespace CryptoExchange.Net
                     responseStream.Close();
                     response.Close();
                     var parseResult = ValidateJson(data);
-                    var error = parseResult.Success ? ParseErrorResponse(parseResult.Data) : parseResult.Error!;
+                    var error = parseResult.Success ? ParseErrorResponse(parseResult.Data) : new ServerError(data)!;
                     if(error.Code == null || error.Code == 0)
                         error.Code = (int)response.StatusCode;
                     return new WebCallResult<T>(statusCode, headers, sw.Elapsed, data, request.Uri.ToString(), request.Content, request.Method, request.GetHeaders(), default, error);

CryptoExchange.Net/Clients/BaseSocketClient.cs

@@ -3,7 +3,7 @@ using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics.CodeAnalysis;
 using System.Linq;
-using System.Net.WebSockets;
+using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using CryptoExchange.Net.Authentication;
@@ -11,7 +11,6 @@ using CryptoExchange.Net.Interfaces;
 using CryptoExchange.Net.Objects;
 using CryptoExchange.Net.Sockets;
 using Microsoft.Extensions.Logging;
-using Newtonsoft.Json;
 using Newtonsoft.Json.Linq;
 
 namespace CryptoExchange.Net
@@ -36,9 +35,9 @@ namespace CryptoExchange.Net
         /// </summary>
         protected internal readonly SemaphoreSlim semaphoreSlim = new(1);
         /// <summary>
-        /// The max amount of concurrent socket connections
+        /// Keep alive interval for websocket connection
         /// </summary>
-        protected int MaxSocketConnections { get; set; } = 9999;        
+        protected TimeSpan KeepAliveInterval { get; set; } = TimeSpan.FromSeconds(10);
         /// <summary>
         /// Delegate used for processing byte data received from socket connections before it is processed by handlers
         /// </summary>
@@ -92,6 +91,20 @@ namespace CryptoExchange.Net
             }
         }
 
+        /// <inheritdoc />
+        public int CurrentConnections => socketConnections.Count;
+        /// <inheritdoc />
+        public int CurrentSubscriptions
+        {
+            get
+            {
+                if (!socketConnections.Any())
+                    return 0;
+
+                return socketConnections.Sum(s => s.Value.SubscriptionCount);
+            }
+        }
+
         /// <summary>
         /// Client options
         /// </summary>
@@ -106,10 +119,7 @@ namespace CryptoExchange.Net
         /// <param name="options">The options for this client</param>
         protected BaseSocketClient(string name, BaseSocketClientOptions options) : base(name, options)
         {
-            if (options == null)
-                throw new ArgumentNullException(nameof(options));
-
-            ClientOptions = options;
+            ClientOptions = options ?? throw new ArgumentNullException(nameof(options));
         }
 
         /// <inheritdoc />
@@ -164,7 +174,7 @@ namespace CryptoExchange.Net
                 return new CallResult<UpdateSubscription>(new InvalidOperationError("Client disposed, can't subscribe"));
 
             SocketConnection socketConnection;
-            SocketSubscription subscription;
+            SocketSubscription? subscription;
             var released = false;
             // Wait for a semaphore here, so we only connect 1 socket at a time.
             // This is necessary for being able to see if connections can be combined
@@ -179,23 +189,38 @@ namespace CryptoExchange.Net
 
             try
             {
-                // Get a new or existing socket connection
-                socketConnection = GetSocketConnection(apiClient, url, authenticated);
-
-                // Add a subscription on the socket connection
-                subscription = AddSubscription(request, identifier, true, socketConnection, dataHandler);
-                if (ClientOptions.SocketSubscriptionsCombineTarget == 1)
+                while (true)
                 {
-                    // Only 1 subscription per connection, so no need to wait for connection since a new subscription will create a new connection anyway
-                    semaphoreSlim.Release();
-                    released = true;
+                    // Get a new or existing socket connection
+                    var socketResult = await GetSocketConnection(apiClient, url, authenticated).ConfigureAwait(false);
+                    if(!socketResult)
+                        return socketResult.As<UpdateSubscription>(null);                    
+
+                    socketConnection = socketResult.Data;
+
+                    // Add a subscription on the socket connection
+                    subscription = AddSubscription(request, identifier, true, socketConnection, dataHandler, authenticated);
+                    if (subscription == null)
+                    {
+                        log.Write(LogLevel.Trace, $"Socket {socketConnection.SocketId} failed to add subscription, retrying on different connection");
+                        continue;
+                    }
+
+                    if (ClientOptions.SocketSubscriptionsCombineTarget == 1)
+                    {
+                        // Only 1 subscription per connection, so no need to wait for connection since a new subscription will create a new connection anyway
+                        semaphoreSlim.Release();
+                        released = true;
+                    }
+
+                    var needsConnecting = !socketConnection.Connected;
+
+                    var connectResult = await ConnectIfNeededAsync(socketConnection, authenticated).ConfigureAwait(false);
+                    if (!connectResult)
+                        return new CallResult<UpdateSubscription>(connectResult.Error!);
+
+                    break;
                 }
-
-                var needsConnecting = !socketConnection.Connected;
-
-                var connectResult = await ConnectIfNeededAsync(socketConnection, authenticated).ConfigureAwait(false);
-                if (!connectResult)
-                    return new CallResult<UpdateSubscription>(connectResult.Error!);
             }
             finally
             {
@@ -215,6 +240,7 @@ namespace CryptoExchange.Net
                 var subResult = await SubscribeAndWaitAsync(socketConnection, request, subscription).ConfigureAwait(false);
                 if (!subResult)
                 {
+                    log.Write(LogLevel.Warning, $"Socket {socketConnection.SocketId} failed to subscribe: {subResult.Error}");
                     await socketConnection.CloseAsync(subscription).ConfigureAwait(false);
                     return new CallResult<UpdateSubscription>(subResult.Error!);
                 }
@@ -225,7 +251,6 @@ namespace CryptoExchange.Net
                 subscription.Confirmed = true;
             }
 
-            socketConnection.ShouldReconnect = true;
             if (ct != default)
             {
                 subscription.CancellationTokenRegistration = ct.Register(async () =>
@@ -235,7 +260,7 @@ namespace CryptoExchange.Net
                 }, false);
             }
 
-            log.Write(LogLevel.Information, $"Socket {socketConnection.SocketId} subscription completed");
+            log.Write(LogLevel.Information, $"Socket {socketConnection.SocketId} subscription {subscription.Id} completed successfully");
             return new CallResult<UpdateSubscription>(new UpdateSubscription(socketConnection, subscription));
         }
 
@@ -295,7 +320,12 @@ namespace CryptoExchange.Net
             await semaphoreSlim.WaitAsync().ConfigureAwait(false);
             try
             {
-                socketConnection = GetSocketConnection(apiClient, url, authenticated);
+                var socketResult = await GetSocketConnection(apiClient, url, authenticated).ConfigureAwait(false);
+                if (!socketResult)
+                    return socketResult.As<T>(default);
+
+                socketConnection = socketResult.Data;
+
                 if (ClientOptions.SocketSubscriptionsCombineTarget == 1)
                 {
                     // Can release early when only a single sub per connection
@@ -309,8 +339,6 @@ namespace CryptoExchange.Net
             }
             finally
             {
-                //When the task is ready, release the semaphore. It is vital to ALWAYS release the semaphore when we are ready, or else we will end up with a Semaphore that is forever locked.
-                //This is why it is important to do the Release within a try...finally clause; program execution may crash or take a different path, this way you are guaranteed execution
                 if (!released)
                     semaphoreSlim.Release();
             }
@@ -364,11 +392,14 @@ namespace CryptoExchange.Net
             if (!authenticated || socket.Authenticated)
                 return new CallResult<bool>(true);
 
+            log.Write(LogLevel.Debug, $"Attempting to authenticate {socket.SocketId}");
             var result = await AuthenticateSocketAsync(socket).ConfigureAwait(false);
             if (!result)
             {
-                await socket.CloseAsync().ConfigureAwait(false);
                 log.Write(LogLevel.Warning, $"Socket {socket.SocketId} authentication failed");
+                if(socket.Connected)
+                    await socket.CloseAsync().ConfigureAwait(false);
+
                 result.Error!.Message = "Authentication failed: " + result.Error.Message;
                 return new CallResult<bool>(result.Error);
             }
@@ -443,9 +474,6 @@ namespace CryptoExchange.Net
         /// <param name="message"></param>
         /// <returns></returns>
         protected internal virtual JToken ProcessTokenData(JToken message)
-
-
-
         {
             return message;
         }
@@ -459,8 +487,9 @@ namespace CryptoExchange.Net
         /// <param name="userSubscription">Whether or not this is a user subscription (counts towards the max amount of handlers on a socket)</param>
         /// <param name="connection">The socket connection the handler is on</param>
         /// <param name="dataHandler">The handler of the data received</param>
+        /// <param name="authenticated">Whether the subscription needs authentication</param>
         /// <returns></returns>
-        protected virtual SocketSubscription AddSubscription<T>(object? request, string? identifier, bool userSubscription, SocketConnection connection, Action<DataEvent<T>> dataHandler)
+        protected virtual SocketSubscription? AddSubscription<T>(object? request, string? identifier, bool userSubscription, SocketConnection connection, Action<DataEvent<T>> dataHandler, bool authenticated)
         {
             void InternalHandler(MessageEvent messageEvent)
             {
@@ -482,9 +511,10 @@ namespace CryptoExchange.Net
             }
 
             var subscription = request == null
-                ? SocketSubscription.CreateForIdentifier(NextId(), identifier!, userSubscription, InternalHandler)
-                : SocketSubscription.CreateForRequest(NextId(), request, userSubscription, InternalHandler);
-            connection.AddSubscription(subscription);
+                ? SocketSubscription.CreateForIdentifier(NextId(), identifier!, userSubscription, authenticated, InternalHandler)
+                : SocketSubscription.CreateForRequest(NextId(), request, userSubscription, authenticated, InternalHandler);
+            if (!connection.AddSubscription(subscription))
+                return null;
             return subscription;
         }
 
@@ -496,11 +526,34 @@ namespace CryptoExchange.Net
         protected void AddGenericHandler(string identifier, Action<MessageEvent> action)
         {
             genericHandlers.Add(identifier, action);
-            var subscription = SocketSubscription.CreateForIdentifier(NextId(), identifier, false, action);
+            var subscription = SocketSubscription.CreateForIdentifier(NextId(), identifier, false, false, action);
             foreach (var connection in socketConnections.Values)
                 connection.AddSubscription(subscription);
         }
 
+        /// <summary>
+        /// Get the url to connect to (defaults to BaseAddress form the client options)
+        /// </summary>
+        /// <param name="apiClient"></param>
+        /// <param name="address"></param>
+        /// <param name="authentication"></param>
+        /// <returns></returns>
+        protected virtual Task<CallResult<string?>> GetConnectionUrlAsync(SocketApiClient apiClient, string address, bool authentication)
+        {
+            return Task.FromResult(new CallResult<string?>(address));
+        }
+
+        /// <summary>
+        /// Get the url to reconnect to after losing a connection
+        /// </summary>
+        /// <param name="apiClient"></param>
+        /// <param name="connection"></param>
+        /// <returns></returns>
+        public virtual Task<Uri?> GetReconnectUriAsync(SocketApiClient apiClient, SocketConnection connection)
+        {
+            return Task.FromResult<Uri?>(connection.ConnectionUri);
+        }
+
         /// <summary>
         /// Gets a connection for a new subscription or query. Can be an existing if there are open position or a new one.
         /// </summary>
@@ -508,32 +561,43 @@ namespace CryptoExchange.Net
         /// <param name="address">The address the socket is for</param>
         /// <param name="authenticated">Whether the socket should be authenticated</param>
         /// <returns></returns>
-        protected virtual SocketConnection GetSocketConnection(SocketApiClient apiClient, string address, bool authenticated)
+        protected virtual async Task<CallResult<SocketConnection>> GetSocketConnection(SocketApiClient apiClient, string address, bool authenticated)
         {
-            var socketResult = socketConnections.Where(s => s.Value.Uri.ToString().TrimEnd('/') == address.TrimEnd('/')
+            var socketResult = socketConnections.Where(s => (s.Value.Status == SocketConnection.SocketStatus.None || s.Value.Status == SocketConnection.SocketStatus.Connected)
+                                                  && s.Value.Tag.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 || (socketConnections.Count >= MaxSocketConnections && socketConnections.All(s => s.Value.SubscriptionCount >= ClientOptions.SocketSubscriptionsCombineTarget)))
+                if (result.SubscriptionCount < ClientOptions.SocketSubscriptionsCombineTarget || (socketConnections.Count >= ClientOptions.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;
+                    return new CallResult<SocketConnection>(result);
                 }
             }
 
+            var connectionAddress = await GetConnectionUrlAsync(apiClient, address, authenticated).ConfigureAwait(false);
+            if (!connectionAddress)
+            {
+                log.Write(LogLevel.Warning, $"Failed to determine connection url: " + connectionAddress.Error);
+                return connectionAddress.As<SocketConnection>(null);
+            }
+
+            if (connectionAddress.Data != address)
+                log.Write(LogLevel.Debug, $"Connection address set to " + connectionAddress.Data);
+
             // Create new socket
-            var socket = CreateSocket(address);
-            var socketConnection = new SocketConnection(this, apiClient, socket);
+            var socket = CreateSocket(connectionAddress.Data!);
+            var socketConnection = new SocketConnection(this, apiClient, socket, address);
             socketConnection.UnhandledMessage += HandleUnhandledMessage;
             foreach (var kvp in genericHandlers)
             {
-                var handler = SocketSubscription.CreateForIdentifier(NextId(), kvp.Key, false, kvp.Value);
+                var handler = SocketSubscription.CreateForIdentifier(NextId(), kvp.Key, false, false, kvp.Value);
                 socketConnection.AddSubscription(handler);
             }
 
-            return socketConnection;
+            return new CallResult<SocketConnection>(socketConnection);
         }
 
         /// <summary>
@@ -561,6 +625,23 @@ namespace CryptoExchange.Net
             return new CallResult<bool>(new CantConnectError());
         }
 
+        /// <summary>
+        /// Get parameters for the websocket connection
+        /// </summary>
+        /// <param name="address">The address to connect to</param>
+        /// <returns></returns>
+        protected virtual WebSocketParameters GetWebSocketParameters(string address) 
+            => new (new Uri(address), ClientOptions.AutoReconnect)
+               {
+                   DataInterpreterBytes = dataInterpreterBytes,
+                   DataInterpreterString = dataInterpreterString,
+                   KeepAliveInterval = KeepAliveInterval,
+                   ReconnectInterval = ClientOptions.ReconnectInterval,
+                   RatelimitPerSecond = RateLimitPerSocketPerSecond,
+                   Proxy = ClientOptions.Proxy,
+                   Timeout = ClientOptions.SocketNoDataTimeout
+               };        
+
         /// <summary>
         /// Create a socket for an address
         /// </summary>
@@ -568,23 +649,8 @@ namespace CryptoExchange.Net
         /// <returns></returns>
         protected virtual IWebsocket CreateSocket(string address)
         {
-            var socket = SocketFactory.CreateWebsocket(log, address);
+            var socket = SocketFactory.CreateWebsocket(log, GetWebSocketParameters(address));
             log.Write(LogLevel.Debug, $"Socket {socket.Id} new socket created for " + address);
-
-            if (ClientOptions.Proxy != null)
-                socket.SetProxy(ClientOptions.Proxy);
-
-            socket.Timeout = ClientOptions.SocketNoDataTimeout;
-            socket.DataInterpreterBytes = dataInterpreterBytes;
-            socket.DataInterpreterString = dataInterpreterString;
-            socket.RatelimitPerSecond = RateLimitPerSocketPerSecond;
-            socket.OnError += e =>
-            {
-                if(e is WebSocketException wse)
-                    log.Write(LogLevel.Warning, $"Socket {socket.Id} error: Websocket error code {wse.WebSocketErrorCode}, details: " + e.ToLogString());
-                else
-                    log.Write(LogLevel.Warning, $"Socket {socket.Id} error: " + e.ToLogString());
-            };
             return socket;
         }
 
@@ -642,7 +708,6 @@ namespace CryptoExchange.Net
         /// <returns></returns>
         public virtual async Task UnsubscribeAsync(int subscriptionId)
         {
-
             SocketSubscription? subscription = null;
             SocketConnection? connection = null;
             foreach(var socket in socketConnections.Values.ToList())
@@ -658,7 +723,7 @@ namespace CryptoExchange.Net
             if (subscription == null || connection == null)
                 return;
 
-            log.Write(LogLevel.Information, "Closing subscription " + subscriptionId);
+            log.Write(LogLevel.Information, $"Socket {connection.SocketId} Unsubscribing subscription " + subscriptionId);
             await connection.CloseAsync(subscription).ConfigureAwait(false);
         }
 
@@ -672,7 +737,7 @@ namespace CryptoExchange.Net
             if (subscription == null)
                 throw new ArgumentNullException(nameof(subscription));
 
-            log.Write(LogLevel.Information, "Closing subscription " + subscription.Id);
+            log.Write(LogLevel.Information, $"Socket {subscription.SocketId} Unsubscribing subscription  " + subscription.Id);
             await subscription.CloseAsync().ConfigureAwait(false);
         }
 
@@ -682,7 +747,7 @@ namespace CryptoExchange.Net
         /// <returns></returns>
         public virtual async Task UnsubscribeAllAsync()
         {
-            log.Write(LogLevel.Information, $"Closing all {socketConnections.Sum(s => s.Value.SubscriptionCount)} subscriptions");
+            log.Write(LogLevel.Information, $"Unsubscribing all {socketConnections.Sum(s => s.Value.SubscriptionCount)} subscriptions");
             var tasks = new List<Task>();
             {
                 var socketList = socketConnections.Values;
@@ -693,6 +758,39 @@ namespace CryptoExchange.Net
             await Task.WhenAll(tasks.ToArray()).ConfigureAwait(false);
         }
 
+        /// <summary>
+        /// Reconnect all connections
+        /// </summary>
+        /// <returns></returns>
+        public virtual async Task ReconnectAsync()
+        {
+            log.Write(LogLevel.Information, $"Reconnecting all {socketConnections.Count} connections");
+            var tasks = new List<Task>();
+            {
+                var socketList = socketConnections.Values;
+                foreach (var sub in socketList)
+                    tasks.Add(sub.TriggerReconnectAsync());
+            }
+
+            await Task.WhenAll(tasks.ToArray()).ConfigureAwait(false);
+        }
+
+        /// <summary>
+        /// Log the current state of connections and subscriptions
+        /// </summary>
+        public string GetSubscriptionsState()
+        {
+            var sb = new StringBuilder();
+            sb.AppendLine($"{socketConnections.Count} connections, {CurrentSubscriptions} subscriptions, kbps: {IncomingKbps}");
+            foreach(var connection in socketConnections)
+            {
+                sb.AppendLine($"  Connection {connection.Key}: {connection.Value.SubscriptionCount} subscriptions, status: {connection.Value.Status}, authenticated: {connection.Value.Authenticated}, kbps: {connection.Value.IncomingKbps}");
+                foreach (var subscription in connection.Value.Subscriptions)
+                    sb.AppendLine($"    Subscription {subscription.Id}, authenticated: {subscription.Authenticated}, confirmed: {subscription.Confirmed}");
+            }
+            return sb.ToString();
+        }
+
         /// <summary>
         /// Dispose the client
         /// </summary>
@@ -701,8 +799,11 @@ namespace CryptoExchange.Net
             disposing = true;
             periodicEvent?.Set();
             periodicEvent?.Dispose();
-            log.Write(LogLevel.Debug, "Disposing socket client, closing all subscriptions");
-            _ = UnsubscribeAllAsync();
+            if (socketConnections.Sum(s => s.Value.SubscriptionCount) > 0)
+            {
+                log.Write(LogLevel.Debug, "Disposing socket client, closing all subscriptions");
+                _ = UnsubscribeAllAsync();
+            }
             semaphoreSlim?.Dispose();
             base.Dispose();
         }

CryptoExchange.Net/Clients/RestApiClient.cs

@@ -18,7 +18,7 @@ namespace CryptoExchange.Net
         /// Get time sync info for an API client
         /// </summary>
         /// <returns></returns>
-        protected abstract TimeSyncInfo GetTimeSyncInfo();
+        public abstract TimeSyncInfo GetTimeSyncInfo();
         
         /// <summary>
         /// Get time offset for an API client
@@ -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

@@ -31,7 +31,7 @@ namespace CryptoExchange.Net.Converters
             if(reader.TokenType is JsonToken.Integer)
             {
                 var longValue = (long)reader.Value;
-                if (longValue == 0)
+                if (longValue == 0 || longValue == -1)
                     return objectType == typeof(DateTime) ? default(DateTime): null;
                 if (longValue < 19999999999)
                     return ConvertFromSeconds(longValue);
@@ -45,6 +45,9 @@ namespace CryptoExchange.Net.Converters
             else if (reader.TokenType is JsonToken.Float)
             {
                 var doubleValue = (double)reader.Value;
+                if (doubleValue == 0 || doubleValue == -1)
+                    return objectType == typeof(DateTime) ? default(DateTime) : null;
+
                 if (doubleValue < 19999999999)
                     return ConvertFromSeconds(doubleValue);
                 
@@ -56,6 +59,9 @@ namespace CryptoExchange.Net.Converters
                 if (string.IsNullOrWhiteSpace(stringValue))
                     return null;
 
+                if (string.IsNullOrWhiteSpace(stringValue) || stringValue == "0" || stringValue == "-1")
+                    return objectType == typeof(DateTime) ? default(DateTime) : null;
+
                 if (stringValue.Length == 8)
                 {
                     // Parse 20211103 format

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.8</PackageVersion>
-    <AssemblyVersion>5.1.8</AssemblyVersion>
-    <FileVersion>5.1.8</FileVersion>
+    <PackageVersion>5.2.2</PackageVersion>
+    <AssemblyVersion>5.2.2</AssemblyVersion>
+    <FileVersion>5.2.2</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.8 - 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</PackageReleaseNotes>
+    <PackageReleaseNotes>5.2.2 - Added support for retrieving a new url when socket connection is lost and reconnection will happen</PackageReleaseNotes>
     <Nullable>enable</Nullable>
     <LangVersion>9.0</LangVersion>
     <PackageLicenseExpression>MIT</PackageLicenseExpression>

CryptoExchange.Net/ExtensionMethods.cs

@@ -426,6 +426,7 @@ namespace CryptoExchange.Net
             var uriBuilder = new UriBuilder();
             uriBuilder.Scheme = baseUri.Scheme;
             uriBuilder.Host = baseUri.Host;
+            uriBuilder.Port = baseUri.Port;
             uriBuilder.Path = baseUri.AbsolutePath;
             var httpValueCollection = HttpUtility.ParseQueryString(string.Empty);
             foreach (var parameter in parameters)
@@ -454,6 +455,7 @@ namespace CryptoExchange.Net
             var uriBuilder = new UriBuilder();
             uriBuilder.Scheme = baseUri.Scheme;
             uriBuilder.Host = baseUri.Host;
+            uriBuilder.Port = baseUri.Port;
             uriBuilder.Path = baseUri.AbsolutePath;
             var httpValueCollection = HttpUtility.ParseQueryString(string.Empty);
             foreach (var parameter in parameters)

CryptoExchange.Net/Interfaces/ISocketClient.cs

@@ -27,6 +27,16 @@ namespace CryptoExchange.Net.Interfaces
         /// </summary>
         public double IncomingKbps { get; }
 
+        /// <summary>
+        /// The current amount of connections to the API from this client. A connection can have multiple subscriptions.
+        /// </summary>
+        public int CurrentConnections { get; }
+        
+        /// <summary>
+        /// The current amount of subscriptions running from the client
+        /// </summary>
+        public int CurrentSubscriptions { get; }
+
         /// <summary>
         /// Unsubscribe from a stream using the subscription id received when starting the subscription
         /// </summary>

CryptoExchange.Net/Interfaces/IWebsocket.cs

@@ -1,4 +1,5 @@
 using CryptoExchange.Net.Objects;
+using CryptoExchange.Net.Sockets;
 using System;
 using System.Security.Authentication;
 using System.Text;
@@ -27,36 +28,28 @@ namespace CryptoExchange.Net.Interfaces
         /// Websocket opened event
         /// </summary>
         event Action OnOpen;
+        /// <summary>
+        /// Websocket has lost connection to the server and is attempting to reconnect
+        /// </summary>
+        event Action OnReconnecting;
+        /// <summary>
+        /// Websocket has reconnected to the server
+        /// </summary>
+        event Action OnReconnected;
+        /// <summary>
+        /// Get reconntion url
+        /// </summary>
+        Func<Task<Uri?>> GetReconnectionUrl { get; set; }
 
         /// <summary>
         /// Unique id for this socket
         /// </summary>
         int Id { get; }
         /// <summary>
-        /// Origin header
-        /// </summary>
-        string? Origin { get; set; }
-        /// <summary>
-        /// Encoding to use for sending/receiving string data
-        /// </summary>
-        Encoding? Encoding { get; set; }
-        /// <summary>
-        /// The max amount of outgoing messages per second
-        /// </summary>
-        int? RatelimitPerSecond { get; set; }
-        /// <summary>
         /// The current kilobytes per second of data being received, averaged over the last 3 seconds
         /// </summary>
         double IncomingKbps { get; }
         /// <summary>
-        /// Handler for byte data
-        /// </summary>
-        Func<byte[], string>? DataInterpreterBytes { get; set; }
-        /// <summary>
-        /// Handler for string data
-        /// </summary>
-        Func<string, string>? DataInterpreterString { get; set; }
-        /// <summary>
         /// The uri the socket connects to
         /// </summary>
         Uri Uri { get; }
@@ -69,37 +62,20 @@ namespace CryptoExchange.Net.Interfaces
         /// </summary>
         bool IsOpen { get; }
         /// <summary>
-        /// Supported ssl protocols
-        /// </summary>
-        SslProtocols SSLProtocols { get; set; }
-        /// <summary>
-        /// The max time for no data being received before the connection is considered lost
-        /// </summary>
-        TimeSpan Timeout { get; set; }
-        /// <summary>
-        /// Set a proxy to use when connecting
-        /// </summary>
-        /// <param name="proxy"></param>
-        void SetProxy(ApiProxy proxy);
-        /// <summary>
         /// Connect the socket
         /// </summary>
         /// <returns></returns>
-        Task<bool> ConnectAsync();        
-        /// <summary>
-        /// Receive and send messages over the connection. Resulting task should complete when closing the socket.
-        /// </summary>
-        /// <returns></returns>
-        Task ProcessAsync();
+        Task<bool> ConnectAsync();
         /// <summary>
         /// Send data
         /// </summary>
         /// <param name="data"></param>
         void Send(string data);
         /// <summary>
-        /// Reset socket when a connection is lost to prepare for a new connection
+        /// Reconnect the socket
         /// </summary>
-        void Reset();
+        /// <returns></returns>
+        Task ReconnectAsync();
         /// <summary>
         /// Close the connection
         /// </summary>

CryptoExchange.Net/Interfaces/IWebsocketFactory.cs

@@ -1,5 +1,5 @@
-using System.Collections.Generic;
-using CryptoExchange.Net.Logging;
+using CryptoExchange.Net.Logging;
+using CryptoExchange.Net.Sockets;
 
 namespace CryptoExchange.Net.Interfaces
 {
@@ -12,17 +12,8 @@ namespace CryptoExchange.Net.Interfaces
         /// Create a websocket for an url
         /// </summary>
         /// <param name="log">The logger</param>
-        /// <param name="url">The url the socket is fo</param>
+        /// <param name="parameters">The parameters to use for the connection</param>
         /// <returns></returns>
-        IWebsocket CreateWebsocket(Log log, string url);
-        /// <summary>
-        /// Create a websocket for an url
-        /// </summary>
-        /// <param name="log">The logger</param>
-        /// <param name="url">The url the socket is fo</param>
-        /// <param name="cookies">Cookies to be send in the initial request</param>
-        /// <param name="headers">Headers to be send in the initial request</param>
-        /// <returns></returns>
-        IWebsocket CreateWebsocket(Log log, string url, IDictionary<string, string> cookies, IDictionary<string, string> headers);
+        IWebsocket CreateWebsocket(Log log, WebSocketParameters parameters);
     }
 }

CryptoExchange.Net/Logging/Log.cs

@@ -26,6 +26,8 @@ namespace CryptoExchange.Net.Logging
         /// </summary>
         public string ClientName { get; set; }
 
+        private readonly object _lock = new object();
+
         /// <summary>
         /// ctor
         /// </summary>
@@ -42,7 +44,8 @@ namespace CryptoExchange.Net.Logging
         /// <param name="textWriters"></param>
         public void UpdateWriters(List<ILogger> textWriters)
         {
-            writers = textWriters;
+            lock (_lock)
+                writers = textWriters;
         }
 
         /// <summary>
@@ -56,16 +59,19 @@ namespace CryptoExchange.Net.Logging
                 return;
 
             var logMessage = $"{ClientName,-10} | {message}";
-            foreach (var writer in writers.ToList())
+            lock (_lock)
             {
-                try
+                foreach (var writer in writers)
                 {
-                    writer.Log(logLevel, logMessage);
-                }
-                catch (Exception e)
-                {
-                    // Can't write to the logging so where else to output..
-                    Trace.WriteLine($"{DateTime.Now:yyyy/MM/dd HH:mm:ss:fff} | Warning | Failed to write log to writer {writer.GetType()}: " + e.ToLogString());
+                    try
+                    {
+                        writer.Log(logLevel, logMessage);
+                    }
+                    catch (Exception e)
+                    {
+                        // Can't write to the logging so where else to output..
+                        Trace.WriteLine($"{DateTime.Now:yyyy/MM/dd HH:mm:ss:fff} | Warning | Failed to write log to writer {writer.GetType()}: " + e.ToLogString());
+                    }
                 }
             }
         }

CryptoExchange.Net/Objects/Options.cs

@@ -14,15 +14,35 @@ namespace CryptoExchange.Net.Objects
     /// </summary>
     public class BaseOptions
     {
+        internal event Action? OnLoggingChanged;
+
+        private LogLevel _logLevel = LogLevel.Information;
         /// <summary>
         /// The minimum log level to output
         /// </summary>
-        public LogLevel LogLevel { get; set; } = LogLevel.Information;
+        public LogLevel LogLevel
+        {
+            get => _logLevel;
+            set
+            {
+                _logLevel = value;
+                OnLoggingChanged?.Invoke();
+            }
+        }
 
+        private List<ILogger> _logWriters = new List<ILogger> { new DebugLogger() };
         /// <summary>
         /// The log writers
         /// </summary>
-        public List<ILogger> LogWriters { get; set; } = new List<ILogger> { new DebugLogger() };
+        public List<ILogger> LogWriters
+        {
+            get => _logWriters;
+            set
+            {
+                _logWriters = value;
+                OnLoggingChanged?.Invoke();
+            }
+        }
 
         /// <summary>
         /// If true, the CallResult and DataEvent objects will also include the originally received json data in the OriginalData property
@@ -156,16 +176,6 @@ namespace CryptoExchange.Net.Objects
         /// </summary>
         public TimeSpan ReconnectInterval { get; set; } = TimeSpan.FromSeconds(5);
 
-        /// <summary>
-        /// The maximum number of times to try to reconnect, default null will retry indefinitely 
-        /// </summary>
-        public int? MaxReconnectTries { get; set; }
-
-        /// <summary>
-        /// The maximum number of times to try to resubscribe after reconnecting
-        /// </summary>
-        public int? MaxResubscribeTries { get; set; }
-
         /// <summary>
         /// Max number of concurrent resubscription tasks per socket after reconnecting a socket
         /// </summary>
@@ -189,6 +199,11 @@ namespace CryptoExchange.Net.Objects
         /// </summary>
         public int? SocketSubscriptionsCombineTarget { get; set; }
 
+        /// <summary>
+        /// The max amount of connections to make to the server. Can be used for API's which only allow a certain number of connections. Changing this to a high value might cause issues.
+        /// </summary>
+        public int? MaxSocketConnections { get; set; }
+
         /// <summary>
         /// ctor
         /// </summary>
@@ -207,18 +222,17 @@ namespace CryptoExchange.Net.Objects
 
             AutoReconnect = baseOptions.AutoReconnect;
             ReconnectInterval = baseOptions.ReconnectInterval;
-            MaxReconnectTries = baseOptions.MaxReconnectTries;
-            MaxResubscribeTries = baseOptions.MaxResubscribeTries;
             MaxConcurrentResubscriptionsPerSocket = baseOptions.MaxConcurrentResubscriptionsPerSocket;
             SocketResponseTimeout = baseOptions.SocketResponseTimeout;
             SocketNoDataTimeout = baseOptions.SocketNoDataTimeout;
             SocketSubscriptionsCombineTarget = baseOptions.SocketSubscriptionsCombineTarget;
+            MaxSocketConnections = baseOptions.MaxSocketConnections;
         }
 
         /// <inheritdoc />
         public override string ToString()
         {
-            return $"{base.ToString()}, AutoReconnect: {AutoReconnect}, ReconnectInterval: {ReconnectInterval}, MaxReconnectTries: {MaxReconnectTries}, MaxResubscribeTries: {MaxResubscribeTries}, MaxConcurrentResubscriptionsPerSocket: {MaxConcurrentResubscriptionsPerSocket}, SocketResponseTimeout: {SocketResponseTimeout:c}, SocketNoDataTimeout: {SocketNoDataTimeout}, SocketSubscriptionsCombineTarget: {SocketSubscriptionsCombineTarget}";
+            return $"{base.ToString()}, AutoReconnect: {AutoReconnect}, ReconnectInterval: {ReconnectInterval}, MaxConcurrentResubscriptionsPerSocket: {MaxConcurrentResubscriptionsPerSocket}, SocketResponseTimeout: {SocketResponseTimeout:c}, SocketNoDataTimeout: {SocketNoDataTimeout}, SocketSubscriptionsCombineTarget: {SocketSubscriptionsCombineTarget}, MaxSocketConnections: {MaxSocketConnections}";
         }
     }
 

CryptoExchange.Net/OrderBook/SymbolOrderBook.cs

@@ -258,24 +258,32 @@ namespace CryptoExchange.Net.OrderBook
             }
 
             _subscription = startResult.Data;
-            _subscription.ConnectionLost += () =>
-            {
-                log.Write(LogLevel.Warning, $"{Id} order book {Symbol} connection lost");
-                Status = OrderBookStatus.Reconnecting;
-                Reset();
-            };
-            _subscription.ConnectionClosed += () =>
-            {
-                log.Write(LogLevel.Warning, $"{Id} order book {Symbol} disconnected");
-                Status = OrderBookStatus.Disconnected;
-                _ = StopAsync();
-            };
+            _subscription.ConnectionLost += HandleConnectionLost;
+            _subscription.ConnectionClosed += HandleConnectionClosed;
+            _subscription.ConnectionRestored += HandleConnectionRestored;
 
-            _subscription.ConnectionRestored += async time => await ResyncAsync().ConfigureAwait(false);
             Status = OrderBookStatus.Synced;
             return new CallResult<bool>(true);
         }
 
+        private void HandleConnectionLost() {
+             log.Write(LogLevel.Warning, $"{Id} order book {Symbol} connection lost");
+             if (Status != OrderBookStatus.Disposed) {
+                Status = OrderBookStatus.Reconnecting;
+                Reset();
+            }
+        }
+
+        private void HandleConnectionClosed() {
+            log.Write(LogLevel.Warning, $"{Id} order book {Symbol} disconnected");
+            Status = OrderBookStatus.Disconnected;
+            _ = StopAsync();
+        }
+
+        private async void HandleConnectionRestored(TimeSpan _) {
+            await ResyncAsync().ConfigureAwait(false);
+        }
+
         /// <inheritdoc/>
         public async Task StopAsync()
         {
@@ -286,8 +294,12 @@ namespace CryptoExchange.Net.OrderBook
             if (_processTask != null)
                 await _processTask.ConfigureAwait(false);
 
-            if (_subscription != null)
+            if (_subscription != null) {
                 await _subscription.CloseAsync().ConfigureAwait(false);
+                _subscription.ConnectionLost -= HandleConnectionLost;
+                _subscription.ConnectionClosed -= HandleConnectionClosed;
+                _subscription.ConnectionRestored -= HandleConnectionRestored;
+            }
             log.Write(LogLevel.Trace, $"{Id} order book {Symbol} stopped");
         }
 
@@ -601,13 +613,13 @@ namespace CryptoExchange.Net.OrderBook
 
         private async Task ProcessQueue()
         {
-            while (Status != OrderBookStatus.Disconnected)
+            while (Status != OrderBookStatus.Disconnected && Status != OrderBookStatus.Disposed)
             {
                 await _queueEvent.WaitAsync().ConfigureAwait(false);
 
                 while (_processQueue.TryDequeue(out var item))
                 {
-                    if (Status == OrderBookStatus.Disconnected)
+                    if (Status == OrderBookStatus.Disconnected || Status == OrderBookStatus.Disposed)
                         break;
 
                     if (_stopProcessing)

CryptoExchange.Net/Sockets/CryptoExchangeWebSocketClient.cs

@@ -5,13 +5,10 @@ using Microsoft.Extensions.Logging;
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
-using System.Diagnostics;
 using System.IO;
 using System.Linq;
 using System.Net;
 using System.Net.WebSockets;
-using System.Security.Authentication;
-using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -22,18 +19,31 @@ namespace CryptoExchange.Net.Sockets
     /// </summary>
     public class CryptoExchangeWebSocketClient : IWebsocket
     {
+        enum ProcessState
+        {
+            Idle,
+            Processing,
+            WaitingForClose,
+            Reconnecting
+        }
+
         internal static int lastStreamId;
         private static readonly object streamIdLock = new();
 
-        private ClientWebSocket _socket;
         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 readonly SemaphoreSlim _closeSem;
         private readonly List<DateTime> _outgoingMessages;
+
+        private ClientWebSocket _socket;
+        private CancellationTokenSource _ctsSource;
         private DateTime _lastReceivedMessagesUpdate;
+        private Task? _processTask;
+        private Task? _closeTask;
+        private bool _stopRequested;
+        private bool _disposed;
+        private ProcessState _processState;
+
 
         /// <summary>
         /// Received messages, the size and the timstamp
@@ -48,48 +58,21 @@ namespace CryptoExchange.Net.Sockets
         /// <summary>
         /// Log
         /// </summary>
-        protected Log log;
-
-        /// <summary>
-        /// Handlers for when an error happens on the socket
-        /// </summary>
-        protected readonly List<Action<Exception>> errorHandlers = new();
-        /// <summary>
-        /// Handlers for when the socket connection is opened
-        /// </summary>
-        protected readonly List<Action> openHandlers = new();
-        /// <summary>
-        /// Handlers for when the connection is closed
-        /// </summary>
-        protected readonly List<Action> closeHandlers = new();
-        /// <summary>
-        /// Handlers for when a message is received
-        /// </summary>
-        protected readonly List<Action<string>> messageHandlers = new();
+        protected Log _log;
 
         /// <inheritdoc />
         public int Id { get; }
 
         /// <inheritdoc />
-        public string? Origin { get; set; }
+        public WebSocketParameters Parameters { get; }
 
         /// <summary>
         /// The timestamp this socket has been active for the last time
         /// </summary>
         public DateTime LastActionTime { get; private set; }
-
-        /// <summary>
-        /// Delegate used for processing byte data received from socket connections before it is processed by handlers
-        /// </summary>
-        public Func<byte[], string>? DataInterpreterBytes { get; set; }
-
-        /// <summary>
-        /// Delegate used for processing string data received from socket connections before it is processed by handlers
-        /// </summary>
-        public Func<string, string>? DataInterpreterString { get; set; }
-
+        
         /// <inheritdoc />
-        public Uri Uri { get; }
+        public Uri Uri => Parameters.Uri;
 
         /// <inheritdoc />
         public bool IsClosed => _socket.State == WebSocketState.Closed;
@@ -97,31 +80,6 @@ namespace CryptoExchange.Net.Sockets
         /// <inheritdoc />
         public bool IsOpen => _socket.State == WebSocketState.Open && !_ctsSource.IsCancellationRequested;
 
-        /// <summary>
-        /// Ssl protocols supported. NOT USED BY THIS IMPLEMENTATION
-        /// </summary>
-        public SslProtocols SSLProtocols { get; set; }
-
-        private Encoding _encoding = Encoding.UTF8;
-        /// <inheritdoc />
-        public Encoding? Encoding
-        {
-            get => _encoding;
-            set
-            {
-                if(value != null)
-                    _encoding = value;
-            }
-        }
-
-        /// <summary>
-        /// The max amount of outgoing messages per second
-        /// </summary>
-        public int? RatelimitPerSecond { get; set; }
-
-        /// <inheritdoc />
-        public TimeSpan Timeout { get; set; }
-
         /// <inheritdoc />
         public double IncomingKbps
         {
@@ -140,57 +98,31 @@ namespace CryptoExchange.Net.Sockets
         }
 
         /// <inheritdoc />
-        public event Action OnClose
-        {
-            add => closeHandlers.Add(value);
-            remove => closeHandlers.Remove(value);
-        }
-
+        public event Action? OnClose;
         /// <inheritdoc />
-        public event Action<string> OnMessage
-        {
-            add => messageHandlers.Add(value);
-            remove => messageHandlers.Remove(value);
-        }
-
+        public event Action<string>? OnMessage;
         /// <inheritdoc />
-        public event Action<Exception> OnError
-        {
-            add => errorHandlers.Add(value);
-            remove => errorHandlers.Remove(value);
-        }
-
+        public event Action<Exception>? OnError;
         /// <inheritdoc />
-        public event Action OnOpen
-        {
-            add => openHandlers.Add(value);
-            remove => openHandlers.Remove(value);
-        }
+        public event Action? OnOpen;
+        /// <inheritdoc />
+        public event Action? OnReconnecting;
+        /// <inheritdoc />
+        public event Action? OnReconnected;
+        /// <inheritdoc />
+        public Func<Task<Uri?>>? GetReconnectionUrl { get; set; }
 
         /// <summary>
         /// ctor
         /// </summary>
         /// <param name="log">The log object to use</param>
-        /// <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>())
-        {
-        }
-
-        /// <summary>
-        /// ctor
-        /// </summary>
-        /// <param name="log">The log object to use</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, Uri uri, IDictionary<string, string> cookies, IDictionary<string, string> headers)
+        /// <param name="websocketParameters">The parameters for this socket</param>
+        public CryptoExchangeWebSocketClient(Log log, WebSocketParameters websocketParameters)
         {
             Id = NextStreamId();
-            this.log = log;
-            Uri = uri;
-            this.cookies = cookies;
-            this.headers = headers;
+            _log = log;
 
+            Parameters = websocketParameters;
             _outgoingMessages = new List<DateTime>();
             _receivedMessages = new List<ReceiveItem>();
             _sendEvent = new AsyncResetEvent();
@@ -198,92 +130,224 @@ namespace CryptoExchange.Net.Sockets
             _ctsSource = new CancellationTokenSource();
             _receivedMessagesLock = new object();
 
+            _closeSem = new SemaphoreSlim(1, 1);
             _socket = CreateSocket();
         }
 
-        /// <inheritdoc />
-        public virtual void SetProxy(ApiProxy proxy)
-        {
-            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);
-        }
-
         /// <inheritdoc />
         public virtual async Task<bool> ConnectAsync()
         {
-            log.Write(LogLevel.Debug, $"Socket {Id} connecting");
+            if (!await ConnectInternalAsync().ConfigureAwait(false))
+                return false;
+            
+            OnOpen?.Invoke();
+            _processTask = ProcessAsync();
+            return true;            
+        }
+
+        /// <summary>
+        /// Create the socket object
+        /// </summary>
+        private ClientWebSocket CreateSocket()
+        {
+            var cookieContainer = new CookieContainer();
+            foreach (var cookie in Parameters.Cookies)
+                cookieContainer.Add(new Cookie(cookie.Key, cookie.Value));
+
+            var socket = new ClientWebSocket();
+            socket.Options.Cookies = cookieContainer;
+            foreach (var header in Parameters.Headers)
+                socket.Options.SetRequestHeader(header.Key, header.Value);
+            socket.Options.KeepAliveInterval = Parameters.KeepAliveInterval ?? TimeSpan.Zero;
+            socket.Options.SetBuffer(65536, 65536); // Setting it to anything bigger than 65536 throws an exception in .net framework
+            if (Parameters.Proxy != null)
+                SetProxy(Parameters.Proxy);
+            return socket;
+        }
+
+        private async Task<bool> ConnectInternalAsync()
+        {
+            _log.Write(LogLevel.Debug, $"Socket {Id} connecting");
             try
             {
-                using CancellationTokenSource tcs = new(TimeSpan.FromSeconds(10));                
+                using CancellationTokenSource tcs = new(TimeSpan.FromSeconds(10));
                 await _socket.ConnectAsync(Uri, tcs.Token).ConfigureAwait(false);
-                
-                Handle(openHandlers);
             }
             catch (Exception e)
             {
-                log.Write(LogLevel.Debug, $"Socket {Id} connection failed: " + e.ToLogString());
+                _log.Write(LogLevel.Debug, $"Socket {Id} connection failed: " + e.ToLogString());
                 return false;
             }
-    
-            log.Write(LogLevel.Debug, $"Socket {Id} connected to {Uri}");
+
+            _log.Write(LogLevel.Debug, $"Socket {Id} connected to {Uri}");
             return true;
         }
 
         /// <inheritdoc />
-        public virtual async Task ProcessAsync()
+        private 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");
+            while (!_stopRequested)
+            {
+                _log.Write(LogLevel.Debug, $"Socket {Id} starting processing tasks");
+                _processState = ProcessState.Processing;
+                var sendTask = SendLoopAsync();
+                var receiveTask = ReceiveLoopAsync();
+                var timeoutTask = Parameters.Timeout != null && Parameters.Timeout > TimeSpan.FromSeconds(0) ? CheckTimeoutAsync() : Task.CompletedTask;
+                await Task.WhenAll(sendTask, receiveTask, timeoutTask).ConfigureAwait(false);
+                _log.Write(LogLevel.Debug, $"Socket {Id} processing tasks finished");
+
+                _processState = ProcessState.WaitingForClose;
+                while (_closeTask == null)
+                    await Task.Delay(50).ConfigureAwait(false);
+
+                await _closeTask.ConfigureAwait(false);
+                _closeTask = null;
+
+                if (!Parameters.AutoReconnect)
+                {
+                    _processState = ProcessState.Idle;
+                    OnClose?.Invoke();
+                    return;
+                }    
+
+                if (!_stopRequested)
+                {
+                    _processState = ProcessState.Reconnecting;
+                    OnReconnecting?.Invoke();                    
+                }
+
+                while (!_stopRequested)
+                {
+                    _log.Write(LogLevel.Debug, $"Socket {Id} attempting to reconnect");
+                    var task = GetReconnectionUrl?.Invoke();
+                    if (task != null)
+                    {
+                        var reconnectUri = await task.ConfigureAwait(false);
+                        if (reconnectUri != null && Parameters.Uri != reconnectUri)
+                        {
+                            _log.Write(LogLevel.Debug, $"Socket {Id} reconnect URI set to {reconnectUri}");
+                            Parameters.Uri = reconnectUri;
+                        }
+                    }
+
+                    _socket = CreateSocket();
+                    _ctsSource.Dispose();
+                    _ctsSource = new CancellationTokenSource();
+                    while (_sendBuffer.TryDequeue(out _)) { } // Clear send buffer
+
+                    var connected = await ConnectInternalAsync().ConfigureAwait(false);
+                    if (!connected)
+                    {
+                        await Task.Delay(Parameters.ReconnectInterval).ConfigureAwait(false);
+                        continue;
+                    }
+
+                    OnReconnected?.Invoke();
+                    break;
+                }
+            }
+
+            _processState = ProcessState.Idle;
         }
 
         /// <inheritdoc />
         public virtual void Send(string data)
         {
             if (_ctsSource.IsCancellationRequested)
-                throw new InvalidOperationException($"Socket {Id} Can't send data when socket is not connected");
+                return;
 
-            var bytes = _encoding.GetBytes(data);
-            log.Write(LogLevel.Trace, $"Socket {Id} Adding {bytes.Length} to sent buffer");
+            var bytes = Parameters.Encoding.GetBytes(data);
+            _log.Write(LogLevel.Trace, $"Socket {Id} Adding {bytes.Length} to sent buffer");
             _sendBuffer.Enqueue(bytes);
             _sendEvent.Set();
         }
 
+        /// <inheritdoc />
+        public virtual async Task ReconnectAsync()
+        {
+            if (_processState != ProcessState.Processing && IsOpen)
+                return;
+
+            _log.Write(LogLevel.Debug, $"Socket {Id} reconnect requested");
+            _closeTask = CloseInternalAsync();
+            await _closeTask.ConfigureAwait(false);
+        }
+
         /// <inheritdoc />
         public virtual async Task CloseAsync()
         {
-            log.Write(LogLevel.Debug, $"Socket {Id} closing");
-            await CloseInternalAsync().ConfigureAwait(false);
+            await _closeSem.WaitAsync().ConfigureAwait(false);
+            try
+            {
+                if (_closeTask != null && !_closeTask.IsCompleted)
+                {
+                    _log.Write(LogLevel.Debug, $"Socket {Id} CloseAsync() waiting for existing close task");
+                    await _closeTask.ConfigureAwait(false);
+                    return;
+                }
+
+                _stopRequested = true;
+
+                if (!IsOpen)
+                {
+                    _log.Write(LogLevel.Debug, $"Socket {Id} CloseAsync() socket not open");
+                    return;
+                }
+
+                _log.Write(LogLevel.Debug, $"Socket {Id} closing");
+                _closeTask = CloseInternalAsync();
+            }
+            finally
+            {
+                _closeSem.Release();
+            }
+
+            await _closeTask.ConfigureAwait(false);
+            if(_processTask != null)
+                await _processTask.ConfigureAwait(false);
+            OnClose?.Invoke();
+            _log.Write(LogLevel.Debug, $"Socket {Id} closed");
         }
-        
+
         /// <summary>
         /// Internal close method
         /// </summary>
         /// <returns></returns>
         private async Task CloseInternalAsync()
         {
+            if (_disposed)
+                return;
+
+            //_closeState = CloseState.Closing;
             _ctsSource.Cancel();
             _sendEvent.Set();
 
             if (_socket.State == WebSocketState.Open)
-                await _socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "Closing", default).ConfigureAwait(false);
-
-            log.Write(LogLevel.Debug, $"Socket {Id} closed");
-            Handle(closeHandlers);
+            {
+                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
+                    // Websocket is set to Aborted state when the cancelation token is set during SendAsync/ReceiveAsync
+                    // So socket might go to aborted state, might still be open
+                }
+            }
+            else if(_socket.State == WebSocketState.CloseReceived)
+            {
+                try
+                {
+                    await _socket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", default).ConfigureAwait(false);
+                }
+                catch (Exception)
+                {
+                    // Can sometimes throw an exception when socket is in aborted state due to timing
+                    // Websocket is set to Aborted state when the cancelation token is set during SendAsync/ReceiveAsync
+                    // So socket might go to aborted state, might still be open
+                }
+            }
         }
 
         /// <summary>
@@ -291,44 +355,14 @@ namespace CryptoExchange.Net.Sockets
         /// </summary>
         public void Dispose()
         {
-            log.Write(LogLevel.Debug, $"Socket {Id} disposing");
+            if (_disposed)
+                return;
+
+            _log.Write(LogLevel.Debug, $"Socket {Id} disposing");
+            _disposed = true;
             _socket.Dispose();
             _ctsSource.Dispose();
-
-            errorHandlers.Clear();
-            openHandlers.Clear();
-            closeHandlers.Clear();
-            messageHandlers.Clear();
-            log.Write(LogLevel.Trace, $"Socket {Id} disposed");
-        }
-
-        /// <inheritdoc />
-        public void Reset()
-        {
-            log.Write(LogLevel.Debug, $"Socket {Id} resetting");
-            _ctsSource = new CancellationTokenSource();
-
-            while (_sendBuffer.TryDequeue(out _)) { } // Clear send buffer
-
-            _socket = CreateSocket();
-        }
-        
-        /// <summary>
-        /// Create the socket object
-        /// </summary>
-        private ClientWebSocket CreateSocket()
-        {
-            var cookieContainer = new CookieContainer();
-            foreach (var cookie in cookies)
-                cookieContainer.Add(new Cookie(cookie.Key, cookie.Value));
-
-            var socket = new ClientWebSocket();
-            socket.Options.Cookies = cookieContainer;
-            foreach (var header in headers)
-                socket.Options.SetRequestHeader(header.Key, header.Value);
-            socket.Options.KeepAliveInterval = TimeSpan.FromSeconds(10);
-            socket.Options.SetBuffer(65536, 65536); // Setting it to anything bigger than 65536 throws an exception in .net framework
-            return socket;
+            _log.Write(LogLevel.Trace, $"Socket {Id} disposed");
         }
 
         /// <summary>
@@ -351,25 +385,25 @@ namespace CryptoExchange.Net.Sockets
 
                     while (_sendBuffer.TryDequeue(out var data))
                     {
-                        if (RatelimitPerSecond != null)
+                        if (Parameters.RatelimitPerSecond != null)
                         {
                             // Wait for rate limit
                             DateTime? start = null;
-                            while (MessagesSentLastSecond() >= RatelimitPerSecond)
+                            while (MessagesSentLastSecond() >= Parameters.RatelimitPerSecond)
                             {
                                 start ??= DateTime.UtcNow;
                                 await Task.Delay(50).ConfigureAwait(false);
                             }
 
                             if (start != null)
-                                log.Write(LogLevel.Trace, $"Socket {Id} sent delayed {Math.Round((DateTime.UtcNow - start.Value).TotalMilliseconds)}ms because of rate limit");
+                                _log.Write(LogLevel.Debug, $"Socket {Id} sent delayed {Math.Round((DateTime.UtcNow - start.Value).TotalMilliseconds)}ms because of rate limit");
                         }
 
                         try
                         {
                             await _socket.SendAsync(new ArraySegment<byte>(data, 0, data.Length), WebSocketMessageType.Text, true, _ctsSource.Token).ConfigureAwait(false);
                             _outgoingMessages.Add(DateTime.UtcNow);
-                            log.Write(LogLevel.Trace, $"Socket {Id} sent {data.Length} bytes");
+                            _log.Write(LogLevel.Trace, $"Socket {Id} sent {data.Length} bytes");
                         }
                         catch (OperationCanceledException)
                         {
@@ -378,9 +412,9 @@ namespace CryptoExchange.Net.Sockets
                         }
                         catch (Exception ioe)
                         {
-                            // Connection closed unexpectedly, .NET framework                      
-                            Handle(errorHandlers, ioe);
-                            await CloseInternalAsync().ConfigureAwait(false);
+                            // Connection closed unexpectedly, .NET framework
+                            OnError?.Invoke(ioe);
+                            _closeTask = CloseInternalAsync();
                             break;
                         }
                     }
@@ -391,12 +425,12 @@ namespace CryptoExchange.Net.Sockets
                 // Because this is running in a separate task and not awaited until the socket gets closed
                 // any exception here will crash the send processing, but do so silently unless the socket get's stopped.
                 // Make sure we at least let the owner know there was an error
-                Handle(errorHandlers, e);
+                OnError?.Invoke(e);
                 throw;
             }
             finally
             {
-                log.Write(LogLevel.Trace, $"Socket {Id} Send loop finished");
+                _log.Write(LogLevel.Debug, $"Socket {Id} Send loop finished");
             }
         }
 
@@ -434,17 +468,17 @@ namespace CryptoExchange.Net.Sockets
                         }
                         catch (Exception wse)
                         {
-                            // Connection closed unexpectedly        
-                            Handle(errorHandlers, wse);
-                            await CloseInternalAsync().ConfigureAwait(false);
+                            // Connection closed unexpectedly
+                            OnError?.Invoke(wse);
+                            _closeTask = CloseInternalAsync();
                             break;
                         }
 
                         if (receiveResult.MessageType == WebSocketMessageType.Close)
                         {
                             // Connection closed unexpectedly        
-                            log.Write(LogLevel.Debug, $"Socket {Id} received `Close` message");
-                            await CloseInternalAsync().ConfigureAwait(false);
+                            _log.Write(LogLevel.Debug, $"Socket {Id} received `Close` message");
+                            _closeTask = CloseInternalAsync();
                             break;
                         }
 
@@ -453,7 +487,7 @@ namespace CryptoExchange.Net.Sockets
                             // We received data, but it is not complete, write it to a memory stream for reassembling
                             multiPartMessage = true;
                             memoryStream ??= new MemoryStream();
-                            log.Write(LogLevel.Trace, $"Socket {Id} received {receiveResult.Count} bytes in partial message");
+                            _log.Write(LogLevel.Trace, $"Socket {Id} received {receiveResult.Count} bytes in partial message");
                             await memoryStream.WriteAsync(buffer.Array, buffer.Offset, receiveResult.Count).ConfigureAwait(false);
                         }
                         else
@@ -461,13 +495,13 @@ namespace CryptoExchange.Net.Sockets
                             if (!multiPartMessage)
                             {
                                 // Received a complete message and it's not multi part
-                                log.Write(LogLevel.Trace, $"Socket {Id} received {receiveResult.Count} bytes in single message");
+                                _log.Write(LogLevel.Trace, $"Socket {Id} received {receiveResult.Count} bytes in single message");
                                 HandleMessage(buffer.Array!, buffer.Offset, receiveResult.Count, receiveResult.MessageType);
                             }
                             else
                             {
                                 // Received the end of a multipart message, write to memory stream for reassembling
-                                log.Write(LogLevel.Trace, $"Socket {Id} received {receiveResult.Count} bytes in partial message");
+                                _log.Write(LogLevel.Trace, $"Socket {Id} received {receiveResult.Count} bytes in partial message");
                                 await memoryStream!.WriteAsync(buffer.Array, buffer.Offset, receiveResult.Count).ConfigureAwait(false);
                             }
                             break;
@@ -495,12 +529,12 @@ namespace CryptoExchange.Net.Sockets
                         if (receiveResult?.EndOfMessage == true)
                         {
                             // Reassemble complete message from memory stream
-                            log.Write(LogLevel.Trace, $"Socket {Id} reassembled message of {memoryStream!.Length} bytes");
+                            _log.Write(LogLevel.Trace, $"Socket {Id} reassembled message of {memoryStream!.Length} bytes");
                             HandleMessage(memoryStream!.ToArray(), 0, (int)memoryStream.Length, receiveResult.MessageType);
                             memoryStream.Dispose();
                         }
                         else
-                            log.Write(LogLevel.Trace, $"Socket {Id} discarding incomplete message of {memoryStream!.Length} bytes");
+                            _log.Write(LogLevel.Trace, $"Socket {Id} discarding incomplete message of {memoryStream!.Length} bytes");
                     }
                 }
             }
@@ -509,12 +543,12 @@ namespace CryptoExchange.Net.Sockets
                 // Because this is running in a separate task and not awaited until the socket gets closed
                 // any exception here will crash the receive processing, but do so silently unless the socket gets stopped.
                 // Make sure we at least let the owner know there was an error
-                Handle(errorHandlers, e);
+                OnError?.Invoke(e);
                 throw;
             }
             finally
             {
-                log.Write(LogLevel.Trace, $"Socket {Id} Receive loop finished");
+                _log.Write(LogLevel.Debug, $"Socket {Id} Receive loop finished");
             }
         }
 
@@ -530,54 +564,92 @@ namespace CryptoExchange.Net.Sockets
             string strData;
             if (messageType == WebSocketMessageType.Binary)
             {
-                if (DataInterpreterBytes == null)
+                if (Parameters.DataInterpreterBytes == null)
                     throw new Exception("Byte interpreter not set while receiving byte data");
 
                 try
                 {
                     var relevantData = new byte[count];
                     Array.Copy(data, offset, relevantData, 0, count);
-                    strData = DataInterpreterBytes(relevantData);
+                    strData = Parameters.DataInterpreterBytes(relevantData);
                 }
                 catch(Exception e)
                 {
-                    log.Write(LogLevel.Error, $"Socket {Id} unhandled exception during byte data interpretation: " + e.ToLogString());
+                    _log.Write(LogLevel.Error, $"Socket {Id} unhandled exception during byte data interpretation: " + e.ToLogString());
                     return;
                 }
             }
             else
-                strData = _encoding.GetString(data, offset, count);
+                strData = Parameters.Encoding.GetString(data, offset, count);
 
-            if (DataInterpreterString != null)
+            if (Parameters.DataInterpreterString != null)
             {
                 try
                 {
-                    strData = DataInterpreterString(strData);
+                    strData = Parameters.DataInterpreterString(strData);
                 }
                 catch(Exception e)
                 {
-                    log.Write(LogLevel.Error, $"Socket {Id} unhandled exception during string data interpretation: " + e.ToLogString());
+                    _log.Write(LogLevel.Error, $"Socket {Id} unhandled exception during string data interpretation: " + e.ToLogString());
                     return;
                 }
             }
 
             try
             {
-                Handle(messageHandlers, strData);
+                LastActionTime = DateTime.UtcNow;
+                OnMessage?.Invoke(strData);
             }
             catch(Exception e)
             {
-                log.Write(LogLevel.Error, $"Socket {Id} unhandled exception during message processing: " + e.ToLogString());
+                _log.Write(LogLevel.Error, $"Socket {Id} unhandled exception during message processing: " + e.ToLogString());
             }
         }
 
+        /// <summary>
+        /// Trigger the OnMessage event
+        /// </summary>
+        /// <param name="data"></param>
+        protected void TriggerOnMessage(string data)
+        {
+            LastActionTime = DateTime.UtcNow;
+            OnMessage?.Invoke(data);
+        }
+
+        /// <summary>
+        /// Trigger the OnError event
+        /// </summary>
+        /// <param name="ex"></param>
+        protected void TriggerOnError(Exception ex) => OnError?.Invoke(ex);
+
+        /// <summary>
+        /// Trigger the OnError event
+        /// </summary>
+        protected void TriggerOnOpen() => OnOpen?.Invoke();
+
+        /// <summary>
+        /// Trigger the OnError event
+        /// </summary>
+        protected void TriggerOnClose() => OnClose?.Invoke();
+
+        /// <summary>
+        /// Trigger the OnReconnecting event
+        /// </summary>
+        protected void TriggerOnReconnecting() => OnReconnecting?.Invoke();
+
+        /// <summary>
+        /// Trigger the OnReconnected event
+        /// </summary>
+        protected void TriggerOnReconnected() => OnReconnected?.Invoke();
+
         /// <summary>
         /// Checks if there is no data received for a period longer than the specified timeout
         /// </summary>
         /// <returns></returns>
         protected async Task CheckTimeoutAsync()
         {
-            log.Write(LogLevel.Debug, $"Socket {Id} Starting task checking for no data received for {Timeout}");
+            _log.Write(LogLevel.Debug, $"Socket {Id} Starting task checking for no data received for {Parameters.Timeout}");
+            LastActionTime = DateTime.UtcNow;
             try 
             { 
                 while (true)
@@ -585,9 +657,9 @@ namespace CryptoExchange.Net.Sockets
                     if (_ctsSource.IsCancellationRequested)
                         return;
 
-                    if (DateTime.UtcNow - LastActionTime > Timeout)
+                    if (DateTime.UtcNow - LastActionTime > Parameters.Timeout)
                     {
-                        log.Write(LogLevel.Warning, $"Socket {Id} No data received for {Timeout}, reconnecting socket");
+                        _log.Write(LogLevel.Warning, $"Socket {Id} No data received for {Parameters.Timeout}, reconnecting socket");
                         _ = CloseAsync().ConfigureAwait(false);
                         return;
                     }
@@ -607,35 +679,11 @@ namespace CryptoExchange.Net.Sockets
                 // Because this is running in a separate task and not awaited until the socket gets closed
                 // any exception here will stop the timeout checking, but do so silently unless the socket get's stopped.
                 // Make sure we at least let the owner know there was an error
-                Handle(errorHandlers, e);
+                OnError?.Invoke(e);
                 throw;
             }
         }
 
-        /// <summary>
-        /// Helper to invoke handlers
-        /// </summary>
-        /// <param name="handlers"></param>
-        protected void Handle(List<Action> handlers)
-        {
-            LastActionTime = DateTime.UtcNow;
-            foreach (var handle in new List<Action>(handlers))
-                handle?.Invoke();
-        }
-
-        /// <summary>
-        /// Helper to invoke handlers
-        /// </summary>
-        /// <typeparam name="T"></typeparam>
-        /// <param name="handlers"></param>
-        /// <param name="data"></param>
-        protected void Handle<T>(List<Action<T>> handlers, T data)
-        {
-            LastActionTime = DateTime.UtcNow;
-            foreach (var handle in new List<Action<T>>(handlers))
-                handle?.Invoke(data);
-        }
-
         /// <summary>
         /// Get the next identifier
         /// </summary>
@@ -671,6 +719,27 @@ namespace CryptoExchange.Net.Sockets
                 _lastReceivedMessagesUpdate = checkTime;
             }
         }
+
+        /// <summary>
+        /// Set proxy on socket
+        /// </summary>
+        /// <param name="proxy"></param>
+        /// <exception cref="ArgumentException"></exception>
+        protected virtual void SetProxy(ApiProxy proxy)
+        {
+            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);
+        }
     }
 
     /// <summary>

CryptoExchange.Net/Sockets/SocketConnection.cs

@@ -3,13 +3,13 @@ using System;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
-using System.Threading;
 using System.Threading.Tasks;
 using CryptoExchange.Net.Logging;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Linq;
 using Microsoft.Extensions.Logging;
 using CryptoExchange.Net.Objects;
+using System.Net.WebSockets;
 
 namespace CryptoExchange.Net.Sockets
 {
@@ -57,10 +57,22 @@ namespace CryptoExchange.Net.Sockets
                 return subscriptions.Count(h => h.UserSubscription); }
         }
 
+        /// <summary>
+        /// Get a copy of the current subscriptions
+        /// </summary>
+        public SocketSubscription[] Subscriptions
+        {
+            get
+            {
+                lock (subscriptionLock)
+                    return subscriptions.Where(h => h.UserSubscription).ToArray();
+            }
+        }
+
         /// <summary>
         /// If the connection has been authenticated
         /// </summary>
-        public bool Authenticated { get; set; }
+        public bool Authenticated { get; internal set; }
 
         /// <summary>
         /// If connection is made
@@ -80,28 +92,13 @@ namespace CryptoExchange.Net.Sockets
         /// <summary>
         /// The connection uri
         /// </summary>
-        public Uri Uri => _socket.Uri;
+        public Uri ConnectionUri => _socket.Uri;
 
         /// <summary>
         /// The API client the connection is for
         /// </summary>
         public SocketApiClient ApiClient { get; set; }
 
-        /// <summary>
-        /// If the socket should be reconnected upon closing
-        /// </summary>
-        public bool ShouldReconnect { get; set; }
-
-        /// <summary>
-        /// Current reconnect try, reset when a successful connection is made
-        /// </summary>
-        public int ReconnectTry { get; set; }
-
-        /// <summary>
-        /// Current resubscribe try, reset when a successful connection is made
-        /// </summary>
-        public int ResubscribeTry { get; set; }
-
         /// <summary>
         /// Time of disconnecting
         /// </summary>
@@ -110,7 +107,7 @@ namespace CryptoExchange.Net.Sockets
         /// <summary>
         /// Tag for identificaion
         /// </summary>
-        public string? Tag { get; set; }
+        public string Tag { get; set; }
 
         /// <summary>
         /// If activity is paused
@@ -124,22 +121,39 @@ namespace CryptoExchange.Net.Sockets
                 {
                     pausedActivity = value;
                     log.Write(LogLevel.Information, $"Socket {SocketId} Paused activity: " + value);
-                    if(pausedActivity) ActivityPaused?.Invoke();
-                    else ActivityUnpaused?.Invoke();
+                    if(pausedActivity) _ = Task.Run(() => ActivityPaused?.Invoke());
+                    else _ = Task.Run(() => ActivityUnpaused?.Invoke());
                 }
             }
         }
 
+        /// <summary>
+        /// Status of the socket connection
+        /// </summary>
+        public SocketStatus Status
+        {
+            get => _status;
+            private set
+            {
+                if (_status == value)
+                    return;
+
+                var oldStatus = _status;
+                _status = value;
+                log.Write(LogLevel.Debug, $"Socket {SocketId} status changed from {oldStatus} to {_status}");
+            }
+        }
+
         private bool pausedActivity;
         private readonly List<SocketSubscription> subscriptions;
         private readonly object subscriptionLock = new();
 
-        private bool lostTriggered;
         private readonly Log log;
         private readonly BaseSocketClient socketClient;
 
         private readonly List<PendingRequest> pendingRequests;
-        private Task? _socketProcessReconnectTask;
+
+        private SocketStatus _status;
 
         /// <summary>
         /// The underlying websocket
@@ -152,120 +166,83 @@ namespace CryptoExchange.Net.Sockets
         /// <param name="client">The socket client</param>
         /// <param name="apiClient">The api client</param>
         /// <param name="socket">The socket</param>
-        public SocketConnection(BaseSocketClient client, SocketApiClient apiClient, IWebsocket socket)
+        /// <param name="tag"></param>
+        public SocketConnection(BaseSocketClient client, SocketApiClient apiClient, IWebsocket socket, string tag)
         {
             log = client.log;
             socketClient = client;
             ApiClient = apiClient;
+            Tag = tag;
 
             pendingRequests = new List<PendingRequest>();
-
             subscriptions = new List<SocketSubscription>();
+
             _socket = socket;
-
-            _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;
+            _socket.OnMessage += HandleMessage;
+            _socket.OnOpen += HandleOpen;
+            _socket.OnClose += HandleClose;
+            _socket.OnReconnecting += HandleReconnecting;
+            _socket.OnReconnected += HandleReconnected;
+            _socket.OnError += HandleError;
+            _socket.GetReconnectionUrl = GetReconnectionUrlAsync;
         }
 
         /// <summary>
-        /// Retrieve the underlying socket
+        /// Handler for a socket opening
         /// </summary>
-        /// <returns></returns>
-        public IWebsocket GetSocket()
+        protected virtual void HandleOpen()
         {
-            return _socket;
+            Status = SocketStatus.Connected;
+            PausedActivity = false;
         }
 
         /// <summary>
-        /// Trigger a reconnect of the socket connection
+        /// Handler for a socket closing without reconnect
         /// </summary>
-        /// <returns></returns>
-        public async Task TriggerReconnectAsync()
+        protected virtual void HandleClose()
         {
-            await _socket.CloseAsync().ConfigureAwait(false);
+            Status = SocketStatus.Closed;
+            Authenticated = false;
+            lock(subscriptionLock)
+            {
+                foreach (var sub in subscriptions)
+                    sub.Confirmed = false;
+            }    
+            Task.Run(() => ConnectionClosed?.Invoke());
         }
 
         /// <summary>
-        /// Close the connection
+        /// Handler for a socket losing conenction and starting reconnect
         /// </summary>
-        /// <returns></returns>
-        public async Task CloseAsync()
+        protected virtual void HandleReconnecting()
         {
-            ShouldReconnect = false;
-            if (socketClient.socketConnections.ContainsKey(SocketId))
-                socketClient.socketConnections.TryRemove(SocketId, out _);
-
+            Status = SocketStatus.Reconnecting;
+            DisconnectTime = DateTime.UtcNow;
+            Authenticated = false;
             lock (subscriptionLock)
             {
-                foreach (var subscription in subscriptions)
-                {
-                    if (subscription.CancellationTokenRegistration.HasValue)
-                        subscription.CancellationTokenRegistration.Value.Dispose();
-                }
+                foreach (var sub in subscriptions)
+                    sub.Confirmed = false;
             }
-            await _socket.CloseAsync().ConfigureAwait(false);
 
-            if (_socketProcessReconnectTask != null)
-                await _socketProcessReconnectTask.ConfigureAwait(false);
-
-            _socket.Dispose();
+            _ = Task.Run(() => ConnectionLost?.Invoke());
         }
 
         /// <summary>
-        /// Close a subscription on this connection. If all subscriptions on this connection are closed the connection gets closed as well
+        /// Get the url to connect to when reconnecting
         /// </summary>
-        /// <param name="subscription">Subscription to close</param>
         /// <returns></returns>
-        public async Task CloseAsync(SocketSubscription subscription)
+        protected virtual async Task<Uri?> GetReconnectionUrlAsync()
         {
-            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);
+            return await socketClient.GetReconnectUriAsync(ApiClient, this).ConfigureAwait(false);
         }
 
-        private void StartProcessingTask()
+        /// <summary>
+        /// Handler for a socket which has reconnected
+        /// </summary>
+        protected virtual async void HandleReconnected()
         {
-            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
+            Status = SocketStatus.Resubscribing;
             lock (pendingRequests)
             {
                 foreach (var pendingRequest in pendingRequests.ToList())
@@ -275,135 +252,37 @@ namespace CryptoExchange.Net.Sockets
                 }
             }
 
-            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;
-                    }
-                }
-            }
+            var reconnectSuccessful = await ProcessReconnectAsync().ConfigureAwait(false);
+            if (!reconnectSuccessful)
+                await _socket.ReconnectAsync().ConfigureAwait(false);
             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 _);
+                Status = SocketStatus.Connected;
+                _ = Task.Run(() =>
+                {
+                    ConnectionRestored?.Invoke(DateTime.UtcNow - DisconnectTime!.Value);
+                    DisconnectTime = null;
+                });
             }
         }
 
         /// <summary>
-        /// Dispose the connection
+        /// Handler for an error on a websocket
         /// </summary>
-        public void Dispose()
+        /// <param name="e">The exception</param>
+        protected virtual void HandleError(Exception e)
         {
-            _socket.Dispose();
+            if (e is WebSocketException wse)
+                log.Write(LogLevel.Warning, $"Socket {SocketId} error: Websocket error code {wse.WebSocketErrorCode}, details: " + e.ToLogString());
+            else
+                log.Write(LogLevel.Warning, $"Socket {SocketId} error: " + e.ToLogString());
         }
 
         /// <summary>
         /// Process a message received by the socket
         /// </summary>
         /// <param name="data">The received data</param>
-        private void ProcessMessage(string data)
+        protected virtual void HandleMessage(string data)
         {
             var timestamp = DateTime.UtcNow;
             log.Write(LogLevel.Trace, $"Socket {SocketId} received data: " + data);
@@ -419,15 +298,13 @@ namespace CryptoExchange.Net.Sockets
             }
 
             var handledResponse = false;
-            PendingRequest[] requests;
-            lock(pendingRequests)			
-                requests = pendingRequests.ToArray();
 
             // Remove any timed out requests
-            foreach (var request in requests.Where(r => r.Completed))
+            PendingRequest[] requests;
+            lock (pendingRequests)
             {
-                lock (pendingRequests)
-                    pendingRequests.Remove(request);
+                pendingRequests.RemoveAll(r => r.Completed);
+                requests = pendingRequests.ToArray();
             }
 
             // Check if this message is an answer on any pending requests
@@ -435,7 +312,7 @@ namespace CryptoExchange.Net.Sockets
             {
                 if (pendingRequest.CheckData(tokenData))
                 {
-                    lock (pendingRequests)                    
+                    lock (pendingRequests)
                         pendingRequests.Remove(pendingRequest);
 
                     if (!socketClient.ContinueOnQueryResponse)
@@ -447,8 +324,8 @@ 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);
-            var (handled, userProcessTime) = HandleData(messageEvent);
+            var messageEvent = new MessageEvent(this, tokenData, socketClient.ClientOptions.OutputOriginalData ? data : null, timestamp);
+            var (handled, userProcessTime, subscription) = HandleData(messageEvent);
             if (!handled && !handledResponse)
             {
                 if (!socketClient.UnhandledMessageExpected)
@@ -458,20 +335,129 @@ namespace CryptoExchange.Net.Sockets
 
             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. " +
+                log.Write(LogLevel.Debug, $"Socket {SocketId}{(subscription == null ? "" : " subscription " + subscription!.Id)} message processing slow ({(int)total.TotalMilliseconds}ms, {(int)userProcessTime.TotalMilliseconds}ms user code), 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)");
+
+            log.Write(LogLevel.Trace, $"Socket {SocketId}{(subscription == null ? "" : " subscription " + subscription!.Id)} message processed in {(int)total.TotalMilliseconds}ms, ({(int)userProcessTime.TotalMilliseconds}ms user code)");
+        }
+
+        /// <summary>
+        /// Connect the websocket
+        /// </summary>
+        /// <returns></returns>
+        public async Task<bool> ConnectAsync() => await _socket.ConnectAsync().ConfigureAwait(false);
+
+        /// <summary>
+        /// Retrieve the underlying socket
+        /// </summary>
+        /// <returns></returns>
+        public IWebsocket GetSocket() => _socket;
+
+        /// <summary>
+        /// Trigger a reconnect of the socket connection
+        /// </summary>
+        /// <returns></returns>
+        public async Task TriggerReconnectAsync() => await _socket.ReconnectAsync().ConfigureAwait(false);
+
+        /// <summary>
+        /// Close the connection
+        /// </summary>
+        /// <returns></returns>
+        public async Task CloseAsync()
+        {
+            if (Status == SocketStatus.Closed || Status == SocketStatus.Disposed)
+                return;
+
+            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);
+            _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)
+        {
+            lock (subscriptionLock)
+            {
+                if (!subscriptions.Contains(subscription))
+                    return;
+
+                subscription.Closed = true;
+            }
+
+            if (Status == SocketStatus.Closing || Status == SocketStatus.Closed || Status == SocketStatus.Disposed)
+                return;
+
+            log.Write(LogLevel.Debug, $"Socket {SocketId} closing subscription {subscription.Id}");
+            if (subscription.CancellationTokenRegistration.HasValue)
+                subscription.CancellationTokenRegistration.Value.Dispose();
+
+            if (subscription.Confirmed && _socket.IsOpen)
+                await socketClient.UnsubscribeAsync(this, subscription).ConfigureAwait(false);
+
+            bool shouldCloseConnection;
+            lock (subscriptionLock)
+            {
+                if (Status == SocketStatus.Closing)
+                {
+                    log.Write(LogLevel.Debug, $"Socket {SocketId} already closing");
+                    return;
+                }
+
+                shouldCloseConnection = subscriptions.All(r => !r.UserSubscription || r.Closed);
+                if (shouldCloseConnection)
+                    Status = SocketStatus.Closing;
+            }
+
+            if (shouldCloseConnection)
+            {
+                log.Write(LogLevel.Debug, $"Socket {SocketId} closing as there are no more subscriptions");
+                await CloseAsync().ConfigureAwait(false);
+            }
+
+            lock (subscriptionLock)
+                subscriptions.Remove(subscription);
+        }
+
+        /// <summary>
+        /// Dispose the connection
+        /// </summary>
+        public void Dispose()
+        {
+            Status = SocketStatus.Disposed;
+            _socket.Dispose();
         }
 
         /// <summary>
         /// Add a subscription to this connection
         /// </summary>
         /// <param name="subscription"></param>
-        public void AddSubscription(SocketSubscription subscription)
+        public bool AddSubscription(SocketSubscription subscription)
         {
-            lock(subscriptionLock)
+            lock (subscriptionLock)
+            {
+                if (Status != SocketStatus.None && Status != SocketStatus.Connected)
+                    return false;
+
                 subscriptions.Add(subscription);
+                if(subscription.UserSubscription)
+                    log.Write(LogLevel.Debug, $"Socket {SocketId} adding new subscription with id {subscription.Id}, total subscriptions on connection: {subscriptions.Count(s => s.UserSubscription)}");
+                return true;
+            }
         }
 
         /// <summary>
@@ -500,7 +486,7 @@ namespace CryptoExchange.Net.Sockets
         /// </summary>
         /// <param name="messageEvent"></param>
         /// <returns>True if the data was successfully handled</returns>
-        private (bool, TimeSpan) HandleData(MessageEvent messageEvent)
+        private (bool, TimeSpan, SocketSubscription?) HandleData(MessageEvent messageEvent)
         {
             SocketSubscription? currentSubscription = null;
             try
@@ -541,13 +527,13 @@ namespace CryptoExchange.Net.Sockets
                     }
                 }
                                
-                return (handled, userCodeDuration);
+                return (handled, userCodeDuration, currentSubscription);
             }
             catch (Exception ex)
             {
                 log.Write(LogLevel.Error, $"Socket {SocketId} Exception during message processing\r\nException: {ex.ToLogString()}\r\nData: {messageEvent.JsonData}");
                 currentSubscription?.InvokeExceptionHandler(ex);
-                return (false, TimeSpan.Zero);
+                return (false, TimeSpan.Zero, null);
             }
         }
 
@@ -605,21 +591,24 @@ namespace CryptoExchange.Net.Sockets
             }
         }
 
-        /// <summary>
-        /// Handler for a socket opening
-        /// </summary>
-        protected virtual void SocketOnOpen()
-        {
-            ReconnectTry = 0;
-            PausedActivity = false;
-        }
-
         private async Task<CallResult<bool>> ProcessReconnectAsync()
         {
             if (!_socket.IsOpen)
                 return new CallResult<bool>(new WebError("Socket not connected"));
 
-            if (Authenticated)
+            bool anySubscriptions = false;
+            lock (subscriptionLock)
+                anySubscriptions = subscriptions.Any(s => s.UserSubscription);
+
+            if (!anySubscriptions)
+            {
+                // No need to resubscribe anything
+                log.Write(LogLevel.Debug, $"Socket {SocketId} Nothing to resubscribe, closing connection");
+                _ = _socket.CloseAsync();
+                return new CallResult<bool>(true);
+            }
+
+            if (subscriptions.Any(s => s.Authenticated))
             {
                 // If we reconnected a authenticated connection we need to re-authenticate
                 var authResult = await socketClient.AuthenticateSocketAsync(this).ConfigureAwait(false);
@@ -629,13 +618,22 @@ namespace CryptoExchange.Net.Sockets
                     return authResult;
                 }
 
+                Authenticated = true;
                 log.Write(LogLevel.Debug, $"Socket {SocketId} authentication succeeded on reconnected socket.");
             }
 
             // Get a list of all subscriptions on the socket
-            List<SocketSubscription> subscriptionList;
+            List<SocketSubscription> subscriptionList = new List<SocketSubscription>();
             lock (subscriptionLock)
-                subscriptionList = subscriptions.Where(h => h.Request != null).ToList();
+            {
+                foreach (var subscription in subscriptions)
+                {
+                    if (subscription.Request != null)
+                        subscriptionList.Add(subscription);
+                    else
+                        subscription.Confirmed = true;
+                }
+            }
 
             // 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)
@@ -645,13 +643,16 @@ namespace CryptoExchange.Net.Sockets
 
                 var taskList = new List<Task<CallResult<bool>>>();
                 foreach (var subscription in subscriptionList.Skip(i).Take(socketClient.ClientOptions.MaxConcurrentResubscriptionsPerSocket))
-                    taskList.Add(socketClient.SubscribeAndWaitAsync(this, subscription.Request!, subscription));                
+                    taskList.Add(socketClient.SubscribeAndWaitAsync(this, subscription.Request!, subscription));
 
                 await Task.WhenAll(taskList).ConfigureAwait(false);
                 if (taskList.Any(t => !t.Result.Success))
                     return taskList.First(t => !t.Result.Success).Result;
             }
 
+            foreach (var subscription in subscriptionList)
+                subscription.Confirmed = true;
+
             if (!_socket.IsOpen)
                 return new CallResult<bool>(new WebError("Socket not connected"));
 
@@ -671,5 +672,40 @@ namespace CryptoExchange.Net.Sockets
 
             return await socketClient.SubscribeAndWaitAsync(this, socketSubscription.Request!, socketSubscription).ConfigureAwait(false);
         }
+
+        /// <summary>
+        /// Status of the socket connection
+        /// </summary>
+        public enum SocketStatus
+        {
+            /// <summary>
+            /// None/Initial
+            /// </summary>
+            None,
+            /// <summary>
+            /// Connected
+            /// </summary>
+            Connected,
+            /// <summary>
+            /// Reconnecting
+            /// </summary>
+            Reconnecting,
+            /// <summary>
+            /// Resubscribing on reconnected socket
+            /// </summary>
+            Resubscribing,
+            /// <summary>
+            /// Closing
+            /// </summary>
+            Closing,
+            /// <summary>
+            /// Closed
+            /// </summary>
+            Closed,
+            /// <summary>
+            /// Disposed
+            /// </summary>
+            Disposed
+        }
     }
 }

CryptoExchange.Net/Sockets/SocketSubscription.cs

@@ -43,19 +43,30 @@ namespace CryptoExchange.Net.Sockets
         /// </summary>
         public bool Confirmed { get; set; }
 
+        /// <summary>
+        /// Whether authentication is needed for this subscription
+        /// </summary>
+        public bool Authenticated { get; set; }
+
+        /// <summary>
+        /// Whether we're closing this subscription and a socket connection shouldn't be kept open for it
+        /// </summary>
+        public bool Closed { get; set; }
+
         /// <summary>
         /// Cancellation token registration, should be disposed when subscription is closed. Used for closing the subscription with 
         /// a provided cancelation token
         /// </summary>
         public CancellationTokenRegistration? CancellationTokenRegistration { get; set; }
 
-        private SocketSubscription(int id, object? request, string? identifier, bool userSubscription, Action<MessageEvent> dataHandler)
+        private SocketSubscription(int id, object? request, string? identifier, bool userSubscription, bool authenticated, Action<MessageEvent> dataHandler)
         {
             Id = id;
             UserSubscription = userSubscription;
             MessageHandler = dataHandler;
             Request = request;
             Identifier = identifier;
+            Authenticated = authenticated;
         }
 
         /// <summary>
@@ -64,12 +75,13 @@ namespace CryptoExchange.Net.Sockets
         /// <param name="id"></param>
         /// <param name="request"></param>
         /// <param name="userSubscription"></param>
+        /// <param name="authenticated"></param>
         /// <param name="dataHandler"></param>
         /// <returns></returns>
         public static SocketSubscription CreateForRequest(int id, object request, bool userSubscription,
-            Action<MessageEvent> dataHandler)
+            bool authenticated, Action<MessageEvent> dataHandler)
         {
-            return new SocketSubscription(id, request, null, userSubscription, dataHandler);
+            return new SocketSubscription(id, request, null, userSubscription, authenticated, dataHandler);
         }
 
         /// <summary>
@@ -78,12 +90,13 @@ namespace CryptoExchange.Net.Sockets
         /// <param name="id"></param>
         /// <param name="identifier"></param>
         /// <param name="userSubscription"></param>
+        /// <param name="authenticated"></param>
         /// <param name="dataHandler"></param>
         /// <returns></returns>
         public static SocketSubscription CreateForIdentifier(int id, string identifier, bool userSubscription,
-            Action<MessageEvent> dataHandler)
+            bool authenticated, Action<MessageEvent> dataHandler)
         {
-            return new SocketSubscription(id, null, identifier, userSubscription, dataHandler);
+            return new SocketSubscription(id, null, identifier, userSubscription, authenticated, dataHandler);
         }
 
         /// <summary>

CryptoExchange.Net/Sockets/UpdateSubscription.cs

@@ -22,8 +22,7 @@ namespace CryptoExchange.Net.Sockets
         }
 
         /// <summary>
-        /// Event when the connection is closed. This event happens when reconnecting/resubscribing has failed too often based on the <see cref="BaseSocketClientOptions.MaxReconnectTries"/> and <see cref="BaseSocketClientOptions.MaxResubscribeTries"/> options,
-        /// or <see cref="BaseSocketClientOptions.AutoReconnect"/> is false. The socket will not be reconnected
+        /// Event when the connection is closed and will not be reconnected
         /// </summary>
         public event Action ConnectionClosed
         {

CryptoExchange.Net/Sockets/WebSocketParameters.cs

@@ -0,0 +1,80 @@
+using CryptoExchange.Net.Objects;
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace CryptoExchange.Net.Sockets
+{
+    /// <summary>
+    /// Parameters for a websocket
+    /// </summary>
+    public class WebSocketParameters
+    {
+        /// <summary>
+        /// The uri to connect to
+        /// </summary>
+        public Uri Uri { get; set; }
+        /// <summary>
+        /// Headers to send in the connection handshake
+        /// </summary>
+        public IDictionary<string, string> Headers { get; set; } = new Dictionary<string, string>();
+        /// <summary>
+        /// Cookies to send in the connection handshake
+        /// </summary>
+        public IDictionary<string, string> Cookies { get; set; } = new Dictionary<string, string>();
+        /// <summary>
+        /// The time to wait between reconnect attempts
+        /// </summary>
+        public TimeSpan ReconnectInterval { get; set; } = TimeSpan.FromSeconds(5);
+        /// <summary>
+        /// Proxy for the connection
+        /// </summary>
+        public ApiProxy? Proxy { get; set; }
+        /// <summary>
+        /// Whether the socket should automatically reconnect when connection is lost
+        /// </summary>
+        public bool AutoReconnect { get; set; }
+        /// <summary>
+        /// The maximum time of no data received before considering the connection lost and closting/reconnecting the socket
+        /// </summary>
+        public TimeSpan? Timeout { get; set; }
+        /// <summary>
+        /// Interval at which to send ping frames
+        /// </summary>
+        public TimeSpan? KeepAliveInterval { get; set; }
+        /// <summary>
+        /// The max amount of messages to send per second
+        /// </summary>
+        public int? RatelimitPerSecond { get; set; }
+        /// <summary>
+        /// Origin header value to send in the connection handshake
+        /// </summary>
+        public string? Origin { get; set; }
+        /// <summary>
+        /// Delegate used for processing byte data received from socket connections before it is processed by handlers
+        /// </summary>
+        public Func<byte[], string>? DataInterpreterBytes { get; set; }
+
+        /// <summary>
+        /// Delegate used for processing string data received from socket connections before it is processed by handlers
+        /// </summary>
+        public Func<string, string>? DataInterpreterString { get; set; }
+
+        /// <summary>
+        /// Encoding for sending/receiving data
+        /// </summary>
+        public Encoding Encoding { get; set; } = Encoding.UTF8;
+
+        /// <summary>
+        /// ctor
+        /// </summary>
+        /// <param name="uri">Uri</param>
+        /// <param name="autoReconnect">Auto reconnect</param>
+        public WebSocketParameters(Uri uri, bool autoReconnect)
+        {
+            Uri = uri;
+            AutoReconnect = autoReconnect;
+        }
+    }
+}

CryptoExchange.Net/Sockets/WebsocketFactory.cs

@@ -1,6 +1,4 @@
-using System;
-using System.Collections.Generic;
-using CryptoExchange.Net.Interfaces;
+using CryptoExchange.Net.Interfaces;
 using CryptoExchange.Net.Logging;
 
 namespace CryptoExchange.Net.Sockets
@@ -11,15 +9,9 @@ namespace CryptoExchange.Net.Sockets
     public class WebsocketFactory : IWebsocketFactory
     {
         /// <inheritdoc />
-        public IWebsocket CreateWebsocket(Log log, string url)
+        public IWebsocket CreateWebsocket(Log log, WebSocketParameters parameters)
         {
-            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, new Uri(url), cookies, headers);
+            return new CryptoExchangeWebSocketClient(log, parameters);
         }
     }
 }

README.md

@@ -18,6 +18,46 @@ 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.2.2 - 17 Jul 2022
+    * Added support for retrieving a new url when socket connection is lost and reconnection will happen
+
+* Version 5.2.1 - 16 Jul 2022
+    * Fixed socket reconnect issue
+    * Fixed `message not handled` messages after unsubscribing
+    * Fixed error returning for non-json error responses
+
+* Version 5.2.0 - 10 Jul 2022
+    * Refactored websocket code, removed some clutter and simplified
+    * Added ReconnectAsync and GetSubscriptionsState methods on socket clients
+
+* Version 5.1.12 - 12 Jun 2022
+    * Changed time sync so requests no longer wait for it to complete unless it's the first time
+    * Made log client options changable after client creation
+    * Fixed proxy setting not used when reconnecting socket
+    * Changed MaxSocketConnections to a client options
+    * Updated socket reconnection logic
+
+* Version 5.1.12 - 12 Jun 2022
+    * Changed time sync so requests no longer wait for it to complete unless it's the first time
+    * Made log client options changable after client creation
+    * Fixed proxy setting not used when reconnecting socket
+    * Updated socket reconnection logic
+
+* Version 5.1.11 - 24 May 2022
+    * Added KeepAliveInterval setting
+    * Fixed port not being copied when setting parameters on request
+    * Fixed inconsistent PackageReference casing in csproj
+
+* Version 5.1.10 - 22 May 2022
+    * Fixed order book reconnecting while Diposed
+    * Fixed exception when disposing socket client while reconnecting
+    * Added additional null/default checking in DateTimeConverter
+    * Changed ConnectionLost subscription event to run in seperate task to prevent exception/longer operations from intervering with reconnecting
+
+* 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

docs/FAQ.md

@@ -61,4 +61,7 @@ var client = new BinanceClient(new BinanceClientOptions
 		BaseAddress = BinanceApiAddresses.TestNet.UsdFuturesRestClientAddress
 	}
 });
-```
+```
+
+### How are timezones handled / Timestamps are off by xx
+Exchange API's treat all timestamps as UTC, both incoming and outgoing. The client libraries do no conversion so be sure to use UTC as well.