Squashed commit of the following:

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
2026-08-15 02:13:01 +00:00 · 2022-07-10 19:57:10 +02:00
parent ea9375d582
commit 50715ff2f7
15 changed files with 760 additions and 747 deletions
@@ -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
@@ -120,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 />
@@ -196,10 +192,14 @@ namespace CryptoExchange.Net
                while (true)
                {
                    // Get a new or existing socket connection
-                    socketConnection = GetSocketConnection(apiClient, url, authenticated);
+                    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);
+                    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");
@@ -240,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!);
                }
@@ -250,7 +251,6 @@ namespace CryptoExchange.Net
                subscription.Confirmed = true;
            }

-            socketConnection.ShouldReconnect = true;
            if (ct != default)
            {
                subscription.CancellationTokenRegistration = ct.Register(async () =>
@@ -260,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));
        }

@@ -320,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
@@ -334,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();
            }
@@ -481,8 +484,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)
            {
@@ -504,8 +508,8 @@ namespace CryptoExchange.Net
            }

            var subscription = request == null
-                ? SocketSubscription.CreateForIdentifier(NextId(), identifier!, userSubscription, InternalHandler)
-                : SocketSubscription.CreateForRequest(NextId(), request, userSubscription, InternalHandler);
+                ? SocketSubscription.CreateForIdentifier(NextId(), identifier!, userSubscription, authenticated, InternalHandler)
+                : SocketSubscription.CreateForRequest(NextId(), request, userSubscription, authenticated, InternalHandler);
            if (!connection.AddSubscription(subscription))
                return null;
            return subscription;
@@ -519,11 +523,23 @@ 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>
        /// Gets a connection for a new subscription or query. Can be an existing if there are open position or a new one.
        /// </summary>
@@ -531,10 +547,10 @@ 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.Status == SocketConnection.SocketStatus.None || s.Value.Status == SocketConnection.SocketStatus.Connected)
-                                                  && s.Value.Uri.ToString().TrimEnd('/') == address.TrimEnd('/')
+                                                  && 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;
@@ -543,21 +559,31 @@ namespace CryptoExchange.Net
                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>
@@ -585,6 +611,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>
@@ -592,24 +635,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.KeepAliveInterval = KeepAliveInterval;
-            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;
        }

@@ -667,7 +694,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())
@@ -683,7 +709,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);
        }

@@ -697,7 +723,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);
        }

@@ -707,7 +733,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;
@@ -718,6 +744,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>
@@ -726,8 +785,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();
        }