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

wip

Browse Source
This commit is contained in:
JKorf 2023-12-28 10:39:35 +01:00
parent 58098edaa6
commit eee19b28a5
13 changed files with 227 additions and 311 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
CryptoExchange.Net/Clients/SocketApiClient.cs
View File

@ -111,7 +111,7 @@ namespace CryptoExchange.Net
public new SocketApiOptions ApiOptions => (SocketApiOptions)base.ApiOptions; public new SocketApiOptions ApiOptions => (SocketApiOptions)base.ApiOptions;
/// <inheritdoc /> /// <inheritdoc />
public abstract SocketConverter StreamConverter { get; } public abstract MessageInterpreterPipeline Pipeline { get; }
#endregion #endregion
/// <summary> /// <summary>
@ -686,7 +686,7 @@ namespace CryptoExchange.Net
sb.AppendLine($" Id: {subscription.Id}"); sb.AppendLine($" Id: {subscription.Id}");
sb.AppendLine($" Confirmed: {subscription.Confirmed}"); sb.AppendLine($" Confirmed: {subscription.Confirmed}");
sb.AppendLine($" Invocations: {subscription.TotalInvocations}"); sb.AppendLine($" Invocations: {subscription.TotalInvocations}");
sb.AppendLine($" Identifiers: [{string.Join(", ", subscription.Identifiers)}]"); sb.AppendLine($" Identifiers: [{string.Join(", ", subscription.StreamIdentifiers)}]");
} }
} }
return sb.ToString(); return sb.ToString();

36
CryptoExchange.Net/Converters/JTokenAccessor.cs
View File

@ -1,8 +1,12 @@
using CryptoExchange.Net.Interfaces; using CryptoExchange.Net.Interfaces;
using CryptoExchange.Net.Objects.Sockets;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq; using Newtonsoft.Json.Linq;
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.IO;
using System.Linq; using System.Linq;
using System.Runtime.InteropServices.ComTypes;
using System.Text; using System.Text;
namespace CryptoExchange.Net.Converters namespace CryptoExchange.Net.Converters
@ -10,11 +14,39 @@ namespace CryptoExchange.Net.Converters
internal class JTokenAccessor : IMessageAccessor internal class JTokenAccessor : IMessageAccessor
{ {
private readonly JToken _token; private readonly JToken _token;
private readonly Stream _stream;
private readonly StreamReader _reader;
private Dictionary<string, JToken?> _cache = new Dictionary<string, JToken?>(); private Dictionary<string, JToken?> _cache = new Dictionary<string, JToken?>();
private static JsonSerializer _serializer = JsonSerializer.Create(SerializerOptions.WithConverters);
public JTokenAccessor(JToken token) public JTokenAccessor(Stream stream)
{ {
_token = token; _stream = stream;
_reader = new StreamReader(stream, Encoding.UTF8, false, (int)stream.Length, true);
using var jsonTextReader = new JsonTextReader(_reader);
JToken token;
try
{
_token = JToken.Load(jsonTextReader);
}
catch (Exception ex)
{
// Not a json message
throw;
}
}
public BaseParsedMessage Instantiate(Type type)
{
var resultMessageType = typeof(ParsedMessage<>).MakeGenericType(type);
var instance = (BaseParsedMessage)Activator.CreateInstance(resultMessageType, type == null ? null : _token.ToObject(type, _serializer));
return instance;
}
public string GetOriginalString()
{
_stream.Position = 0;
return _reader.ReadToEnd();
} }
public int? GetArrayIntValue(string? key, int index) public int? GetArrayIntValue(string? key, int index)

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

@ -1,195 +1,58 @@
using CryptoExchange.Net.Interfaces; //using CryptoExchange.Net.Interfaces;
using CryptoExchange.Net.Objects; //using CryptoExchange.Net.Objects;
using CryptoExchange.Net.Objects.Sockets; //using CryptoExchange.Net.Objects.Sockets;
using CryptoExchange.Net.Sockets; //using CryptoExchange.Net.Sockets;
using Newtonsoft.Json; //using Newtonsoft.Json;
using Newtonsoft.Json.Linq; //using Newtonsoft.Json.Linq;
using System; //using System;
using System.Collections.Generic; //using System.Collections.Generic;
using System.IO; //using System.IO;
using System.Linq; //using System.Linq;
using System.Net.WebSockets; //using System.Net.WebSockets;
using System.Text; //using System.Text;
namespace CryptoExchange.Net.Converters //namespace CryptoExchange.Net.Converters
{
/// <summary>
/// Socket message converter
/// </summary>
public abstract class SocketConverter
{
private static JsonSerializer _serializer = JsonSerializer.Create(SerializerOptions.WithConverters);
public abstract MessageInterpreterPipeline InterpreterPipeline { get; }
/// <inheritdoc />
public BaseParsedMessage? ReadJson(WebSocketMessageType websocketMessageType, Stream stream, SocketListenerManager listenerManager, bool outputOriginalData)
{
// Start reading the data
// Once we reach the properties that identify the message we save those in a dict
// Once all id properties have been read callback to see what the deserialization type should be
// Deserialize to the correct type
if (InterpreterPipeline.PreProcessCallback != null)
stream = InterpreterPipeline.PreProcessCallback(websocketMessageType, stream);
using var sr = new StreamReader(stream, Encoding.UTF8, false, (int)stream.Length, true);
foreach (var callback in InterpreterPipeline.PreInspectCallbacks)
{
var result = callback.Callback(stream);
if (result.Matched)
{
var data = sr.ReadToEnd();
var messageType = typeof(ParsedMessage<>).MakeGenericType(typeof(string));
var preInstance = (BaseParsedMessage)Activator.CreateInstance(messageType, data);
if (outputOriginalData)
{
stream.Position = 0;
preInstance.OriginalData = data;
}
preInstance.StreamIdentifier = result.StreamIdentifier;
preInstance.TypeIdentifier = result.TypeIdentifier;
preInstance.Parsed = true;
return preInstance;
}
}
using var jsonTextReader = new JsonTextReader(sr);
JToken token;
try
{
token = JToken.Load(jsonTextReader);
}
catch(Exception ex)
{
// Not a json message
return null;
}
var accessor = new JTokenAccessor(token);
if (InterpreterPipeline.GetIdentity != null)
{
var (streamIdentity, typeIdentity) = InterpreterPipeline.GetIdentity(accessor);
if (streamIdentity != null)
{
var result = listenerManager.IdToType(streamIdentity, typeIdentity);
if (result != null)
{
var idInstance = InterpreterPipeline.ObjectInitializer(token, result!);
if (outputOriginalData)
{
stream.Position = 0;
idInstance.OriginalData = sr.ReadToEnd();
}
idInstance.StreamIdentifier = streamIdentity;
idInstance.TypeIdentifier = typeIdentity;
idInstance.Parsed = true;
return idInstance;
}
else
{
}
}
else
{
// Message not identified
// TODO return
}
}
PostInspectResult? inspectResult = null;
object? usedParser = null;
//if (token.Type == JTokenType.Object)
//{ //{
// foreach (var callback in InterpreterPipeline.PostInspectCallbacks.OfType<PostInspectCallback>()) // /// <summary>
// /// Socket message converter
// /// </summary>
// public abstract class SocketConverter
// { // {
// bool allFieldsPresent = true;
// foreach (var field in callback.TypeFields) // public abstract MessageInterpreterPipeline InterpreterPipeline { get; }
// /// <inheritdoc />
// public BaseParsedMessage? ReadJson(WebSocketMessageType websocketMessageType, Stream stream, SocketListenerManager listenerManager, bool outputOriginalData)
// { // {
// var value = accessor.GetStringValue(field.Key); // // Start reading the data
// if (value == null) // // Once we reach the properties that identify the message we save those in a dict
// { // // Once all id properties have been read callback to see what the deserialization type should be
// if (field.Required) // // Deserialize to the correct type
// {
// allFieldsPresent = false; // if (InterpreterPipeline.PreProcessCallback != null)
// break; // stream = InterpreterPipeline.PreProcessCallback(websocketMessageType, stream);
// var accessor = new JTokenAccessor(stream);
// if (accessor == null)
// return null;
// var streamIdentity = InterpreterPipeline.GetStreamIdentifier(accessor);
// if (streamIdentity == null)
// return null;
// var typeIdentity = InterpreterPipeline.GetTypeIdentifier(accessor);
// var typeResult = listenerManager.IdToType(streamIdentity, typeIdentity);
// if (typeResult == null)
// return null;
// var idInstance = accessor.Instantiate(typeResult);
// if (outputOriginalData)
// idInstance.OriginalData = idInstance.OriginalData;
// idInstance.StreamIdentifier = streamIdentity;
// idInstance.TypeIdentifier = typeIdentity;
// idInstance.Parsed = true;
// return idInstance;
// } // }
// } // }
//} //}
// if (allFieldsPresent)
// {
// inspectResult = callback.Callback(accessor, processors);
// usedParser = callback;
// if (inspectResult.Type != null)
// break;
// }
// }
//}
//else
//{
// foreach (var callback in InterpreterPipeline.PostInspectCallbacks.OfType<PostInspectArrayCallback>())
// {
// var typeIdArrayDict = new Dictionary<int, string>();
// bool allFieldsPresent = true;
// var maxIndex = callback.TypeFields.Max();
// if (((JArray)token).Count <= maxIndex)
// continue;
// foreach (var field in callback.TypeFields)
// {
// var value = token[field];
// if (value == null)
// {
// allFieldsPresent = false;
// break;
// }
// typeIdArrayDict[field] = value.ToString();
// }
// if (allFieldsPresent)
// {
// inspectResult = callback.Callback(typeIdArrayDict, processors);
// usedParser = callback;
// break;
// }
// }
//}
if (usedParser == null)
{
//throw new Exception("No parser found for message");
return null;
}
BaseParsedMessage instance;
if (inspectResult.Type != null)
instance = InterpreterPipeline.ObjectInitializer(token, inspectResult.Type);
else
instance = new ParsedMessage<object>(null);
if (outputOriginalData)
{
stream.Position = 0;
instance.OriginalData = sr.ReadToEnd();
}
instance.StreamIdentifier = inspectResult.StreamIdentifier;
instance.TypeIdentifier = inspectResult.TypeIdentifier;
instance.Parsed = inspectResult.Type != null;
return instance;
}
public static BaseParsedMessage InstantiateMessageObject(JToken token, Type type)
{
var resultMessageType = typeof(ParsedMessage<>).MakeGenericType(type);
var instance = (BaseParsedMessage)Activator.CreateInstance(resultMessageType, type == null ? null : token.ToObject(type, _serializer));
return instance;
}
}
}

5
CryptoExchange.Net/Interfaces/IMessageAccessor.cs
View File

@ -1,4 +1,5 @@
using System; using CryptoExchange.Net.Objects.Sockets;
using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Text; using System.Text;
@ -13,5 +14,7 @@ namespace CryptoExchange.Net.Interfaces
public int? GetCount(string key); public int? GetCount(string key);
public int? GetArrayIntValue(string? key, int index); public int? GetArrayIntValue(string? key, int index);
public string? GetArrayStringValue(string? key, int index); public string? GetArrayStringValue(string? key, int index);
public BaseParsedMessage Instantiate(Type type);
} }
} }

4
CryptoExchange.Net/Interfaces/IMessageProcessor.cs
View File

@ -11,8 +11,8 @@ namespace CryptoExchange.Net.Interfaces
public interface IMessageProcessor public interface IMessageProcessor
{ {
public int Id { get; } public int Id { get; }
public List<string> Identifiers { get; } public List<string> StreamIdentifiers { get; }
Task<CallResult> HandleMessageAsync(SocketConnection connection, DataEvent<BaseParsedMessage> message); Task<CallResult> HandleMessageAsync(SocketConnection connection, DataEvent<BaseParsedMessage> message);
public Func<string, Type> ExpectedTypeDelegate { get; } Dictionary<string, Type> TypeMapping { get; }
} }
} }

12
CryptoExchange.Net/Interfaces/IWebsocket.cs
View File

@ -17,7 +17,7 @@ namespace CryptoExchange.Net.Interfaces
/// <summary> /// <summary>
/// Websocket closed event /// Websocket closed event
/// </summary> /// </summary>
event Action OnClose; event Func<Task> OnClose;
/// <summary> /// <summary>
/// Websocket message received event /// Websocket message received event
/// </summary> /// </summary>
@ -25,23 +25,23 @@ namespace CryptoExchange.Net.Interfaces
/// <summary> /// <summary>
/// Websocket sent event, RequestId as parameter /// Websocket sent event, RequestId as parameter
/// </summary> /// </summary>
event Action<int> OnRequestSent; event Func<int, Task> OnRequestSent;
/// <summary> /// <summary>
/// Websocket error event /// Websocket error event
/// </summary> /// </summary>
event Action<Exception> OnError; event Func<Exception, Task> OnError;
/// <summary> /// <summary>
/// Websocket opened event /// Websocket opened event
/// </summary> /// </summary>
event Action OnOpen; event Func<Task> OnOpen;
/// <summary> /// <summary>
/// Websocket has lost connection to the server and is attempting to reconnect /// Websocket has lost connection to the server and is attempting to reconnect
/// </summary> /// </summary>
event Action OnReconnecting; event Func<Task> OnReconnecting;
/// <summary> /// <summary>
/// Websocket has reconnected to the server /// Websocket has reconnected to the server
/// </summary> /// </summary>
event Action OnReconnected; event Func<Task> OnReconnected;
/// <summary> /// <summary>
/// Get reconntion url /// Get reconntion url
/// </summary> /// </summary>

49
CryptoExchange.Net/Objects/Sockets/StreamMessageParseCallback.cs
View File

@ -13,52 +13,7 @@ namespace CryptoExchange.Net.Objects.Sockets
public class MessageInterpreterPipeline public class MessageInterpreterPipeline
{ {
public Func<WebSocketMessageType, Stream, Stream>? PreProcessCallback { get; set; } public Func<WebSocketMessageType, Stream, Stream>? PreProcessCallback { get; set; }
public List<PreInspectCallback> PreInspectCallbacks { get; set; } = new List<PreInspectCallback>(); public Func<IMessageAccessor, string?> GetStreamIdentifier { get; set; }
public Func<IMessageAccessor, (string?, string?)> GetIdentity { get; set; } public Func<IMessageAccessor, string?> GetTypeIdentifier { get; set; }
public List<object> PostInspectCallbacks { get; set; } = new List<object>();
public Func<JToken, Type, BaseParsedMessage> ObjectInitializer { get; set; } = SocketConverter.InstantiateMessageObject;
}
public class PreInspectCallback
{
public Func<Stream, PreInspectResult> Callback { get; set; }
}
public class PostInspectCallback
{
public List<TypeField> TypeFields { get; set; } = new List<TypeField>();
public Func<IMessageAccessor, Dictionary<string, Type>, PostInspectResult> Callback { get; set; }
}
public class TypeField
{
public string Key { get; set; }
public bool Required { get; set; }
public TypeField(string key, bool required = true)
{
Key = key;
Required = required;
}
}
public class PostInspectArrayCallback
{
public List<int> TypeFields { get; set; } = new List<int>();
public Func<IMessageAccessor, SocketListenerManager, PostInspectResult> Callback { get; set; }
}
public class PreInspectResult
{
public bool Matched { get; set; }
public string StreamIdentifier { get; set; }
public string TypeIdentifier { get; set; }
}
public class PostInspectResult
{
public Type? Type { get; set; }
public string StreamIdentifier { get; set; }
public string TypeIdentifier { get; set; }
} }
} }

34
CryptoExchange.Net/Sockets/CryptoExchangeWebSocketClient.cs
View File

@ -98,25 +98,25 @@ namespace CryptoExchange.Net.Sockets
} }
/// <inheritdoc /> /// <inheritdoc />
public event Action? OnClose; public event Func<Task>? OnClose;
/// <inheritdoc /> /// <inheritdoc />
public event Func<WebSocketMessageType, Stream, Task>? OnStreamMessage; public event Func<WebSocketMessageType, Stream, Task>? OnStreamMessage;
/// <inheritdoc /> /// <inheritdoc />
public event Action<int>? OnRequestSent; public event Func<int, Task>? OnRequestSent;
/// <inheritdoc /> /// <inheritdoc />
public event Action<Exception>? OnError; public event Func<Exception, Task>? OnError;
/// <inheritdoc /> /// <inheritdoc />
public event Action? OnOpen; public event Func<Task>? OnOpen;
/// <inheritdoc /> /// <inheritdoc />
public event Action? OnReconnecting; public event Func<Task>? OnReconnecting;
/// <inheritdoc /> /// <inheritdoc />
public event Action? OnReconnected; public event Func<Task>? OnReconnected;
/// <inheritdoc /> /// <inheritdoc />
public Func<Task<Uri?>>? GetReconnectionUrl { get; set; } public Func<Task<Uri?>>? GetReconnectionUrl { get; set; }
@ -147,7 +147,7 @@ namespace CryptoExchange.Net.Sockets
if (!await ConnectInternalAsync().ConfigureAwait(false)) if (!await ConnectInternalAsync().ConfigureAwait(false))
return false; return false;
OnOpen?.Invoke(); await (OnOpen?.Invoke() ?? Task.CompletedTask).ConfigureAwait(false);
_processTask = ProcessAsync(); _processTask = ProcessAsync();
return true; return true;
} }
@ -222,14 +222,14 @@ namespace CryptoExchange.Net.Sockets
if (!Parameters.AutoReconnect) if (!Parameters.AutoReconnect)
{ {
_processState = ProcessState.Idle; _processState = ProcessState.Idle;
OnClose?.Invoke(); await (OnClose?.Invoke() ?? Task.CompletedTask).ConfigureAwait(false);
return; return;
} }
if (!_stopRequested) if (!_stopRequested)
{ {
_processState = ProcessState.Reconnecting; _processState = ProcessState.Reconnecting;
OnReconnecting?.Invoke(); await (OnReconnecting?.Invoke() ?? Task.CompletedTask).ConfigureAwait(false);
} }
var sinceLastReconnect = DateTime.UtcNow - _lastReconnectTime; var sinceLastReconnect = DateTime.UtcNow - _lastReconnectTime;
@ -263,7 +263,7 @@ namespace CryptoExchange.Net.Sockets
} }
_lastReconnectTime = DateTime.UtcNow; _lastReconnectTime = DateTime.UtcNow;
OnReconnected?.Invoke(); await (OnReconnected?.Invoke() ?? Task.CompletedTask).ConfigureAwait(false);
break; break;
} }
} }
@ -326,7 +326,7 @@ namespace CryptoExchange.Net.Sockets
await _closeTask.ConfigureAwait(false); await _closeTask.ConfigureAwait(false);
if(_processTask != null) if(_processTask != null)
await _processTask.ConfigureAwait(false); await _processTask.ConfigureAwait(false);
OnClose?.Invoke(); await (OnClose?.Invoke() ?? Task.CompletedTask).ConfigureAwait(false);
_logger.Log(LogLevel.Debug, $"Socket {Id} closed"); _logger.Log(LogLevel.Debug, $"Socket {Id} closed");
} }
@ -423,7 +423,7 @@ namespace CryptoExchange.Net.Sockets
try try
{ {
await _socket.SendAsync(new ArraySegment<byte>(data.Bytes, 0, data.Bytes.Length), WebSocketMessageType.Text, true, _ctsSource.Token).ConfigureAwait(false); await _socket.SendAsync(new ArraySegment<byte>(data.Bytes, 0, data.Bytes.Length), WebSocketMessageType.Text, true, _ctsSource.Token).ConfigureAwait(false);
OnRequestSent?.Invoke(data.Id); await (OnRequestSent?.Invoke(data.Id) ?? Task.CompletedTask).ConfigureAwait(false);
_logger.Log(LogLevel.Trace, $"Socket {Id} - msg {data.Id} - sent {data.Bytes.Length} bytes"); _logger.Log(LogLevel.Trace, $"Socket {Id} - msg {data.Id} - sent {data.Bytes.Length} bytes");
} }
catch (OperationCanceledException) catch (OperationCanceledException)
@ -434,7 +434,7 @@ namespace CryptoExchange.Net.Sockets
catch (Exception ioe) catch (Exception ioe)
{ {
// Connection closed unexpectedly, .NET framework // Connection closed unexpectedly, .NET framework
OnError?.Invoke(ioe); await (OnError?.Invoke(ioe) ?? Task.CompletedTask).ConfigureAwait(false);
if (_closeTask?.IsCompleted != false) if (_closeTask?.IsCompleted != false)
_closeTask = CloseInternalAsync(); _closeTask = CloseInternalAsync();
break; break;
@ -448,7 +448,7 @@ namespace CryptoExchange.Net.Sockets
// any exception here will crash the send processing, but do so silently unless the socket get's stopped. // any exception here will crash the send processing, but do so silently unless the socket get's stopped.
// Make sure we at least let the owner know there was an error // Make sure we at least let the owner know there was an error
_logger.Log(LogLevel.Warning, $"Socket {Id} Send loop stopped with exception"); _logger.Log(LogLevel.Warning, $"Socket {Id} Send loop stopped with exception");
OnError?.Invoke(e); await (OnError?.Invoke(e) ?? Task.CompletedTask).ConfigureAwait(false);
throw; throw;
} }
finally finally
@ -492,7 +492,7 @@ namespace CryptoExchange.Net.Sockets
catch (Exception wse) catch (Exception wse)
{ {
// Connection closed unexpectedly // Connection closed unexpectedly
OnError?.Invoke(wse); await (OnError?.Invoke(wse) ?? Task.CompletedTask).ConfigureAwait(false);
if (_closeTask?.IsCompleted != false) if (_closeTask?.IsCompleted != false)
_closeTask = CloseInternalAsync(); _closeTask = CloseInternalAsync();
break; break;
@ -571,7 +571,7 @@ namespace CryptoExchange.Net.Sockets
// any exception here will crash the receive processing, but do so silently unless the socket gets stopped. // any exception here will crash the receive processing, but do so silently unless the socket gets stopped.
// Make sure we at least let the owner know there was an error // Make sure we at least let the owner know there was an error
_logger.Log(LogLevel.Warning, $"Socket {Id} Receive loop stopped with exception"); _logger.Log(LogLevel.Warning, $"Socket {Id} Receive loop stopped with exception");
OnError?.Invoke(e); await (OnError?.Invoke(e) ?? Task.CompletedTask).ConfigureAwait(false);
throw; throw;
} }
finally finally
@ -627,7 +627,7 @@ namespace CryptoExchange.Net.Sockets
// Because this is running in a separate task and not awaited until the socket gets closed // Because this is running in a separate task and not awaited until the socket gets closed
// any exception here will stop the timeout checking, but do so silently unless the socket get's stopped. // any exception here will stop the timeout checking, but do so silently unless the socket get's stopped.
// Make sure we at least let the owner know there was an error // Make sure we at least let the owner know there was an error
OnError?.Invoke(e); await (OnError?.Invoke(e) ?? Task.CompletedTask).ConfigureAwait(false);
throw; throw;
} }
} }

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

@ -29,7 +29,7 @@ namespace CryptoExchange.Net.Sockets
/// <summary> /// <summary>
/// Strings to identify this subscription with /// Strings to identify this subscription with
/// </summary> /// </summary>
public abstract List<string> Identifiers { get; } public abstract List<string> StreamIdentifiers { get; }
/// <summary> /// <summary>
/// The query request object /// The query request object
@ -46,7 +46,7 @@ namespace CryptoExchange.Net.Sockets
/// </summary> /// </summary>
public int Weight { get; } public int Weight { get; }
public abstract Func<string, Type> ExpectedTypeDelegate { get; } public abstract Dictionary<string, Type> TypeMapping { get; }
/// <summary> /// <summary>
/// ctor /// ctor
@ -107,7 +107,10 @@ namespace CryptoExchange.Net.Sockets
/// <typeparam name="TResponse">Response object type</typeparam> /// <typeparam name="TResponse">Response object type</typeparam>
public abstract class Query<TResponse> : BaseQuery public abstract class Query<TResponse> : BaseQuery
{ {
public override Func<string, Type> ExpectedTypeDelegate => x => typeof(TResponse); public override Dictionary<string, Type> TypeMapping => new Dictionary<string, Type>
{
{ "", typeof(TResponse) }
};
/// <summary> /// <summary>
/// The typed call result /// The typed call result

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

@ -11,6 +11,8 @@ using System.IO;
using CryptoExchange.Net.Objects.Sockets; using CryptoExchange.Net.Objects.Sockets;
using System.Text; using System.Text;
using System.Diagnostics.CodeAnalysis; using System.Diagnostics.CodeAnalysis;
using CryptoExchange.Net.Converters;
using System.Diagnostics;
namespace CryptoExchange.Net.Sockets namespace CryptoExchange.Net.Sockets
{ {
@ -176,12 +178,12 @@ namespace CryptoExchange.Net.Sockets
_socket = socket; _socket = socket;
_socket.OnStreamMessage += HandleStreamMessage; _socket.OnStreamMessage += HandleStreamMessage;
_socket.OnRequestSent += HandleRequestSent; _socket.OnRequestSent += HandleRequestSentAsync;
_socket.OnOpen += HandleOpen; _socket.OnOpen += HandleOpenAsync;
_socket.OnClose += HandleClose; _socket.OnClose += HandleCloseAsync;
_socket.OnReconnecting += HandleReconnecting; _socket.OnReconnecting += HandleReconnectingAsync;
_socket.OnReconnected += HandleReconnected; _socket.OnReconnected += HandleReconnectedAsync;
_socket.OnError += HandleError; _socket.OnError += HandleErrorAsync;
_socket.GetReconnectionUrl = GetReconnectionUrlAsync; _socket.GetReconnectionUrl = GetReconnectionUrlAsync;
_listenerManager = new SocketListenerManager(_logger, SocketId); _listenerManager = new SocketListenerManager(_logger, SocketId);
@ -190,7 +192,7 @@ namespace CryptoExchange.Net.Sockets
/// <summary> /// <summary>
/// Handler for a socket opening /// Handler for a socket opening
/// </summary> /// </summary>
protected virtual void HandleOpen() protected virtual async Task HandleOpenAsync()
{ {
Status = SocketStatus.Connected; Status = SocketStatus.Connected;
PausedActivity = false; PausedActivity = false;
@ -199,7 +201,7 @@ namespace CryptoExchange.Net.Sockets
/// <summary> /// <summary>
/// Handler for a socket closing without reconnect /// Handler for a socket closing without reconnect
/// </summary> /// </summary>
protected virtual void HandleClose() protected virtual async Task HandleCloseAsync()
{ {
Status = SocketStatus.Closed; Status = SocketStatus.Closed;
Authenticated = false; Authenticated = false;
@ -219,7 +221,7 @@ namespace CryptoExchange.Net.Sockets
/// <summary> /// <summary>
/// Handler for a socket losing conenction and starting reconnect /// Handler for a socket losing conenction and starting reconnect
/// </summary> /// </summary>
protected virtual void HandleReconnecting() protected virtual async Task HandleReconnectingAsync()
{ {
Status = SocketStatus.Reconnecting; Status = SocketStatus.Reconnecting;
DisconnectTime = DateTime.UtcNow; DisconnectTime = DateTime.UtcNow;
@ -249,7 +251,7 @@ namespace CryptoExchange.Net.Sockets
/// <summary> /// <summary>
/// Handler for a socket which has reconnected /// Handler for a socket which has reconnected
/// </summary> /// </summary>
protected virtual async void HandleReconnected() protected virtual async Task HandleReconnectedAsync()
{ {
Status = SocketStatus.Resubscribing; Status = SocketStatus.Resubscribing;
@ -258,6 +260,7 @@ namespace CryptoExchange.Net.Sockets
query.Fail("Connection interupted"); query.Fail("Connection interupted");
_listenerManager.Remove(query); _listenerManager.Remove(query);
} }
// Mark subscription as 'not confirmed', only map updates to them if confirmed. Don't await sub answer here
var reconnectSuccessful = await ProcessReconnectAsync().ConfigureAwait(false); var reconnectSuccessful = await ProcessReconnectAsync().ConfigureAwait(false);
if (!reconnectSuccessful) if (!reconnectSuccessful)
@ -280,7 +283,7 @@ namespace CryptoExchange.Net.Sockets
/// Handler for an error on a websocket /// Handler for an error on a websocket
/// </summary> /// </summary>
/// <param name="e">The exception</param> /// <param name="e">The exception</param>
protected virtual void HandleError(Exception e) protected virtual async Task HandleErrorAsync(Exception e)
{ {
if (e is WebSocketException wse) if (e is WebSocketException wse)
_logger.Log(LogLevel.Warning, $"Socket {SocketId} error: Websocket error code {wse.WebSocketErrorCode}, details: " + e.ToLogString()); _logger.Log(LogLevel.Warning, $"Socket {SocketId} error: Websocket error code {wse.WebSocketErrorCode}, details: " + e.ToLogString());
@ -292,7 +295,7 @@ namespace CryptoExchange.Net.Sockets
/// Handler for whenever a request is sent over the websocket /// Handler for whenever a request is sent over the websocket
/// </summary> /// </summary>
/// <param name="requestId">Id of the request sent</param> /// <param name="requestId">Id of the request sent</param>
protected virtual void HandleRequestSent(int requestId) protected virtual async Task HandleRequestSentAsync(int requestId)
{ {
var query = _listenerManager.GetById<BaseQuery>(requestId); var query = _listenerManager.GetById<BaseQuery>(requestId);
if (query == null) if (query == null)
@ -312,7 +315,12 @@ namespace CryptoExchange.Net.Sockets
/// <returns></returns> /// <returns></returns>
protected virtual async Task HandleStreamMessage(WebSocketMessageType type, Stream stream) protected virtual async Task HandleStreamMessage(WebSocketMessageType type, Stream stream)
{ {
var result = ApiClient.StreamConverter.ReadJson(type, stream, _listenerManager, ApiClient.ApiOptions.OutputOriginalData ?? ApiClient.ClientOptions.OutputOriginalData); var buffer2 = new byte[stream.Length];
stream.Position = 0;
stream.Read(buffer2, 0, buffer2.Length);
Debug.WriteLine("0 " + Encoding.UTF8.GetString(buffer2));
stream.Position = 0;
var result = ReadJson(type, stream);
if (result == null) if (result == null)
{ {
// Not able to parse at all // Not able to parse at all
@ -349,6 +357,45 @@ namespace CryptoExchange.Net.Sockets
stream.Dispose(); stream.Dispose();
} }
/// <summary>
/// Read a message from stream
/// </summary>
/// <param name="websocketMessageType"></param>
/// <param name="stream"></param>
/// <returns></returns>
protected virtual BaseParsedMessage? ReadJson(WebSocketMessageType websocketMessageType, Stream stream)
{
// Start reading the data
// Once we reach the properties that identify the message we save those in a dict
// Once all id properties have been read callback to see what the deserialization type should be
// Deserialize to the correct type
if (ApiClient.Pipeline.PreProcessCallback != null)
stream = ApiClient.Pipeline.PreProcessCallback(websocketMessageType, stream);
var accessor = new JTokenAccessor(stream);
if (accessor == null)
return null;
var streamIdentity = ApiClient.Pipeline.GetStreamIdentifier(accessor);
if (streamIdentity == null)
return null;
var typeIdentity = ApiClient.Pipeline.GetTypeIdentifier(accessor);
var typeResult = _listenerManager.IdToType(streamIdentity, typeIdentity);
if (typeResult == null)
return null;
var idInstance = accessor.Instantiate(typeResult);
if (ApiClient.ApiOptions.OutputOriginalData ?? ApiClient.ClientOptions.OutputOriginalData)
idInstance.OriginalData = idInstance.OriginalData;
idInstance.StreamIdentifier = streamIdentity;
idInstance.TypeIdentifier = typeIdentity;
idInstance.Parsed = true;
return idInstance;
}
/// <summary> /// <summary>
/// Connect the websocket /// Connect the websocket
/// </summary> /// </summary>
@ -620,7 +667,11 @@ namespace CryptoExchange.Net.Sockets
taskList.Add(SendAndWaitQueryAsync(subQuery).ContinueWith((x) => taskList.Add(SendAndWaitQueryAsync(subQuery).ContinueWith((x) =>
{ {
Debug.WriteLine("1");
subscription.HandleSubQueryResponse(subQuery.Response); subscription.HandleSubQueryResponse(subQuery.Response);
Debug.WriteLine("2");
_listenerManager.Reset(subscription);
Debug.WriteLine("3");
return x.Result; return x.Result;
})); }));
} }

56
CryptoExchange.Net/Sockets/SocketListenerManager.cs
View File

@ -14,15 +14,15 @@ namespace CryptoExchange.Net.Sockets
private ILogger _logger; private ILogger _logger;
private int _socketId; private int _socketId;
private object _lock = new object(); private object _lock = new object();
private Dictionary<int, IMessageProcessor> _idMap; //private Dictionary<int, IMessageProcessor> _idMap;
private Dictionary<string, Func<string, Type>> _typeMap; //private Dictionary<string, Dictionary<string, Type>> _typeMap;
private Dictionary<string, List<IMessageProcessor>> _listeners; private Dictionary<string, List<IMessageProcessor>> _listeners;
public SocketListenerManager(ILogger logger, int socketId) public SocketListenerManager(ILogger logger, int socketId)
{ {
_idMap = new Dictionary<int, IMessageProcessor>(); //_idMap = new Dictionary<int, IMessageProcessor>();
_listeners = new Dictionary<string, List<IMessageProcessor>>(); _listeners = new Dictionary<string, List<IMessageProcessor>>();
_typeMap = new Dictionary<string, Func<string, Type>>(); //_typeMap = new Dictionary<string, Dictionary<string, Type>>();
_logger = logger; _logger = logger;
_socketId = socketId; _socketId = socketId;
} }
@ -31,8 +31,12 @@ namespace CryptoExchange.Net.Sockets
{ {
lock (_lock) lock (_lock)
{ {
_typeMap.TryGetValue(streamIdentifier, out var typeDelegate); _listeners.TryGetValue(streamIdentifier, out var listeners);
return typeDelegate?.Invoke(typeIdentifier); if (listeners == null)
return null;
listeners.First().TypeMapping.TryGetValue(typeIdentifier ?? "", out var type);
return type;
} }
} }
@ -46,10 +50,9 @@ namespace CryptoExchange.Net.Sockets
{ {
lock (_lock) lock (_lock)
{ {
_idMap.Add(processor.Id, processor); if (processor.StreamIdentifiers?.Any() == true)
if (processor.Identifiers?.Any() == true)
{ {
foreach (var identifier in processor.Identifiers) foreach (var identifier in processor.StreamIdentifiers)
{ {
if (!_listeners.TryGetValue(identifier, out var list)) if (!_listeners.TryGetValue(identifier, out var list))
{ {
@ -61,7 +64,16 @@ namespace CryptoExchange.Net.Sockets
} }
} }
UpdateMap(); }
}
public void Reset(IMessageProcessor processor)
{
lock (_lock)
{
Debug.WriteLine("4 Resetting");
Remove(processor);
Add(processor);
} }
} }
@ -116,7 +128,7 @@ namespace CryptoExchange.Net.Sockets
{ {
lock (_lock) lock (_lock)
{ {
_idMap.TryGetValue(id, out var val); var val = _listeners.Values.SelectMany(x => x).FirstOrDefault(x => x.Id == id);
return (T)val; return (T)val;
} }
} }
@ -143,24 +155,16 @@ namespace CryptoExchange.Net.Sockets
{ {
lock (_lock) lock (_lock)
{ {
_idMap.Remove(processor.Id); if (processor.StreamIdentifiers?.Any() != true)
if (processor.Identifiers?.Any() == true) return;
foreach(var kv in _listeners)
{ {
foreach (var identifier in processor.Identifiers) if (kv.Value.Contains(processor))
{ kv.Value.Remove(processor);
_listeners[identifier].Remove(processor); }
if (!_listeners[identifier].Any())
_listeners.Remove(identifier);
} }
} }
UpdateMap();
}
}
private void UpdateMap()
{
_typeMap = _listeners.ToDictionary(x => x.Key, x => x.Value.First().ExpectedTypeDelegate);
}
} }
} }

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

@ -57,7 +57,7 @@ namespace CryptoExchange.Net.Sockets
/// <summary> /// <summary>
/// Strings to identify this subscription with /// Strings to identify this subscription with
/// </summary> /// </summary>
public abstract List<string> Identifiers { get; } public abstract List<string> StreamIdentifiers { get; }
/// <summary> /// <summary>
/// Cancellation token registration /// Cancellation token registration
@ -69,7 +69,7 @@ namespace CryptoExchange.Net.Sockets
/// </summary> /// </summary>
public event Action<Exception>? Exception; public event Action<Exception>? Exception;
public abstract Func<string, Type> ExpectedTypeDelegate { get; } public abstract Dictionary<string, Type> TypeMapping { get; }
/// <summary> /// <summary>
/// ctor /// ctor

7
CryptoExchange.Net/Sockets/SystemSubscription.cs
View File

@ -2,6 +2,7 @@
using CryptoExchange.Net.Objects.Sockets; using CryptoExchange.Net.Objects.Sockets;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using System; using System;
using System.Collections.Generic;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace CryptoExchange.Net.Sockets namespace CryptoExchange.Net.Sockets
@ -29,7 +30,11 @@ namespace CryptoExchange.Net.Sockets
public abstract class SystemSubscription<T> : SystemSubscription public abstract class SystemSubscription<T> : SystemSubscription
{ {
public override Func<string, Type> ExpectedTypeDelegate => (x) => typeof(T); public override Dictionary<string, Type> TypeMapping => new Dictionary<string, Type>
{
{ "", typeof(T) }
};
public override Task<CallResult> DoHandleMessageAsync(SocketConnection connection, DataEvent<BaseParsedMessage> message) public override Task<CallResult> DoHandleMessageAsync(SocketConnection connection, DataEvent<BaseParsedMessage> message)
=> HandleMessageAsync(connection, message.As((ParsedMessage<T>)message.Data)); => HandleMessageAsync(connection, message.As((ParsedMessage<T>)message.Data));