1
0
mirror of https://github.com/JKorf/CryptoExchange.Net synced 2025-08-31 12:42:00 +00:00

Compare commits

..

3 Commits

9 changed files with 93 additions and 19 deletions

View File

@ -470,10 +470,8 @@ namespace CryptoExchange.Net.Clients
error = ParseErrorResponse((int)response.StatusCode, response.ResponseHeaders, accessor, readResult.Error?.Exception);
}
#pragma warning disable CS0618 // Type or member is obsolete
if (error.Code == null || error.Code == 0)
error.Code = (int)response.StatusCode;
#pragma warning restore CS0618 // Type or member is obsolete
return new WebCallResult<T>(response.StatusCode, response.ResponseHeaders, sw.Elapsed, responseLength, OutputOriginalData ? accessor.GetOriginalString() : null, request.RequestId, request.Uri.ToString(), request.Content, request.Method, request.GetHeaders(), ResultDataSource.Server, default, error!);
}

View File

@ -270,7 +270,7 @@ namespace CryptoExchange.Net.Clients
}
var waitEvent = new AsyncResetEvent(false);
var subQuery = subscription.GetSubQuery(socketConnection);
var subQuery = subscription.CreateSubscriptionQuery(socketConnection);
if (subQuery != null)
{
// Send the request and wait for answer

View File

@ -251,4 +251,20 @@ namespace CryptoExchange.Net.Objects
/// </summary>
DEX
}
/// <summary>
/// Timeout behavior for queries
/// </summary>
public enum TimeoutBehavior
{
/// <summary>
/// Fail the request
/// </summary>
Fail,
/// <summary>
/// Mark the query as successful
/// </summary>
Succeed
}
}

View File

@ -11,9 +11,11 @@ namespace CryptoExchange.Net.Objects
private int? _code;
/// <summary>
/// The error code from the server
/// The int error code the server returned; or the http status code int value if there was no error code.<br />
/// <br />
/// <i>Note:</i><br />
/// The <see cref="ErrorCode"/> property should be used for more generic error checking; it might contain a string error code if the server does not return an int code.
/// </summary>
[Obsolete("Use ErrorCode instead", false)]
public int? Code
{
get

View File

@ -1,6 +1,7 @@
using CryptoExchange.Net.Interfaces;
using CryptoExchange.Net.Logging.Extensions;
using CryptoExchange.Net.Objects;
using CryptoExchange.Net.Objects.Errors;
using CryptoExchange.Net.Objects.Sockets;
using CryptoExchange.Net.RateLimiting;
using Microsoft.Extensions.Logging;
@ -252,6 +253,11 @@ namespace CryptoExchange.Net.Sockets
await (OnConnectRateLimited?.Invoke() ?? Task.CompletedTask).ConfigureAwait(false);
return new CallResult(new ServerRateLimitError(we.Message, we));
}
if (_socket.HttpStatusCode == HttpStatusCode.Unauthorized)
{
return new CallResult(new ServerError(new ErrorInfo(ErrorType.Unauthorized, "Server returned status code `401` when `101` was expected")));
}
#else
// ClientWebSocket.HttpStatusCode is only available in .NET6+ https://learn.microsoft.com/en-us/dotnet/api/system.net.websockets.clientwebsocket.httpstatuscode?view=net-8.0
// Try to read 429 from the message instead

View File

@ -29,6 +29,11 @@ namespace CryptoExchange.Net.Sockets
/// </summary>
public TimeSpan? RequestTimeout { get; set; }
/// <summary>
/// What should happen if the query times out
/// </summary>
public TimeoutBehavior TimeoutBehavior { get; set; } = TimeoutBehavior.Fail;
/// <summary>
/// The number of required responses. Can be more than 1 when for example subscribing multiple symbols streams in a single request,
/// and each symbol receives it's own confirmation response
@ -183,7 +188,7 @@ namespace CryptoExchange.Net.Sockets
/// <inheritdoc />
public override async Task<CallResult> Handle(SocketConnection connection, DataEvent<object> message, MessageHandlerLink check)
{
if (!PreCheckMessage(message))
if (!PreCheckMessage(connection, message))
return CallResult.SuccessResult;
CurrentResponses++;
@ -208,18 +213,20 @@ namespace CryptoExchange.Net.Sockets
/// <summary>
/// Validate if a message is actually processable by this query
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
public virtual bool PreCheckMessage(DataEvent<object> message) => true;
public virtual bool PreCheckMessage(SocketConnection connection, DataEvent<object> message) => true;
/// <inheritdoc />
public override void Timeout()
{
if (Completed)
return;
Completed = true;
Result = new CallResult<THandlerResponse>(new TimeoutError());
if (TimeoutBehavior == TimeoutBehavior.Fail)
Result = new CallResult<THandlerResponse>(new TimeoutError());
else
Result = new CallResult<THandlerResponse>(default, null, default);
ContinueAwaiter?.Set();
_event.Set();
}

View File

@ -202,6 +202,18 @@ namespace CryptoExchange.Net.Sockets
}
}
/// <summary>
/// The number of current pending requests
/// </summary>
public int PendingRequests
{
get
{
lock (_listenersLock)
return _listeners.OfType<Query>().Where(x => !x.Completed).Count();
}
}
private bool _pausedActivity;
private readonly object _listenersLock;
private readonly List<IMessageProcessor> _listeners;
@ -519,7 +531,10 @@ namespace CryptoExchange.Net.Sockets
{
// If this message is for this listener then it is automatically confirmed, even if the subscription is not (yet) confirmed
subscriptionProcessor.Confirmed = true;
// This doesn't trigger a waiting subscribe query, should probably also somehow set the wait event for that
if (subscriptionProcessor.SubscriptionQuery?.TimeoutBehavior == TimeoutBehavior.Succeed)
// If this subscription has a query waiting for a timeout (success if there is no error response)
// then time it out now as the data is being received, so we assume it's successful
subscriptionProcessor.SubscriptionQuery.Timeout();
}
// 5. Deserialize the message
@ -996,7 +1011,7 @@ namespace CryptoExchange.Net.Sockets
return result;
}
var subQuery = subscription.GetSubQuery(this);
var subQuery = subscription.CreateSubscriptionQuery(this);
if (subQuery == null)
{
subscription.IsResubscribing = false;
@ -1031,7 +1046,7 @@ namespace CryptoExchange.Net.Sockets
internal async Task UnsubscribeAsync(Subscription subscription)
{
var unsubscribeRequest = subscription.GetUnsubQuery();
var unsubscribeRequest = subscription.CreateUnsubscriptionQuery(this);
if (unsubscribeRequest == null)
return;
@ -1044,7 +1059,7 @@ namespace CryptoExchange.Net.Sockets
if (!_socket.IsOpen)
return new CallResult(new WebError("Socket is not connected"));
var subQuery = subscription.GetSubQuery(this);
var subQuery = subscription.CreateSubscriptionQuery(this);
if (subQuery == null)
return CallResult.SuccessResult;

View File

@ -80,6 +80,16 @@ namespace CryptoExchange.Net.Sockets
/// </summary>
public string? Topic { get; set; }
/// <summary>
/// The subscribe query for this subscription
/// </summary>
public Query? SubscriptionQuery { get; private set; }
/// <summary>
/// The unsubscribe query for this subscription
/// </summary>
public Query? UnsubscriptionQuery { get; private set; }
/// <summary>
/// ctor
/// </summary>
@ -91,11 +101,21 @@ namespace CryptoExchange.Net.Sockets
Id = ExchangeHelpers.NextId();
}
/// <summary>
/// Create a new subscription query
/// </summary>
public Query? CreateSubscriptionQuery(SocketConnection connection)
{
var query = GetSubQuery(connection);
SubscriptionQuery = query;
return query;
}
/// <summary>
/// Get the subscribe query to send when subscribing
/// </summary>
/// <returns></returns>
public abstract Query? GetSubQuery(SocketConnection connection);
protected abstract Query? GetSubQuery(SocketConnection connection);
/// <summary>
/// Handle a subscription query response
@ -109,11 +129,21 @@ namespace CryptoExchange.Net.Sockets
/// <param name="message"></param>
public virtual void HandleUnsubQueryResponse(object message) { }
/// <summary>
/// Create a new unsubscription query
/// </summary>
public Query? CreateUnsubscriptionQuery(SocketConnection connection)
{
var query = GetUnsubQuery(connection);
UnsubscriptionQuery = query;
return query;
}
/// <summary>
/// Get the unsubscribe query to send when unsubscribing
/// </summary>
/// <returns></returns>
public abstract Query? GetUnsubQuery();
protected abstract Query? GetUnsubQuery(SocketConnection connection);
/// <inheritdoc />
public virtual CallResult<object> Deserialize(IMessageAccessor message, Type type) => message.Deserialize(type);

View File

@ -22,9 +22,9 @@ namespace CryptoExchange.Net.Sockets
}
/// <inheritdoc />
public override Query? GetSubQuery(SocketConnection connection) => null;
protected override Query? GetSubQuery(SocketConnection connection) => null;
/// <inheritdoc />
public override Query? GetUnsubQuery() => null;
protected override Query? GetUnsubQuery(SocketConnection connection) => null;
}
}