1
0
mirror of https://github.com/JKorf/CryptoExchange.Net synced 2025-06-06 23:46:12 +00:00

Improved response time on CancellationToken cancel during subscribing

This commit is contained in:
Jkorf 2025-05-20 14:49:43 +02:00
parent 2cf10668dd
commit 7da8cedf66
8 changed files with 41 additions and 20 deletions

View File

@ -114,7 +114,7 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
public CallResult ConnectSocketSub(SocketConnection sub) public CallResult ConnectSocketSub(SocketConnection sub)
{ {
return ConnectSocketAsync(sub).Result; return ConnectSocketAsync(sub, default).Result;
} }
public override string GetListenerIdentifier(IMessageAccessor message) public override string GetListenerIdentifier(IMessageAccessor message)

View File

@ -244,7 +244,7 @@ namespace CryptoExchange.Net.Clients
var needsConnecting = !socketConnection.Connected; var needsConnecting = !socketConnection.Connected;
var connectResult = await ConnectIfNeededAsync(socketConnection, subscription.Authenticated).ConfigureAwait(false); var connectResult = await ConnectIfNeededAsync(socketConnection, subscription.Authenticated, ct).ConfigureAwait(false);
if (!connectResult) if (!connectResult)
return new CallResult<UpdateSubscription>(connectResult.Error!); return new CallResult<UpdateSubscription>(connectResult.Error!);
@ -268,7 +268,7 @@ namespace CryptoExchange.Net.Clients
if (subQuery != null) if (subQuery != null)
{ {
// Send the request and wait for answer // Send the request and wait for answer
var subResult = await socketConnection.SendAndWaitQueryAsync(subQuery, waitEvent).ConfigureAwait(false); var subResult = await socketConnection.SendAndWaitQueryAsync(subQuery, waitEvent, ct).ConfigureAwait(false);
if (!subResult) if (!subResult)
{ {
waitEvent?.Set(); waitEvent?.Set();
@ -352,7 +352,7 @@ namespace CryptoExchange.Net.Clients
released = true; released = true;
} }
var connectResult = await ConnectIfNeededAsync(socketConnection, query.Authenticated).ConfigureAwait(false); var connectResult = await ConnectIfNeededAsync(socketConnection, query.Authenticated, ct).ConfigureAwait(false);
if (!connectResult) if (!connectResult)
return new CallResult<THandlerResponse>(connectResult.Error!); return new CallResult<THandlerResponse>(connectResult.Error!);
} }
@ -379,13 +379,14 @@ namespace CryptoExchange.Net.Clients
/// </summary> /// </summary>
/// <param name="socket">The connection to check</param> /// <param name="socket">The connection to check</param>
/// <param name="authenticated">Whether the socket should authenticated</param> /// <param name="authenticated">Whether the socket should authenticated</param>
/// <param name="ct">Cancellation token</param>
/// <returns></returns> /// <returns></returns>
protected virtual async Task<CallResult> ConnectIfNeededAsync(SocketConnection socket, bool authenticated) protected virtual async Task<CallResult> ConnectIfNeededAsync(SocketConnection socket, bool authenticated, CancellationToken ct)
{ {
if (socket.Connected) if (socket.Connected)
return CallResult.SuccessResult; return CallResult.SuccessResult;
var connectResult = await ConnectSocketAsync(socket).ConfigureAwait(false); var connectResult = await ConnectSocketAsync(socket, ct).ConfigureAwait(false);
if (!connectResult) if (!connectResult)
return connectResult; return connectResult;
@ -579,10 +580,11 @@ namespace CryptoExchange.Net.Clients
/// Connect a socket /// Connect a socket
/// </summary> /// </summary>
/// <param name="socketConnection">The socket to connect</param> /// <param name="socketConnection">The socket to connect</param>
/// <param name="ct">Cancellation token</param>
/// <returns></returns> /// <returns></returns>
protected virtual async Task<CallResult> ConnectSocketAsync(SocketConnection socketConnection) protected virtual async Task<CallResult> ConnectSocketAsync(SocketConnection socketConnection, CancellationToken ct)
{ {
var connectResult = await socketConnection.ConnectAsync().ConfigureAwait(false); var connectResult = await socketConnection.ConnectAsync(ct).ConfigureAwait(false);
if (connectResult) if (connectResult)
{ {
socketConnections.TryAdd(socketConnection.SocketId, socketConnection); socketConnections.TryAdd(socketConnection.SocketId, socketConnection);
@ -714,7 +716,7 @@ namespace CryptoExchange.Net.Clients
if (!socketResult) if (!socketResult)
return socketResult.AsDataless(); return socketResult.AsDataless();
var connectResult = await ConnectIfNeededAsync(socketResult.Data, item.Authenticated).ConfigureAwait(false); var connectResult = await ConnectIfNeededAsync(socketResult.Data, item.Authenticated, default).ConfigureAwait(false);
if (!connectResult) if (!connectResult)
return new CallResult(connectResult.Error!); return new CallResult(connectResult.Error!);
} }

View File

@ -1,6 +1,7 @@
using CryptoExchange.Net.Objects; using CryptoExchange.Net.Objects;
using System; using System;
using System.Net.WebSockets; using System.Net.WebSockets;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace CryptoExchange.Net.Interfaces namespace CryptoExchange.Net.Interfaces
@ -75,7 +76,7 @@ namespace CryptoExchange.Net.Interfaces
/// Connect the socket /// Connect the socket
/// </summary> /// </summary>
/// <returns></returns> /// <returns></returns>
Task<CallResult> ConnectAsync(); Task<CallResult> ConnectAsync(CancellationToken ct);
/// <summary> /// <summary>
/// Send data /// Send data
/// </summary> /// </summary>

View File

@ -8,6 +8,7 @@ namespace CryptoExchange.Net.Logging.Extensions
{ {
private static readonly Action<ILogger, int, Exception?> _connecting; private static readonly Action<ILogger, int, Exception?> _connecting;
private static readonly Action<ILogger, int, string, Exception?> _connectionFailed; private static readonly Action<ILogger, int, string, Exception?> _connectionFailed;
private static readonly Action<ILogger, int, Exception?> _connectingCanceled;
private static readonly Action<ILogger, int, Uri, Exception?> _connected; private static readonly Action<ILogger, int, Uri, Exception?> _connected;
private static readonly Action<ILogger, int, Exception?> _startingProcessing; private static readonly Action<ILogger, int, Exception?> _startingProcessing;
private static readonly Action<ILogger, int, Exception?> _finishedProcessing; private static readonly Action<ILogger, int, Exception?> _finishedProcessing;
@ -189,6 +190,12 @@ namespace CryptoExchange.Net.Logging.Extensions
new EventId(1030, "SocketPingTimeout"), new EventId(1030, "SocketPingTimeout"),
"[Sckt {Id}] ping frame timeout; reconnecting socket"); "[Sckt {Id}] ping frame timeout; reconnecting socket");
_connectingCanceled = LoggerMessage.Define<int>(
LogLevel.Debug,
new EventId(1031, "ConnectingCanceled"),
"[Sckt {SocketId}] connecting canceled");
} }
public static void SocketConnecting( public static void SocketConnecting(
@ -370,5 +377,11 @@ namespace CryptoExchange.Net.Logging.Extensions
{ {
_socketPingTimeout(logger, socketId, null); _socketPingTimeout(logger, socketId, null);
} }
public static void SocketConnectingCanceled(
this ILogger logger, int socketId)
{
_connectingCanceled(logger, socketId, null);
}
} }
} }

View File

@ -166,9 +166,9 @@ namespace CryptoExchange.Net.Sockets
} }
/// <inheritdoc /> /// <inheritdoc />
public virtual async Task<CallResult> ConnectAsync() public virtual async Task<CallResult> ConnectAsync(CancellationToken ct)
{ {
var connectResult = await ConnectInternalAsync().ConfigureAwait(false); var connectResult = await ConnectInternalAsync(ct).ConfigureAwait(false);
if (!connectResult) if (!connectResult)
return connectResult; return connectResult;
@ -215,7 +215,7 @@ namespace CryptoExchange.Net.Sockets
return socket; return socket;
} }
private async Task<CallResult> ConnectInternalAsync() private async Task<CallResult> ConnectInternalAsync(CancellationToken ct)
{ {
_logger.SocketConnecting(Id); _logger.SocketConnecting(Id);
try try
@ -229,12 +229,16 @@ namespace CryptoExchange.Net.Sockets
} }
using CancellationTokenSource tcs = new(TimeSpan.FromSeconds(10)); using CancellationTokenSource tcs = new(TimeSpan.FromSeconds(10));
using var linked = CancellationTokenSource.CreateLinkedTokenSource(tcs.Token, _ctsSource.Token); using var linked = CancellationTokenSource.CreateLinkedTokenSource(tcs.Token, _ctsSource.Token, ct);
await _socket.ConnectAsync(Uri, linked.Token).ConfigureAwait(false); await _socket.ConnectAsync(Uri, linked.Token).ConfigureAwait(false);
} }
catch (Exception e) catch (Exception e)
{ {
if (!_ctsSource.IsCancellationRequested) if (ct.IsCancellationRequested)
{
_logger.SocketConnectingCanceled(Id);
}
else if (!_ctsSource.IsCancellationRequested)
{ {
// if _ctsSource was canceled this was already logged // if _ctsSource was canceled this was already logged
_logger.SocketConnectionFailed(Id, e.Message, e); _logger.SocketConnectionFailed(Id, e.Message, e);
@ -325,7 +329,7 @@ namespace CryptoExchange.Net.Sockets
while (_sendBuffer.TryDequeue(out _)) { } // Clear send buffer while (_sendBuffer.TryDequeue(out _)) { } // Clear send buffer
_reconnectAttempt++; _reconnectAttempt++;
var connected = await ConnectInternalAsync().ConfigureAwait(false); var connected = await ConnectInternalAsync(default).ConfigureAwait(false);
if (!connected) if (!connected)
{ {
// Delay between reconnect attempts // Delay between reconnect attempts

View File

@ -571,7 +571,7 @@ namespace CryptoExchange.Net.Sockets
/// Connect the websocket /// Connect the websocket
/// </summary> /// </summary>
/// <returns></returns> /// <returns></returns>
public async Task<CallResult> ConnectAsync() => await _socket.ConnectAsync().ConfigureAwait(false); public async Task<CallResult> ConnectAsync(CancellationToken ct) => await _socket.ConnectAsync(ct).ConfigureAwait(false);
/// <summary> /// <summary>
/// Retrieve the underlying socket /// Retrieve the underlying socket

View File

@ -2,6 +2,7 @@
using System.Net.WebSockets; using System.Net.WebSockets;
using System.Text; using System.Text;
using System.Text.Json; using System.Text.Json;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using CryptoExchange.Net.Interfaces; using CryptoExchange.Net.Interfaces;
using CryptoExchange.Net.Objects; using CryptoExchange.Net.Objects;
@ -50,7 +51,7 @@ namespace CryptoExchange.Net.Testing.Implementations
} }
} }
public Task<CallResult> ConnectAsync() public Task<CallResult> ConnectAsync(CancellationToken ct)
{ {
Connected = CanConnect; Connected = CanConnect;
return Task.FromResult(CanConnect ? new CallResult(null) : new CallResult(new CantConnectError())); return Task.FromResult(CanConnect ? new CallResult(null) : new CallResult(new CantConnectError()));

View File

@ -98,9 +98,9 @@ namespace CryptoExchange.Net.Testing
{ {
var task = methodInvoke(_client, x => { update = x.Data; }); var task = methodInvoke(_client, x => { update = x.Data; });
} }
catch(Exception ex) catch(Exception)
{ {
throw;
} }
var replaceValues = new Dictionary<string, string>(); var replaceValues = new Dictionary<string, string>();