Fixed socket individual subscription target calculation (#282)

* Fixed socket individual subscription target calculation

* Fixed socket selection when the least-loaded connection has reached its individual subscription limit

CryptoExchange.Net.UnitTests/ClientTests/SocketClientTests.cs

@@ -138,6 +138,68 @@ namespace CryptoExchange.Net.UnitTests.ClientTests
             Assert.That(socket2.Connected == false);
         }
 
+        [TestCase()]
+        public async Task BatchedSubscription_Should_NotExceedIndividualCombineTarget()
+        {
+            // arrange
+            var client = new TestSocketClient(options =>
+            {
+                options.SocketSubscriptionsCombineTarget = 10;
+                options.SocketIndividualSubscriptionCombineTarget = 10;
+            });
+            TestHelpers.ConfigureSocketClient(client, "wss://localhost");
+
+            // act
+            await client.ApiClient1.SubscribeToUpdatesAsync<TestObject>(x => { }, false, default, individualSubscriptionCount: 6);
+            TestHelpers.ConfigureSocketClient(client, "wss://localhost");
+            await client.ApiClient1.SubscribeToUpdatesAsync<TestObject>(x => { }, false, default, individualSubscriptionCount: 6);
+
+            // assert
+            Assert.That(client.ApiClient1._socketConnections.Count == 2);
+            Assert.That(client.ApiClient1._socketConnections.Values.All(connection => connection.Subscriptions.Sum(subscription => subscription.IndividualSubscriptionCount) <= 10));
+        }
+
+        [TestCase()]
+        public async Task BatchedSubscription_FullIndividualConnection_Should_NotPreventEligibleConnectionReuse()
+        {
+            // arrange
+            var client = new TestSocketClient(options =>
+            {
+                options.SocketSubscriptionsCombineTarget = 5;
+                options.SocketIndividualSubscriptionCombineTarget = 10;
+            });
+            TestHelpers.ConfigureSocketClient(client, "wss://localhost");
+            await client.ApiClient1.SubscribeToUpdatesAsync<TestObject>(x => { }, false, default);
+            await client.ApiClient1.SubscribeToUpdatesAsync<TestObject>(x => { }, false, default);
+
+            TestHelpers.ConfigureSocketClient(client, "wss://localhost");
+            await client.ApiClient1.SubscribeToUpdatesAsync<TestObject>(x => { }, false, default, individualSubscriptionCount: 10);
+
+            TestHelpers.ConfigureSocketClient(client, "wss://localhost");
+
+            // act
+            await client.ApiClient1.SubscribeToUpdatesAsync<TestObject>(x => { }, false, default);
+
+            // assert
+            Assert.That(
+                client.ApiClient1._socketConnections.Count,
+                Is.EqualTo(2),
+                "The eligible connection should be reused instead of opening a new connection after selecting a full individual-subscription connection");
+
+            var fullConnection = client.ApiClient1._socketConnections.Values
+                .Single(connection => connection.Subscriptions.Sum(subscription => subscription.IndividualSubscriptionCount) == 10);
+            Assert.That(
+                fullConnection.UserSubscriptionCount,
+                Is.EqualTo(1),
+                "The full connection should not receive the normal subscription");
+
+            var eligibleConnection = client.ApiClient1._socketConnections.Values.Single(connection => connection != fullConnection);
+            Assert.That(
+                eligibleConnection.UserSubscriptionCount,
+                Is.EqualTo(3),
+                "The existing eligible connection should receive the normal subscription");
+        }
+
         [TestCase()]
         public async Task ErrorResponse_ShouldNot_ConfirmSubscription()
         {

CryptoExchange.Net.UnitTests/Implementations/TestSocketApiClient.cs

@@ -36,9 +36,13 @@ namespace CryptoExchange.Net.UnitTests.Implementations
         protected override TestAuthenticationProvider CreateAuthenticationProvider(TestCredentials credentials) =>
             new TestAuthenticationProvider(credentials);
 
-        public async Task<WebSocketResult<UpdateSubscription>> SubscribeToUpdatesAsync<T>(Action<DataEvent<T>> handler, bool subQuery, CancellationToken ct)
+        public async Task<WebSocketResult<UpdateSubscription>> SubscribeToUpdatesAsync<T>(Action<DataEvent<T>> handler, bool subQuery, CancellationToken ct, int individualSubscriptionCount = 1)
         {
-            return await base.SubscribeAsync(new TestSubscription<T>(_logger, handler, subQuery, false), ct);
+            var subscription = new TestSubscription<T>(_logger, handler, subQuery, false)
+            {
+                IndividualSubscriptionCount = individualSubscriptionCount
+            };
+            return await base.SubscribeAsync(subscription, ct);
         }
     }
 }

CryptoExchange.Net/Clients/SocketApiClient.cs

@@ -679,45 +679,26 @@ namespace CryptoExchange.Net.Clients
                                                 && (s.Authenticated == authenticated || !authenticated)
                                                 && s.Connected).ToList();
 
-            SocketConnection? connection;
+            bool maxConnectionsReached = _socketConnections.Count >= (ApiOptions.MaxSocketConnections ?? ClientOptions.MaxSocketConnections);
-            if (!dedicatedRequestConnection)
+            SocketConnection? connection = null;
-            {
+            if (dedicatedRequestConnection)
-                connection = socketQuery
-                    .Where(s => !s.DedicatedRequestConnection.IsDedicatedRequestConnection)
-                    .OrderBy(s => s.UserSubscriptionCount)
-                    .FirstOrDefault();
-            }
-            else
             {
                 connection = socketQuery.Where(s => s.DedicatedRequestConnection.IsDedicatedRequestConnection).FirstOrDefault();
                 if (connection != null && !connection.DedicatedRequestConnection.Authenticated)
                     // Mark dedicated request connection as authenticated if the request is authenticated
                     connection.DedicatedRequestConnection.Authenticated = authenticated;
-
-                if (connection == null)
-                    // Fall back to an existing connection if there is no dedicated request connection available
-                    connection = socketQuery.OrderBy(s => s.UserSubscriptionCount).FirstOrDefault();
             }
 
-            bool maxConnectionsReached = _socketConnections.Count >= (ApiOptions.MaxSocketConnections ?? ClientOptions.MaxSocketConnections);
+            if (connection == null)
+                // Use an eligible non-dedicated connection for subscriptions, or as fallback when no dedicated request connection is available
+                connection = socketQuery
+                    .Where(s => !s.DedicatedRequestConnection.IsDedicatedRequestConnection)
+                    .Where(s => IsConnectionEligible(s, individualSubscriptionCount, maxConnectionsReached))
+                    .OrderBy(s => s.UserSubscriptionCount)
+                    .FirstOrDefault();
+
             if (connection != null)
-            {
+                return CallResult.Ok(connection);
-                bool lessThanBatchSubCombineTarget = connection.UserSubscriptionCount < ClientOptions.SocketSubscriptionsCombineTarget;
-                bool lessThanIndividualSubCombineTarget = connection.Subscriptions.Sum(x => x.IndividualSubscriptionCount) < ClientOptions.SocketIndividualSubscriptionCombineTarget;
-
-                if ((lessThanBatchSubCombineTarget && lessThanIndividualSubCombineTarget)
-                    || maxConnectionsReached) 
-                {
-                    // Use existing socket if it has less than target connections OR it has the least connections and we can't make new
-                    // If there is a max subscriptions per connection limit also only use existing if the new subscription doesn't go over the limit
-                    if (MaxIndividualSubscriptionsPerConnection == null)
-                        return CallResult.Ok(connection);
-                                        
-                    var currentCount = connection.Subscriptions.Sum(x => x.IndividualSubscriptionCount);
-                    if (currentCount + individualSubscriptionCount <= MaxIndividualSubscriptionsPerConnection)
-                        return CallResult.Ok(connection);
-                }
-            }
 
             if (maxConnectionsReached)
                 return CallResult.Fail<SocketConnection>(new InvalidOperationError("Max amount of socket connections reached"));
@@ -784,6 +765,21 @@ namespace CryptoExchange.Net.Clients
             return CallResult.Ok(socketConnection);
         }
 
+        private bool IsConnectionEligible(SocketConnection socketConnection, int individualSubscriptionCount, bool maxConnectionsReached)
+        {
+            var currentIndividualSubscriptionCount = socketConnection.Subscriptions.Sum(x => x.IndividualSubscriptionCount);
+            bool lessThanBatchSubCombineTarget = socketConnection.UserSubscriptionCount < ClientOptions.SocketSubscriptionsCombineTarget;
+            // Include the incoming batch so batched subscriptions cannot overshoot the configured socket target.
+            bool lessThanIndividualSubCombineTarget = currentIndividualSubscriptionCount + individualSubscriptionCount <= ClientOptions.SocketIndividualSubscriptionCombineTarget;
+
+            if ((!lessThanBatchSubCombineTarget || !lessThanIndividualSubCombineTarget)
+                && !maxConnectionsReached)
+                return false;
+
+            return MaxIndividualSubscriptionsPerConnection == null
+                || currentIndividualSubscriptionCount + individualSubscriptionCount <= MaxIndividualSubscriptionsPerConnection;
+        }
+
         /// <summary>
         /// Process an unhandled message
         /// </summary>