From 4f25a6e7d6718c6a89361fcdcb75bf96e5b26aa2 Mon Sep 17 00:00:00 2001 From: Jkorf Date: Tue, 13 Oct 2026 10:08:16 +0200 Subject: [PATCH] Added processor parameter for some methods in SocketApiClient to allow specific overriding behavior --- CryptoExchange.Net/Clients/SocketApiClient.cs | 81 ++++++++++++++++--- 1 file changed, 71 insertions(+), 10 deletions(-) diff --git a/CryptoExchange.Net/Clients/SocketApiClient.cs b/CryptoExchange.Net/Clients/SocketApiClient.cs index dfed6d87..493cf010 100644 --- a/CryptoExchange.Net/Clients/SocketApiClient.cs +++ b/CryptoExchange.Net/Clients/SocketApiClient.cs @@ -286,7 +286,7 @@ namespace CryptoExchange.Net.Clients while (true) { // Get a new or existing socket connection - var socketResult = await GetSocketConnection(url, subscription.Authenticated, false, ct, subscription.Topic, subscription.IndividualSubscriptionCount).ConfigureAwait(false); + var socketResult = await GetSocketConnection(url, subscription.Authenticated, false, ct, subscription, subscription.Topic, subscription.IndividualSubscriptionCount).ConfigureAwait(false); if (!socketResult.Success) return WebSocketResult.Fail(Exchange, socketResult.Error); @@ -309,7 +309,7 @@ namespace CryptoExchange.Net.Clients var needsConnecting = !socketConnection.Connected; - var connectResult = await ConnectIfNeededAsync(socketConnection, subscription.Authenticated, ct).ConfigureAwait(false); + var connectResult = await ConnectIfNeededAsync(socketConnection, subscription.Authenticated, ct, subscription).ConfigureAwait(false); if (!connectResult.Success) return WebSocketResult.Fail(Exchange, connectResult.Error!); @@ -482,7 +482,7 @@ namespace CryptoExchange.Net.Clients await semaphoreSlim.WaitAsync().ConfigureAwait(false); try { - var socketResult = await GetSocketConnection(url, query.Authenticated, true, ct).ConfigureAwait(false); + var socketResult = await GetSocketConnection(url, query.Authenticated, true, ct, processor: query).ConfigureAwait(false); if (!socketResult.Success) return QueryResult.Fail(Exchange, socketResult.Error); @@ -495,7 +495,7 @@ namespace CryptoExchange.Net.Clients released = true; } - var connectResult = await ConnectIfNeededAsync(socketConnection, query.Authenticated, ct).ConfigureAwait(false); + var connectResult = await ConnectIfNeededAsync(socketConnection, query.Authenticated, ct, query).ConfigureAwait(false); if (!connectResult.Success) return QueryResult.Fail(Exchange, connectResult.Error!); } @@ -524,7 +524,18 @@ namespace CryptoExchange.Net.Clients /// Whether the socket should authenticated /// Cancellation token /// - protected virtual async Task ConnectIfNeededAsync(ISocketConnection socket, bool authenticated, CancellationToken ct) + protected virtual Task ConnectIfNeededAsync(ISocketConnection socket, bool authenticated, CancellationToken ct) + => ConnectIfNeededAsync(socket, authenticated, ct, null); + + /// + /// Checks if a socket needs to be connected and does so if needed. Also authenticates on the socket if needed + /// + /// The connection to check + /// Whether the socket should authenticated + /// The message processor requesting a connection + /// Cancellation token + /// + protected virtual async Task ConnectIfNeededAsync(ISocketConnection socket, bool authenticated, CancellationToken ct, IMessageProcessor? processor) { if (socket.Connected) return CallResult.Ok(); @@ -539,7 +550,7 @@ namespace CryptoExchange.Net.Clients if (socket is not SocketConnection sc) throw new InvalidOperationException("HighPerfSocketConnection not supported for authentication"); - var result = await AuthenticateSocketAsync(sc).ConfigureAwait(false); + var result = await AuthenticateSocketAsync(sc, processor).ConfigureAwait(false); if (!result.Success) await socket.CloseAsync().ConfigureAwait(false); @@ -551,13 +562,22 @@ namespace CryptoExchange.Net.Clients /// /// Socket to authenticate /// - public virtual async Task AuthenticateSocketAsync(SocketConnection socket) + public virtual Task AuthenticateSocketAsync(SocketConnection socket) + => AuthenticateSocketAsync(socket, null); + + /// + /// Authenticate a socket connection + /// + /// Socket to authenticate + /// The message processor requesting authentication + /// + public virtual async Task AuthenticateSocketAsync(SocketConnection socket, IMessageProcessor? processor) { if (GetAuthenticationProvider() == null) return CallResult.Fail(new NoApiCredentialsError()); _logger.AttemptingToAuthenticate(socket.SocketId); - var authRequest = await GetAuthenticationRequestAsync(socket).ConfigureAwait(false); + var authRequest = await GetAuthenticationRequestAsync(socket, processor).ConfigureAwait(false); if (authRequest != null) { var result = await socket.SendAndWaitQueryAsync(authRequest).ConfigureAwait(false); @@ -578,13 +598,21 @@ namespace CryptoExchange.Net.Clients return CallResult.Ok(); } + /// /// Should return the request which can be used to authenticate a socket connection /// /// - protected internal virtual Task GetAuthenticationRequestAsync(SocketConnection connection) => + protected internal virtual Task GetAuthenticationRequestAsync(SocketConnection connection) => Task.FromResult(GetAuthenticationProvider()!.GetAuthenticationQuery(this, connection)); + /// + /// Should return the request which can be used to authenticate a socket connection + /// + /// + protected internal virtual Task GetAuthenticationRequestAsync(SocketConnection connection, IMessageProcessor? processor) => + GetAuthenticationRequestAsync(connection); + /// /// Adds a system subscription. Used for example to reply to ping requests /// @@ -642,6 +670,18 @@ namespace CryptoExchange.Net.Clients && (AllowTopicsOnTheSameConnection || !connection.Topics.Contains(topic)); } + /// + /// Whether the connection can be used for a new subscription or query with the provided parameters + /// + /// The connection to check + /// The address set by the request + /// Whether the request needs an authenticated connection + /// The processor requesting a connection + /// Topic of the request + /// True if connection can be used + protected virtual bool ConnectionCanBeUsedFor(SocketConnection connection, string address, bool authenticated, IMessageProcessor? processor, string? topic = null) + => ConnectionCanBeUsedFor(connection, address, authenticated, topic); + /// /// Gets a connection for a new subscription or query. Can be an existing if there are open position or a new one. /// @@ -652,15 +692,36 @@ namespace CryptoExchange.Net.Clients /// The subscription topic, can be provided when multiple of the same topics are not allowed on a connection /// The number of individual subscriptions in this subscribe request /// + protected virtual Task> GetSocketConnection( + string address, + bool authenticated, + bool dedicatedRequestConnection, + CancellationToken ct, + string? topic = null, + int individualSubscriptionCount = 1) + => GetSocketConnection(address, authenticated, dedicatedRequestConnection, ct, null, topic, individualSubscriptionCount); + + /// + /// Gets a connection for a new subscription or query. Can be an existing if there are open position or a new one. + /// + /// The address the socket is for + /// Whether the socket should be authenticated + /// Whether a dedicated request connection should be returned + /// Cancellation token + /// The subscription topic, can be provided when multiple of the same topics are not allowed on a connection + /// The number of individual subscriptions in this subscribe request + /// The processor requesting a connection + /// protected virtual async Task> GetSocketConnection( string address, bool authenticated, bool dedicatedRequestConnection, CancellationToken ct, + IMessageProcessor? processor, string? topic = null, int individualSubscriptionCount = 1) { - var socketQuery = _socketConnections.Where(s => ConnectionCanBeUsedFor(s.Value, address, authenticated, topic)).Select(x => x.Value); // Don't ToList this so the query is executed again when called + var socketQuery = _socketConnections.Where(s => ConnectionCanBeUsedFor(s.Value, address, authenticated, processor, topic)).Select(x => x.Value); // Don't ToList this so the query is executed again when called // If all current socket connections are reconnecting or resubscribing wait for that to finish as we can probably use the existing connection var delayStart = DateTime.UtcNow;