flywithu/CryptoExchange.Net
1
0
Fork 0
mirror of https://github.com/JKorf/CryptoExchange.Net.git synced 2026-08-19 04:13:02 +00:00
Code Issues Releases Wiki Activity

Websocket refactoring (#190)

Browse Source 
Websocket refactoring
This commit is contained in:
Jan Korf
2024-02-24 19:21:47 +01:00
committed by GitHub
parent d91755dff5
commit d533557324
585 changed files with 114831 additions and 2198 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CryptoExchange.Net/Clients/SocketApiClient.cs
+165 -331
View File
@@ -1,18 +1,21 @@
using CryptoExchange.Net.Interfaces;
using CryptoExchange.Net.Objects;
using CryptoExchange.Net.Objects.Options;
using CryptoExchange.Net.Objects.Sockets;
using CryptoExchange.Net.Sockets;
using CryptoExchange.Net.Sockets.MessageParsing.Interfaces;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json.Linq;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net.Sockets;
using System.Net.WebSockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using static CryptoExchange.Net.Objects.RateLimiter;
namespace CryptoExchange.Net
{
@@ -40,47 +43,31 @@ namespace CryptoExchange.Net
/// </summary>
protected TimeSpan KeepAliveInterval { get; set; } = TimeSpan.FromSeconds(10);
/// <summary>
/// Delegate used for processing byte data received from socket connections before it is processed by handlers
/// </summary>
protected Func<byte[], string>? dataInterpreterBytes;
/// <summary>
/// Delegate used for processing string data received from socket connections before it is processed by handlers
/// </summary>
protected Func<string, string>? dataInterpreterString;
/// <summary>
/// Handlers for data from the socket which doesn't need to be forwarded to the caller. Ping or welcome messages for example.
/// </summary>
protected Dictionary<string, Action<MessageEvent>> genericHandlers = new();
/// <summary>
/// The task that is sending periodic data on the websocket. Can be used for sending Ping messages every x seconds or similair. Not necesarry.
/// </summary>
protected Task? periodicTask;
/// <summary>
/// Wait event for the periodicTask
/// </summary>
protected AsyncResetEvent? periodicEvent;
/// <summary>
/// If true; data which is a response to a query will also be distributed to subscriptions
/// If false; data which is a response to a query won't get forwarded to subscriptions as well
/// </summary>
protected internal bool ContinueOnQueryResponse { get; protected set; }
protected List<SystemSubscription> systemSubscriptions = new();
/// <summary>
/// If a message is received on the socket which is not handled by a handler this boolean determines whether this logs an error message
/// </summary>
protected internal bool UnhandledMessageExpected { get; set; }
/// <summary>
/// If true a subscription will accept message before the confirmation of a subscription has been received
/// </summary>
protected bool HandleMessageBeforeConfirmation { get; set; }
/// <summary>
/// The rate limiters
/// </summary>
protected internal IEnumerable<IRateLimiter>? RateLimiters { get; set; }
/// <summary>
/// Periodic task regisrations
/// </summary>
protected List<PeriodicTaskRegistration> PeriodicTaskRegistrations { get; set; } = new List<PeriodicTaskRegistration>();
/// <inheritdoc />
public double IncomingKbps
{
@@ -104,16 +91,16 @@ namespace CryptoExchange.Net
if (!socketConnections.Any())
return 0;
return socketConnections.Sum(s => s.Value.SubscriptionCount);
return socketConnections.Sum(s => s.Value.UserSubscriptionCount);
}
}
/// <inheritdoc />
public new SocketExchangeOptions ClientOptions => (SocketExchangeOptions)base.ClientOptions;
/// <inheritdoc />
public new SocketApiOptions ApiOptions => (SocketApiOptions)base.ApiOptions;
#endregion
/// <summary>
@@ -138,49 +125,50 @@ namespace CryptoExchange.Net
}
/// <summary>
/// Set a delegate to be used for processing data received from socket connections before it is processed by handlers
/// Add a query to periodically send on each connection
/// </summary>
/// <param name="byteHandler">Handler for byte data</param>
/// <param name="stringHandler">Handler for string data</param>
protected void SetDataInterpreter(Func<byte[], string>? byteHandler, Func<string, string>? stringHandler)
/// <param name="identifier"></param>
/// <param name="interval"></param>
/// <param name="queryDelegate"></param>
/// <param name="callback"></param>
protected virtual void RegisterPeriodicQuery(string identifier, TimeSpan interval, Func<SocketConnection, Query> queryDelegate, Action<CallResult>? callback)
{
dataInterpreterBytes = byteHandler;
dataInterpreterString = stringHandler;
PeriodicTaskRegistrations.Add(new PeriodicTaskRegistration
{
Identifier = identifier,
Callback = callback,
Interval = interval,
QueryDelegate = queryDelegate
});
}
/// <summary>
/// Connect to an url and listen for data on the BaseAddress
/// </summary>
/// <typeparam name="T">The type of the expected data</typeparam>
/// <param name="request">The optional request object to send, will be serialized to json</param>
/// <param name="identifier">The identifier to use, necessary if no request object is sent</param>
/// <param name="authenticated">If the subscription is to an authenticated endpoint</param>
/// <param name="dataHandler">The handler of update data</param>
/// <param name="subscription">The subscription</param>
/// <param name="ct">Cancellation token for closing this subscription</param>
/// <returns></returns>
protected virtual Task<CallResult<UpdateSubscription>> SubscribeAsync<T>(object? request, string? identifier, bool authenticated, Action<DataEvent<T>> dataHandler, CancellationToken ct)
protected virtual Task<CallResult<UpdateSubscription>> SubscribeAsync(Subscription subscription, CancellationToken ct)
{
return SubscribeAsync(BaseAddress, request, identifier, authenticated, dataHandler, ct);
return SubscribeAsync(BaseAddress, subscription, ct);
}
/// <summary>
/// Connect to an url and listen for data
/// </summary>
/// <typeparam name="T">The type of the expected data</typeparam>
/// <param name="url">The URL to connect to</param>
/// <param name="request">The optional request object to send, will be serialized to json</param>
/// <param name="identifier">The identifier to use, necessary if no request object is sent</param>
/// <param name="authenticated">If the subscription is to an authenticated endpoint</param>
/// <param name="dataHandler">The handler of update data</param>
/// <param name="subscription">The subscription</param>
/// <param name="ct">Cancellation token for closing this subscription</param>
/// <returns></returns>
protected virtual async Task<CallResult<UpdateSubscription>> SubscribeAsync<T>(string url, object? request, string? identifier, bool authenticated, Action<DataEvent<T>> dataHandler, CancellationToken ct)
protected virtual async Task<CallResult<UpdateSubscription>> SubscribeAsync(string url, Subscription subscription, CancellationToken ct)
{
if (_disposing)
return new CallResult<UpdateSubscription>(new InvalidOperationError("Client disposed, can't subscribe"));
if (subscription.Authenticated && AuthenticationProvider == null)
return new CallResult<UpdateSubscription>(new NoApiCredentialsError());
SocketConnection socketConnection;
SocketSubscription? subscription;
var released = false;
// Wait for a semaphore here, so we only connect 1 socket at a time.
// This is necessary for being able to see if connections can be combined
@@ -198,17 +186,17 @@ namespace CryptoExchange.Net
while (true)
{
// Get a new or existing socket connection
var socketResult = await GetSocketConnection(url, authenticated).ConfigureAwait(false);
var socketResult = await GetSocketConnection(url, subscription.Authenticated).ConfigureAwait(false);
if (!socketResult)
return socketResult.As<UpdateSubscription>(null);
socketConnection = socketResult.Data;
// Add a subscription on the socket connection
subscription = AddSubscription(request, identifier, true, socketConnection, dataHandler, authenticated);
if (subscription == null)
var success = socketConnection.CanAddSubscription();
if (!success)
{
_logger.Log(LogLevel.Trace, $"Socket {socketConnection.SocketId} failed to add subscription, retrying on different connection");
_logger.Log(LogLevel.Trace, $"[Sckt {socketConnection.SocketId}] failed to add subscription, retrying on different connection");
continue;
}
@@ -221,7 +209,7 @@ namespace CryptoExchange.Net
var needsConnecting = !socketConnection.Connected;
var connectResult = await ConnectIfNeededAsync(socketConnection, authenticated).ConfigureAwait(false);
var connectResult = await ConnectIfNeededAsync(socketConnection, subscription.Authenticated).ConfigureAwait(false);
if (!connectResult)
return new CallResult<UpdateSubscription>(connectResult.Error!);
@@ -236,75 +224,60 @@ namespace CryptoExchange.Net
if (socketConnection.PausedActivity)
{
_logger.Log(LogLevel.Warning, $"Socket {socketConnection.SocketId} has been paused, can't subscribe at this moment");
_logger.Log(LogLevel.Warning, $"[Sckt {socketConnection.SocketId}] has been paused, can't subscribe at this moment");
return new CallResult<UpdateSubscription>(new ServerError("Socket is paused"));
}
if (request != null)
var waitEvent = new AsyncResetEvent(false);
var subQuery = subscription.GetSubQuery(socketConnection);
if (subQuery != null)
{
if (HandleMessageBeforeConfirmation)
socketConnection.AddSubscription(subscription);
// Send the request and wait for answer
var subResult = await SubscribeAndWaitAsync(socketConnection, request, subscription).ConfigureAwait(false);
var subResult = await socketConnection.SendAndWaitQueryAsync(subQuery, waitEvent).ConfigureAwait(false);
if (!subResult)
{
_logger.Log(LogLevel.Warning, $"Socket {socketConnection.SocketId} failed to subscribe: {subResult.Error}");
await socketConnection.CloseAsync(subscription).ConfigureAwait(false);
waitEvent?.Set();
_logger.Log(LogLevel.Warning, $"[Sckt {socketConnection.SocketId}] failed to subscribe: {subResult.Error}");
// If this was a timeout we still need to send an unsubscribe to prevent messages coming in later
var unsubscribe = subResult.Error is CancellationRequestedError;
await socketConnection.CloseAsync(subscription, unsubscribe).ConfigureAwait(false);
return new CallResult<UpdateSubscription>(subResult.Error!);
}
}
else
{
// No request to be sent, so just mark the subscription as comfirmed
subscription.Confirmed = true;
subscription.HandleSubQueryResponse(subQuery.Response!);
}
subscription.Confirmed = true;
if (ct != default)
{
subscription.CancellationTokenRegistration = ct.Register(async () =>
{
_logger.Log(LogLevel.Information, $"Socket {socketConnection.SocketId} Cancellation token set, closing subscription");
_logger.Log(LogLevel.Information, $"[Sckt {socketConnection.SocketId}] Cancellation token set, closing subscription {subscription.Id}");
await socketConnection.CloseAsync(subscription).ConfigureAwait(false);
}, false);
}
_logger.Log(LogLevel.Information, $"Socket {socketConnection.SocketId} subscription {subscription.Id} completed successfully");
if (!HandleMessageBeforeConfirmation)
socketConnection.AddSubscription(subscription);
waitEvent?.Set();
_logger.Log(LogLevel.Information, $"[Sckt {socketConnection.SocketId}] subscription {subscription.Id} completed successfully");
return new CallResult<UpdateSubscription>(new UpdateSubscription(socketConnection, subscription));
}
/// <summary>
/// Sends the subscribe request and waits for a response to that request
/// </summary>
/// <param name="socketConnection">The connection to send the request on</param>
/// <param name="request">The request to send, will be serialized to json</param>
/// <param name="subscription">The subscription the request is for</param>
/// <returns></returns>
protected internal virtual async Task<CallResult<bool>> SubscribeAndWaitAsync(SocketConnection socketConnection, object request, SocketSubscription subscription)
{
CallResult<object>? callResult = null;
await socketConnection.SendAndWaitAsync(request, ClientOptions.RequestTimeout, subscription, 1, data => HandleSubscriptionResponse(socketConnection, subscription, request, data, out callResult)).ConfigureAwait(false);
if (callResult?.Success == true)
{
subscription.Confirmed = true;
return new CallResult<bool>(true);
}
if (callResult == null)
return new CallResult<bool>(new ServerError("No response on subscription request received"));
return new CallResult<bool>(callResult.Error!);
}
/// <summary>
/// Send a query on a socket connection to the BaseAddress and wait for the response
/// </summary>
/// <typeparam name="T">Expected result type</typeparam>
/// <param name="request">The request to send, will be serialized to json</param>
/// <param name="authenticated">If the query is to an authenticated endpoint</param>
/// <param name="weight">Weight of the request</param>
/// <param name="query">The query</param>
/// <returns></returns>
protected virtual Task<CallResult<T>> QueryAsync<T>(object request, bool authenticated, int weight = 1)
protected virtual Task<CallResult<T>> QueryAsync<T>(Query<T> query)
{
return QueryAsync<T>(BaseAddress, request, authenticated, weight);
return QueryAsync(BaseAddress, query);
}
/// <summary>
@@ -312,11 +285,9 @@ namespace CryptoExchange.Net
/// </summary>
/// <typeparam name="T">The expected result type</typeparam>
/// <param name="url">The url for the request</param>
/// <param name="request">The request to send</param>
/// <param name="authenticated">Whether the socket should be authenticated</param>
/// <param name="weight">Weight of the request</param>
/// <param name="query">The query</param>
/// <returns></returns>
protected virtual async Task<CallResult<T>> QueryAsync<T>(string url, object request, bool authenticated, int weight = 1)
protected virtual async Task<CallResult<T>> QueryAsync<T>(string url, Query<T> query)
{
if (_disposing)
return new CallResult<T>(new InvalidOperationError("Client disposed, can't query"));
@@ -326,7 +297,7 @@ namespace CryptoExchange.Net
await semaphoreSlim.WaitAsync().ConfigureAwait(false);
try
{
var socketResult = await GetSocketConnection(url, authenticated).ConfigureAwait(false);
var socketResult = await GetSocketConnection(url, query.Authenticated).ConfigureAwait(false);
if (!socketResult)
return socketResult.As<T>(default);
@@ -339,7 +310,7 @@ namespace CryptoExchange.Net
released = true;
}
var connectResult = await ConnectIfNeededAsync(socketConnection, authenticated).ConfigureAwait(false);
var connectResult = await ConnectIfNeededAsync(socketConnection, query.Authenticated).ConfigureAwait(false);
if (!connectResult)
return new CallResult<T>(connectResult.Error!);
}
@@ -351,34 +322,11 @@ namespace CryptoExchange.Net
if (socketConnection.PausedActivity)
{
_logger.Log(LogLevel.Warning, $"Socket {socketConnection.SocketId} has been paused, can't send query at this moment");
_logger.Log(LogLevel.Warning, $"[Sckt {socketConnection.SocketId}] has been paused, can't send query at this moment");
return new CallResult<T>(new ServerError("Socket is paused"));
}
return await QueryAndWaitAsync<T>(socketConnection, request, weight).ConfigureAwait(false);
}
/// <summary>
/// Sends the query request and waits for the result
/// </summary>
/// <typeparam name="T">The expected result type</typeparam>
/// <param name="socket">The connection to send and wait on</param>
/// <param name="request">The request to send</param>
/// <param name="weight">The weight of the query</param>
/// <returns></returns>
protected virtual async Task<CallResult<T>> QueryAndWaitAsync<T>(SocketConnection socket, object request, int weight)
{
var dataResult = new CallResult<T>(new ServerError("No response on query received"));
await socket.SendAndWaitAsync(request, ClientOptions.RequestTimeout, null, weight, data =>
{
if (!HandleQueryResponse<T>(socket, request, data, out var callResult))
return false;
dataResult = callResult;
return true;
}).ConfigureAwait(false);
return dataResult;
return await socketConnection.SendAndWaitQueryAsync(query).ConfigureAwait(false);
}
/// <summary>
@@ -402,149 +350,56 @@ namespace CryptoExchange.Net
if (!authenticated || socket.Authenticated)
return new CallResult<bool>(true);
_logger.Log(LogLevel.Debug, $"Socket {socket.SocketId} Attempting to authenticate");
var result = await AuthenticateSocketAsync(socket).ConfigureAwait(false);
if (!result)
{
_logger.Log(LogLevel.Warning, $"Socket {socket.SocketId} authentication failed");
if (socket.Connected)
await socket.CloseAsync().ConfigureAwait(false);
return await AuthenticateSocketAsync(socket).ConfigureAwait(false);
}
result.Error!.Message = "Authentication failed: " + result.Error.Message;
return new CallResult<bool>(result.Error);
/// <summary>
/// Authenticate a socket connection
/// </summary>
/// <param name="socket">Socket to authenticate</param>
/// <returns></returns>
public virtual async Task<CallResult<bool>> AuthenticateSocketAsync(SocketConnection socket)
{
if (AuthenticationProvider == null)
return new CallResult<bool>(new NoApiCredentialsError());
_logger.Log(LogLevel.Debug, $"[Sckt {socket.SocketId}] Attempting to authenticate");
var authRequest = GetAuthenticationRequest();
if (authRequest != null)
{
var result = await socket.SendAndWaitQueryAsync(authRequest).ConfigureAwait(false);
if (!result)
{
_logger.Log(LogLevel.Warning, $"[Sckt {socket.SocketId}] authentication failed");
if (socket.Connected)
await socket.CloseAsync().ConfigureAwait(false);
result.Error!.Message = "Authentication failed: " + result.Error.Message;
return new CallResult<bool>(result.Error)!;
}
}
_logger.Log(LogLevel.Debug, $"Socket {socket.SocketId} authenticated");
_logger.Log(LogLevel.Debug, $"[Sckt {socket.SocketId}] authenticated");
socket.Authenticated = true;
return new CallResult<bool>(true);
}
/// <summary>
/// The socketConnection received data (the data JToken parameter). The implementation of this method should check if the received data is a response to the query that was send (the request parameter).
/// For example; A query is sent in a request message with an Id parameter with value 10. The socket receives data and calls this method to see if the data it received is an
/// anwser to any query that was done. The implementation of this method should check if the response.Id == request.Id to see if they match (assuming the api has some sort of Id tracking on messages,
/// if not some other method has be implemented to match the messages).
/// If the messages match, the callResult out parameter should be set with the deserialized data in the from of (T) and return true.
/// Should return the request which can be used to authenticate a socket connection
/// </summary>
/// <typeparam name="T">The type of response that is expected on the query</typeparam>
/// <param name="socketConnection">The socket connection</param>
/// <param name="request">The request that a response is awaited for</param>
/// <param name="data">The message received from the server</param>
/// <param name="callResult">The interpretation (null if message wasn't a response to the request)</param>
/// <returns>True if the message was a response to the query</returns>
protected internal abstract bool HandleQueryResponse<T>(SocketConnection socketConnection, object request, JToken data, [NotNullWhen(true)] out CallResult<T>? callResult);
/// <summary>
/// The socketConnection received data (the data JToken parameter). The implementation of this method should check if the received data is a response to the subscription request that was send (the request parameter).
/// For example; A subscribe request message is send with an Id parameter with value 10. The socket receives data and calls this method to see if the data it received is an
/// anwser to any subscription request that was done. The implementation of this method should check if the response.Id == request.Id to see if they match (assuming the api has some sort of Id tracking on messages,
/// if not some other method has be implemented to match the messages).
/// If the messages match, the callResult out parameter should be set with the deserialized data in the from of (T) and return true.
/// </summary>
/// <param name="socketConnection">The socket connection</param>
/// <param name="subscription">A subscription that waiting for a subscription response</param>
/// <param name="request">The request that the subscription sent</param>
/// <param name="data">The message received from the server</param>
/// <param name="callResult">The interpretation (null if message wasn't a response to the request)</param>
/// <returns>True if the message was a response to the subscription request</returns>
protected internal abstract bool HandleSubscriptionResponse(SocketConnection socketConnection, SocketSubscription subscription, object request, JToken data, out CallResult<object>? callResult);
/// <summary>
/// Needs to check if a received message matches a handler by request. After subscribing data message will come in. These data messages need to be matched to a specific connection
/// to pass the correct data to the correct handler. The implementation of this method should check if the message received matches the subscribe request that was sent.
/// </summary>
/// <param name="socketConnection">The socket connection the message was recieved on</param>
/// <param name="message">The received data</param>
/// <param name="request">The subscription request</param>
/// <returns>True if the message is for the subscription which sent the request</returns>
protected internal abstract bool MessageMatchesHandler(SocketConnection socketConnection, JToken message, object request);
/// <summary>
/// Needs to check if a received message matches a handler by identifier. Generally used by GenericHandlers. For example; a generic handler is registered which handles ping messages
/// from the server. This method should check if the message received is a ping message and the identifer is the identifier of the GenericHandler
/// </summary>
/// <param name="socketConnection">The socket connection the message was recieved on</param>
/// <param name="message">The received data</param>
/// <param name="identifier">The string identifier of the handler</param>
/// <returns>True if the message is for the handler which has the identifier</returns>
protected internal abstract bool MessageMatchesHandler(SocketConnection socketConnection, JToken message, string identifier);
/// <summary>
/// Needs to authenticate the socket so authenticated queries/subscriptions can be made on this socket connection
/// </summary>
/// <param name="socketConnection">The socket connection that should be authenticated</param>
/// <returns></returns>
protected internal abstract Task<CallResult<bool>> AuthenticateSocketAsync(SocketConnection socketConnection);
/// <summary>
/// Needs to unsubscribe a subscription, typically by sending an unsubscribe request. If multiple subscriptions per socket is not allowed this can just return since the socket will be closed anyway
/// </summary>
/// <param name="connection">The connection on which to unsubscribe</param>
/// <param name="subscriptionToUnsub">The subscription to unsubscribe</param>
/// <returns></returns>
protected internal abstract Task<bool> UnsubscribeAsync(SocketConnection connection, SocketSubscription subscriptionToUnsub);
protected internal virtual Query? GetAuthenticationRequest() => throw new NotImplementedException();
/// <summary>
/// Optional handler to interpolate data before sending it to the handlers
/// Adds a system subscription. Used for example to reply to ping requests
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
protected internal virtual JToken ProcessTokenData(JToken message)
/// <param name="systemSubscription">The subscription</param>
protected void AddSystemSubscription(SystemSubscription systemSubscription)
{
return message;
}
/// <summary>
/// Add a subscription to a connection
/// </summary>
/// <typeparam name="T">The type of data the subscription expects</typeparam>
/// <param name="request">The request of the subscription</param>
/// <param name="identifier">The identifier of the subscription (can be null if request param is used)</param>
/// <param name="userSubscription">Whether or not this is a user subscription (counts towards the max amount of handlers on a socket)</param>
/// <param name="connection">The socket connection the handler is on</param>
/// <param name="dataHandler">The handler of the data received</param>
/// <param name="authenticated">Whether the subscription needs authentication</param>
/// <returns></returns>
protected virtual SocketSubscription? AddSubscription<T>(object? request, string? identifier, bool userSubscription, SocketConnection connection, Action<DataEvent<T>> dataHandler, bool authenticated)
{
void InternalHandler(MessageEvent messageEvent)
{
if (typeof(T) == typeof(string))
{
var stringData = (T)Convert.ChangeType(messageEvent.JsonData.ToString(), typeof(T));
dataHandler(new DataEvent<T>(stringData, null, OutputOriginalData ? messageEvent.OriginalData : null, messageEvent.ReceivedTimestamp, null));
return;
}
var desResult = Deserialize<T>(messageEvent.JsonData);
if (!desResult)
{
_logger.Log(LogLevel.Warning, $"Socket {connection.SocketId} Failed to deserialize data into type {typeof(T)}: {desResult.Error}");
return;
}
dataHandler(new DataEvent<T>(desResult.Data, null, OutputOriginalData ? messageEvent.OriginalData : null, messageEvent.ReceivedTimestamp, null));
}
var subscription = request == null
? SocketSubscription.CreateForIdentifier(ExchangeHelpers.NextId(), identifier!, userSubscription, authenticated, InternalHandler)
: SocketSubscription.CreateForRequest(ExchangeHelpers.NextId(), request, userSubscription, authenticated, InternalHandler);
if (!connection.AddSubscription(subscription))
return null;
return subscription;
}
/// <summary>
/// Adds a generic message handler. Used for example to reply to ping requests
/// </summary>
/// <param name="identifier">The name of the request handler. Needs to be unique</param>
/// <param name="action">The action to execute when receiving a message for this handler (checked by <see cref="MessageMatchesHandler(SocketConnection, Newtonsoft.Json.Linq.JToken,string)"/>)</param>
protected void AddGenericHandler(string identifier, Action<MessageEvent> action)
{
genericHandlers.Add(identifier, action);
var subscription = SocketSubscription.CreateForIdentifier(ExchangeHelpers.NextId(), identifier, false, false, action);
systemSubscriptions.Add(systemSubscription);
foreach (var connection in socketConnections.Values)
connection.AddSubscription(subscription);
connection.AddSubscription(systemSubscription);
}
/// <summary>
@@ -569,13 +424,13 @@ namespace CryptoExchange.Net
}
/// <summary>
/// Update the original request to send when the connection is restored after disconnecting. Can be used to update an authentication token for example.
/// Update the subscription when the connection is restored after disconnecting. Can be used to update an authentication token for example.
/// </summary>
/// <param name="request">The original request</param>
/// <param name="subscription">The subscription</param>
/// <returns></returns>
protected internal virtual Task<CallResult<object>> RevitalizeRequestAsync(object request)
protected internal virtual Task<CallResult> RevitalizeRequestAsync(Subscription subscription)
{
return Task.FromResult(new CallResult<object>(request));
return Task.FromResult(new CallResult(null));
}
/// <summary>
@@ -589,11 +444,11 @@ namespace CryptoExchange.Net
var socketResult = socketConnections.Where(s => (s.Value.Status == SocketConnection.SocketStatus.None || s.Value.Status == SocketConnection.SocketStatus.Connected)
&& s.Value.Tag.TrimEnd('/') == address.TrimEnd('/')
&& (s.Value.ApiClient.GetType() == GetType())
&& (s.Value.Authenticated == authenticated || !authenticated) && s.Value.Connected).OrderBy(s => s.Value.SubscriptionCount).FirstOrDefault();
&& (s.Value.Authenticated == authenticated || !authenticated) && s.Value.Connected).OrderBy(s => s.Value.UserSubscriptionCount).FirstOrDefault();
var result = socketResult.Equals(default(KeyValuePair<int, SocketConnection>)) ? null : socketResult.Value;
if (result != null)
{
if (result.SubscriptionCount < ClientOptions.SocketSubscriptionsCombineTarget || (socketConnections.Count >= (ApiOptions.MaxSocketConnections ?? ClientOptions.MaxSocketConnections) && socketConnections.All(s => s.Value.SubscriptionCount >= ClientOptions.SocketSubscriptionsCombineTarget)))
if (result.UserSubscriptionCount < ClientOptions.SocketSubscriptionsCombineTarget || (socketConnections.Count >= (ApiOptions.MaxSocketConnections ?? ClientOptions.MaxSocketConnections) && socketConnections.All(s => s.Value.UserSubscriptionCount >= ClientOptions.SocketSubscriptionsCombineTarget)))
{
// Use existing socket if it has less than target connections OR it has the least connections and we can't make new
return new CallResult<SocketConnection>(result);
@@ -614,11 +469,12 @@ namespace CryptoExchange.Net
var socket = CreateSocket(connectionAddress.Data!);
var socketConnection = new SocketConnection(_logger, this, socket, address);
socketConnection.UnhandledMessage += HandleUnhandledMessage;
foreach (var kvp in genericHandlers)
{
var handler = SocketSubscription.CreateForIdentifier(ExchangeHelpers.NextId(), kvp.Key, false, false, kvp.Value);
socketConnection.AddSubscription(handler);
}
foreach (var ptg in PeriodicTaskRegistrations)
socketConnection.QueryPeriodic(ptg.Identifier, ptg.Interval, ptg.QueryDelegate, ptg.Callback);
foreach (var systemSubscription in systemSubscriptions)
socketConnection.AddSubscription(systemSubscription);
return new CallResult<SocketConnection>(socketConnection);
}
@@ -626,8 +482,8 @@ namespace CryptoExchange.Net
/// <summary>
/// Process an unhandled message
/// </summary>
/// <param name="token">The token that wasn't processed</param>
protected virtual void HandleUnhandledMessage(JToken token)
/// <param name="message">The message that wasn't processed</param>
protected virtual void HandleUnhandledMessage(IMessageAccessor message)
{
}
@@ -656,8 +512,6 @@ namespace CryptoExchange.Net
protected virtual WebSocketParameters GetWebSocketParameters(string address)
=> new(new Uri(address), ClientOptions.AutoReconnect)
{
DataInterpreterBytes = dataInterpreterBytes,
DataInterpreterString = dataInterpreterString,
KeepAliveInterval = KeepAliveInterval,
ReconnectInterval = ClientOptions.ReconnectInterval,
RateLimiters = RateLimiters,
@@ -673,57 +527,10 @@ namespace CryptoExchange.Net
protected virtual IWebsocket CreateSocket(string address)
{
var socket = SocketFactory.CreateWebsocket(_logger, GetWebSocketParameters(address));
_logger.Log(LogLevel.Debug, $"Socket {socket.Id} new socket created for " + address);
_logger.Log(LogLevel.Debug, $"[Sckt {socket.Id}] created for " + address);
return socket;
}
/// <summary>
/// Periodically sends data over a socket connection
/// </summary>
/// <param name="identifier">Identifier for the periodic send</param>
/// <param name="interval">How often</param>
/// <param name="objGetter">Method returning the object to send</param>
protected virtual void SendPeriodic(string identifier, TimeSpan interval, Func<SocketConnection, object> objGetter)
{
if (objGetter == null)
throw new ArgumentNullException(nameof(objGetter));
periodicEvent = new AsyncResetEvent();
periodicTask = Task.Run(async () =>
{
while (!_disposing)
{
await periodicEvent.WaitAsync(interval).ConfigureAwait(false);
if (_disposing)
break;
foreach (var socketConnection in socketConnections.Values)
{
if (_disposing)
break;
if (!socketConnection.Connected)
continue;
var obj = objGetter(socketConnection);
if (obj == null)
continue;
_logger.Log(LogLevel.Trace, $"Socket {socketConnection.SocketId} sending periodic {identifier}");
try
{
socketConnection.Send(ExchangeHelpers.NextId(), obj, 1);
}
catch (Exception ex)
{
_logger.Log(LogLevel.Warning, $"Socket {socketConnection.SocketId} Periodic send {identifier} failed: " + ex.ToLogString());
}
}
}
});
}
/// <summary>
/// Unsubscribe an update subscription
/// </summary>
@@ -731,7 +538,7 @@ namespace CryptoExchange.Net
/// <returns></returns>
public virtual async Task<bool> UnsubscribeAsync(int subscriptionId)
{
SocketSubscription? subscription = null;
Subscription? subscription = null;
SocketConnection? connection = null;
foreach (var socket in socketConnections.Values.ToList())
{
@@ -746,7 +553,7 @@ namespace CryptoExchange.Net
if (subscription == null || connection == null)
return false;
_logger.Log(LogLevel.Information, $"Socket {connection.SocketId} Unsubscribing subscription " + subscriptionId);
_logger.Log(LogLevel.Information, $"[Sckt {connection.SocketId}] unsubscribing subscription " + subscriptionId);
await connection.CloseAsync(subscription).ConfigureAwait(false);
return true;
}
@@ -761,7 +568,7 @@ namespace CryptoExchange.Net
if (subscription == null)
throw new ArgumentNullException(nameof(subscription));
_logger.Log(LogLevel.Information, $"Socket {subscription.SocketId} Unsubscribing subscription " + subscription.Id);
_logger.Log(LogLevel.Information, $"[Sckt {subscription.SocketId}] Unsubscribing subscription " + subscription.Id);
await subscription.CloseAsync().ConfigureAwait(false);
}
@@ -771,11 +578,11 @@ namespace CryptoExchange.Net
/// <returns></returns>
public virtual async Task UnsubscribeAllAsync()
{
var sum = socketConnections.Sum(s => s.Value.SubscriptionCount);
var sum = socketConnections.Sum(s => s.Value.UserSubscriptionCount);
if (sum == 0)
return;
_logger.Log(LogLevel.Information, $"Unsubscribing all {socketConnections.Sum(s => s.Value.SubscriptionCount)} subscriptions");
_logger.Log(LogLevel.Information, $"Unsubscribing all {socketConnections.Sum(s => s.Value.UserSubscriptionCount)} subscriptions");
var tasks = new List<Task>();
{
var socketList = socketConnections.Values;
@@ -809,12 +616,26 @@ namespace CryptoExchange.Net
public string GetSubscriptionsState()
{
var sb = new StringBuilder();
sb.AppendLine($"{socketConnections.Count} connections, {CurrentSubscriptions} subscriptions, kbps: {IncomingKbps}");
sb.AppendLine($"{GetType().Name}");
sb.AppendLine($" Connections: {socketConnections.Count}");
sb.AppendLine($" Subscriptions: {CurrentSubscriptions}");
sb.AppendLine($" Download speed: {IncomingKbps} kbps");
foreach (var connection in socketConnections)
{
sb.AppendLine($" Connection {connection.Key}: {connection.Value.SubscriptionCount} subscriptions, status: {connection.Value.Status}, authenticated: {connection.Value.Authenticated}, kbps: {connection.Value.IncomingKbps}");
sb.AppendLine($" Id: {connection.Key}");
sb.AppendLine($" Address: {connection.Value.ConnectionUri}");
sb.AppendLine($" Subscriptions: {connection.Value.UserSubscriptionCount}");
sb.AppendLine($" Status: {connection.Value.Status}");
sb.AppendLine($" Authenticated: {connection.Value.Authenticated}");
sb.AppendLine($" Download speed: {connection.Value.IncomingKbps} kbps");
sb.AppendLine($" Subscriptions:");
foreach (var subscription in connection.Value.Subscriptions)
sb.AppendLine($" Subscription {subscription.Id}, authenticated: {subscription.Authenticated}, confirmed: {subscription.Confirmed}");
{
sb.AppendLine($" Id: {subscription.Id}");
sb.AppendLine($" Confirmed: {subscription.Confirmed}");
sb.AppendLine($" Invocations: {subscription.TotalInvocations}");
sb.AppendLine($" Identifiers: [{string.Join(", ", subscription.ListenerIdentifiers)}]");
}
}
return sb.ToString();
}
@@ -825,9 +646,7 @@ namespace CryptoExchange.Net
public override void Dispose()
{
_disposing = true;
periodicEvent?.Set();
periodicEvent?.Dispose();
if (socketConnections.Sum(s => s.Value.SubscriptionCount) > 0)
if (socketConnections.Sum(s => s.Value.UserSubscriptionCount) > 0)
{
_logger.Log(LogLevel.Debug, "Disposing socket client, closing all subscriptions");
_ = UnsubscribeAllAsync();
@@ -835,5 +654,20 @@ namespace CryptoExchange.Net
semaphoreSlim?.Dispose();
base.Dispose();
}
/// <summary>
/// Get the listener identifier for the message
/// </summary>
/// <param name="messageAccessor"></param>
/// <returns></returns>
public abstract string? GetListenerIdentifier(IMessageAccessor messageAccessor);
/// <summary>
/// Preprocess a stream message
/// </summary>
/// <param name="type"></param>
/// <param name="stream"></param>
/// <returns></returns>
public virtual Stream PreprocessStreamMessage(WebSocketMessageType type, Stream stream) => stream;
}
}