flywithu/CryptoExchange.Net
1
0
Fork 0
mirror of https://github.com/JKorf/CryptoExchange.Net synced 2025-06-08 16:36:15 +00:00
Code Issues Releases Wiki Activity

wip

Browse Source
This commit is contained in:
JKorf 2023-11-05 21:21:25 +01:00
parent 081c2d4268
commit 3de04e4828
6 changed files with 123 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
CryptoExchange.Net/Converters/SocketConverter.cs
View File

@ -5,6 +5,7 @@ using Newtonsoft.Json.Linq;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
namespace CryptoExchange.Net.Converters
@ -33,6 +34,8 @@ namespace CryptoExchange.Net.Converters
/// <returns></returns>
public abstract Type? GetDeserializationType(Dictionary<string, string?> idValues, List<BasePendingRequest> pendingRequests, List<Subscription> listeners);
public virtual string CreateIdentifierString(Dictionary<string, string?> idValues) => string.Join("-", idValues.Values.Where(v => v != null).Select(v => v!.ToLower()));
/// <inheritdoc />
public BaseParsedMessage? ReadJson(Stream stream, List<BasePendingRequest> pendingRequests, List<Subscription> listeners, bool outputOriginalData)
{
@ -61,19 +64,15 @@ namespace CryptoExchange.Net.Converters
}
var typeIdDict = new Dictionary<string, string?>();
string idString = "";
foreach (var idField in TypeIdFields)
{
var val = GetValueForKey(token, idField);
idString += val;
typeIdDict[idField] = val;
}
typeIdDict[idField] = GetValueForKey(token, idField);
Dictionary<string, string?>? subIdDict = null;
if (SubscriptionIdFields != null)
{
idString = "";
subIdDict = new Dictionary<string, string?>();
foreach (var idField in SubscriptionIdFields)
idString += GetValueForKey(token, idField);
subIdDict[idField] = GetValueForKey(token, idField);
}
var resultType = GetDeserializationType(typeIdDict, pendingRequests, listeners);
@ -91,7 +90,7 @@ namespace CryptoExchange.Net.Converters
instance.OriginalData = sr.ReadToEnd();
}
instance.Identifier = idString;
instance.Identifier = CreateIdentifierString(subIdDict ?? typeIdDict);
instance.Parsed = resultType != null;
return instance;
}

15
CryptoExchange.Net/Interfaces/IMessageProcessor.cs Normal file
View File

@ -0,0 +1,15 @@
using CryptoExchange.Net.Objects;
using CryptoExchange.Net.Objects.Sockets;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
namespace CryptoExchange.Net.Interfaces
{
public interface IMessageProcessor
{
public int Id { get; }
Task<CallResult> HandleMessageAsync(DataEvent<BaseParsedMessage> message);
}
}

28
CryptoExchange.Net/Objects/Sockets/PendingRequest.cs
View File

@ -15,10 +15,6 @@ namespace CryptoExchange.Net.Objects.Sockets
/// </summary>
public int Id { get; set; }
/// <summary>
/// Callback for checking if a message is a response to this request
/// </summary>
public Func<BaseParsedMessage, bool> MessageMatchesHandler { get; }
/// <summary>
/// If the request is completed
/// </summary>
public bool Completed { get; protected set; }
@ -48,12 +44,10 @@ namespace CryptoExchange.Net.Objects.Sockets
/// </summary>
/// <param name="id"></param>
/// <param name="request"></param>
/// <param name="messageMatchesHandler"></param>
/// <param name="timeout"></param>
protected BasePendingRequest(int id, object request, Func<BaseParsedMessage, bool> messageMatchesHandler, TimeSpan timeout)
protected BasePendingRequest(int id, object request, TimeSpan timeout)
{
Id = id;
MessageMatchesHandler = messageMatchesHandler;
_event = new AsyncResetEvent(false, false);
_timeout = timeout;
Request = request;
@ -93,7 +87,7 @@ namespace CryptoExchange.Net.Objects.Sockets
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
public abstract void ProcessAsync(BaseParsedMessage message);
public abstract Task ProcessAsync(DataEvent<BaseParsedMessage> message);
}
/// <summary>
@ -111,7 +105,7 @@ namespace CryptoExchange.Net.Objects.Sockets
/// <summary>
/// Data handler
/// </summary>
public Func<ParsedMessage<T>, CallResult<T>> Handler { get; }
public Func<DataEvent<ParsedMessage<T>>, Task<CallResult<T>>> Handler { get; }
/// <summary>
/// The response object type
/// </summary>
@ -122,11 +116,10 @@ namespace CryptoExchange.Net.Objects.Sockets
/// </summary>
/// <param name="id"></param>
/// <param name="request"></param>
/// <param name="messageMatchesHandler"></param>
/// <param name="messageHandler"></param>
/// <param name="timeout"></param>
private PendingRequest(int id, object request, Func<ParsedMessage<T>, bool> messageMatchesHandler, Func<ParsedMessage<T>, CallResult<T>> messageHandler, TimeSpan timeout)
: base(id, request, (x) => messageMatchesHandler((ParsedMessage<T>)x), timeout)
private PendingRequest(int id, object request, Func<DataEvent<ParsedMessage<T>>, Task<CallResult<T>>> messageHandler, TimeSpan timeout)
: base(id, request, timeout)
{
Handler = messageHandler;
}
@ -135,12 +128,13 @@ namespace CryptoExchange.Net.Objects.Sockets
/// Create a new pending request for provided query
/// </summary>
/// <param name="query"></param>
/// <param name="id"></param>
/// <returns></returns>
public static PendingRequest<T> CreateForQuery(Query<T> query)
public static PendingRequest<T> CreateForQuery(Query<T> query, int id)
{
return new PendingRequest<T>(ExchangeHelpers.NextId(), query.Request, query.MessageMatchesQuery, x =>
return new PendingRequest<T>(id, query.Request, async x =>
{
var response = query.HandleResponse(x);
var response = await query.HandleMessageAsync(x).ConfigureAwait(false);
return response.As(response.Data);
}, TimeSpan.FromSeconds(5));
}
@ -161,10 +155,10 @@ namespace CryptoExchange.Net.Objects.Sockets
}
/// <inheritdoc />
public override void ProcessAsync(BaseParsedMessage message)
public override async Task ProcessAsync(DataEvent<BaseParsedMessage> message)
{
Completed = true;
Result = Handler((ParsedMessage<T>)message);
Result = await Handler(message.As((ParsedMessage<T>)message.Data)).ConfigureAwait(false);
_event.Set();
}
}

49
CryptoExchange.Net/Sockets/Query.cs
View File

@ -1,13 +1,22 @@
using CryptoExchange.Net.Objects;
using CryptoExchange.Net.Interfaces;
using CryptoExchange.Net.Objects;
using CryptoExchange.Net.Objects.Sockets;
using System.Collections.Generic;
using System.Threading.Tasks;
namespace CryptoExchange.Net.Sockets
{
/// <summary>
/// Query
/// </summary>
public abstract class BaseQuery
public abstract class BaseQuery : IMessageProcessor
{
public int Id { get; } = ExchangeHelpers.NextId();
/// <summary>
/// Strings to identify this subscription with
/// </summary>
public abstract List<string> Identifiers { get; }
/// <summary>
/// The query request object
/// </summary>
@ -23,6 +32,8 @@ namespace CryptoExchange.Net.Sockets
/// </summary>
public int Weight { get; }
public BasePendingRequest PendingRequest { get; private set; }
/// <summary>
/// ctor
/// </summary>
@ -39,7 +50,21 @@ namespace CryptoExchange.Net.Sockets
/// <summary>
/// Create a pending request for this query
/// </summary>
public abstract BasePendingRequest CreatePendingRequest();
public BasePendingRequest CreatePendingRequest()
{
PendingRequest = GetPendingRequest(Id);
return PendingRequest;
}
public abstract BasePendingRequest GetPendingRequest(int id);
/// <summary>
/// Handle a response message
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
public abstract Task<CallResult> HandleMessageAsync(DataEvent<BaseParsedMessage> message);
}
/// <summary>
@ -58,21 +83,21 @@ namespace CryptoExchange.Net.Sockets
{
}
/// <inheritdoc />
public override async Task<CallResult> HandleMessageAsync(DataEvent<BaseParsedMessage> message)
{
await PendingRequest.ProcessAsync(message).ConfigureAwait(false);
return await HandleMessageAsync(message.As((ParsedMessage<TResponse>)message.Data)).ConfigureAwait(false);
}
/// <summary>
/// Handle the query response
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
public abstract CallResult<TResponse> HandleResponse(ParsedMessage<TResponse> message);
/// <summary>
/// Check if the message is the response to the query
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
public abstract bool MessageMatchesQuery(ParsedMessage<TResponse> message);
public virtual Task<CallResult<TResponse>> HandleMessageAsync(DataEvent<ParsedMessage<TResponse>> message) => Task.FromResult(new CallResult<TResponse>(message.Data.Data!));
/// <inheritdoc />
public override BasePendingRequest CreatePendingRequest() => PendingRequest<TResponse>.CreateForQuery(this);
public override BasePendingRequest GetPendingRequest(int id) => PendingRequest<TResponse>.CreateForQuery(this, id);
}
}

92
CryptoExchange.Net/Sockets/SocketConnection.cs
View File

@ -60,7 +60,7 @@ namespace CryptoExchange.Net.Sockets
public int UserSubscriptionCount
{
get { lock (_subscriptionLock)
return _messageIdentifierSubscriptions.Values.Count(h => h.UserSubscription); }
return _subscriptions.Count(h => h.UserSubscription); }
}
/// <summary>
@ -71,7 +71,7 @@ namespace CryptoExchange.Net.Sockets
get
{
lock (_subscriptionLock)
return _messageIdentifierSubscriptions.Values.Where(h => h.UserSubscription).ToArray();
return _subscriptions.Where(h => h.UserSubscription).ToArray();
}
}
@ -158,7 +158,7 @@ namespace CryptoExchange.Net.Sockets
private bool _pausedActivity;
private readonly List<BasePendingRequest> _pendingRequests;
private readonly List<Subscription> _subscriptions;
private readonly Dictionary<string, Subscription> _messageIdentifierSubscriptions;
private readonly Dictionary<string, IMessageProcessor> _messageIdMap;
private readonly object _subscriptionLock = new();
private readonly ILogger _logger;
@ -186,7 +186,7 @@ namespace CryptoExchange.Net.Sockets
_pendingRequests = new List<BasePendingRequest>();
_subscriptions = new List<Subscription>();
_messageIdentifierSubscriptions = new Dictionary<string, Subscription>();
_messageIdMap = new Dictionary<string, IMessageProcessor>();
_socket = socket;
_socket.OnStreamMessage += HandleStreamMessage;
@ -346,48 +346,40 @@ namespace CryptoExchange.Net.Sockets
}
// TODO lock
if (_messageIdentifierSubscriptions.TryGetValue(result.Identifier.ToLowerInvariant(), out var idSubscription))
if (!_messageIdMap.TryGetValue(result.Identifier, out var messageProcessor))
{
// Matched based on identifier
var userSw = Stopwatch.StartNew();
var dataEvent = new DataEvent<BaseParsedMessage>(result, null, result.OriginalData, DateTime.UtcNow, null);
await idSubscription.HandleEventAsync(dataEvent).ConfigureAwait(false);
userSw.Stop();
return;
}
List<BasePendingRequest> pendingRequests;
lock (_pendingRequests)
pendingRequests = _pendingRequests.ToList();
foreach (var pendingRequest in pendingRequests)
{
if (pendingRequest.MessageMatchesHandler(result))
{
lock (_pendingRequests)
_pendingRequests.Remove(pendingRequest);
if (pendingRequest.Completed)
{
// Answer to a timed out request
_logger.Log(LogLevel.Warning, $"Socket {SocketId} Received after request timeout. Consider increasing the RequestTimeout");
}
else
{
_logger.Log(LogLevel.Trace, $"Socket {SocketId} - msg {pendingRequest.Id} - received data matched to pending request");
pendingRequest.ProcessAsync(result);
}
return;
}
}
stream.Position = 0;
var unhandledBuffer = new byte[stream.Length];
stream.Read(unhandledBuffer, 0, unhandledBuffer.Length);
_logger.Log(LogLevel.Warning, $"Socket {SocketId} Message not handled: " + Encoding.UTF8.GetString(unhandledBuffer));
_logger.Log(LogLevel.Warning, $"Socket {SocketId} Message unidentified. Id: {result.Identifier.ToLowerInvariant()}, Message: {Encoding.UTF8.GetString(unhandledBuffer)} ");
UnhandledMessage?.Invoke(result);
return;
}
_logger.Log(LogLevel.Trace, $"Socket {SocketId} Message mapped to processor {messageProcessor.Id} with identifier {result.Identifier}");
if (messageProcessor is BaseQuery query)
{
foreach (var id in query.Identifiers)
_messageIdMap.Remove(id);
if (query.PendingRequest != null)
_pendingRequests.Remove(query.PendingRequest);
if (query.PendingRequest?.Completed == true)
{
// Answer to a timed out request
_logger.Log(LogLevel.Warning, $"Socket {SocketId} Received after request timeout. Consider increasing the RequestTimeout");
}
}
// Matched based on identifier
var userSw = Stopwatch.StartNew();
var dataEvent = new DataEvent<BaseParsedMessage>(result, null, result.OriginalData, DateTime.UtcNow, null);
await messageProcessor.HandleMessageAsync(dataEvent).ConfigureAwait(false);
userSw.Stop();
}
/// <summary>
@ -468,7 +460,7 @@ namespace CryptoExchange.Net.Sockets
return;
}
shouldCloseConnection = _messageIdentifierSubscriptions.All(r => !r.Value.UserSubscription || r.Value.Closed);
shouldCloseConnection = _subscriptions.All(r => !r.UserSubscription || r.Closed);
if (shouldCloseConnection)
Status = SocketStatus.Closing;
}
@ -483,7 +475,7 @@ namespace CryptoExchange.Net.Sockets
{
_subscriptions.Remove(subscription);
foreach (var id in subscription.Identifiers)
_messageIdentifierSubscriptions.Remove(id);
_messageIdMap.Remove(id);
}
}
@ -511,7 +503,7 @@ namespace CryptoExchange.Net.Sockets
if (subscription.Identifiers != null)
{
foreach (var id in subscription.Identifiers)
_messageIdentifierSubscriptions.Add(id.ToLowerInvariant(), subscription);
_messageIdMap.Add(id.ToLowerInvariant(), subscription);
}
if (subscription.UserSubscription)
@ -549,6 +541,12 @@ namespace CryptoExchange.Net.Sockets
public virtual async Task<CallResult> SendAndWaitQueryAsync(BaseQuery query)
{
var pendingRequest = query.CreatePendingRequest();
if (query.Identifiers != null)
{
foreach (var id in query.Identifiers)
_messageIdMap.Add(id.ToLowerInvariant(), query);
}
await SendAndWaitAsync(pendingRequest, query.Weight).ConfigureAwait(false);
return pendingRequest.Result!;
}
@ -561,7 +559,13 @@ namespace CryptoExchange.Net.Sockets
/// <returns></returns>
public virtual async Task<CallResult<T>> SendAndWaitQueryAsync<T>(Query<T> query)
{
var pendingRequest = PendingRequest<T>.CreateForQuery(query);
var pendingRequest = PendingRequest<T>.CreateForQuery(query, query.Id);
if (query.Identifiers != null)
{
foreach (var id in query.Identifiers)
_messageIdMap.Add(id.ToLowerInvariant(), query);
}
await SendAndWaitAsync(pendingRequest, query.Weight).ConfigureAwait(false);
return pendingRequest.TypedResult!;
}

12
CryptoExchange.Net/Sockets/Subscription.cs
View File

@ -1,4 +1,5 @@
using CryptoExchange.Net.Objects;
using CryptoExchange.Net.Interfaces;
using CryptoExchange.Net.Objects;
using CryptoExchange.Net.Objects.Sockets;
using Microsoft.Extensions.Logging;
using System;
@ -11,7 +12,7 @@ namespace CryptoExchange.Net.Sockets
/// <summary>
/// Socket subscription
/// </summary>
public abstract class Subscription
public abstract class Subscription : IMessageProcessor
{
/// <summary>
/// Subscription id
@ -67,6 +68,7 @@ namespace CryptoExchange.Net.Sockets
{
_logger = logger;
Authenticated = authenticated;
Id = ExchangeHelpers.NextId();
}
/// <summary>
@ -86,7 +88,7 @@ namespace CryptoExchange.Net.Sockets
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
public abstract Task HandleEventAsync(DataEvent<BaseParsedMessage> message);
public abstract Task<CallResult> HandleMessageAsync(DataEvent<BaseParsedMessage> message);
/// <summary>
/// Invoke the exception event
@ -124,7 +126,7 @@ namespace CryptoExchange.Net.Sockets
}
/// <inheritdoc />
public override Task HandleEventAsync(DataEvent<BaseParsedMessage> message)
public override Task<CallResult> HandleMessageAsync(DataEvent<BaseParsedMessage> message)
=> HandleEventAsync(message.As((ParsedMessage<TEvent>)message.Data));
/// <summary>
@ -132,6 +134,6 @@ namespace CryptoExchange.Net.Sockets
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
public abstract Task HandleEventAsync(DataEvent<ParsedMessage<TEvent>> message);
public abstract Task<CallResult> HandleEventAsync(DataEvent<ParsedMessage<TEvent>> message);
}
}