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

wip

Browse Source
This commit is contained in:
Jkorf 2026-04-08 09:33:45 +02:00
parent 61c66300af
commit f66730ca8c
8 changed files with 423 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
CryptoExchange.Net/Sockets/Default/Routing/MessageRoute.cs
View File

@ -101,4 +101,14 @@ namespace CryptoExchange.Net.Sockets.Default.Routing
return _handler(connection, receiveTime, originalData, (TMessage)data);
}
}
//public class SubscriptionRoute<TMessage> : MessageRoute<TMessage>
//{
//}
//public class QuertyRoute<TMessage> : MessageRoute<TMessage>
//{
//}
}

21
CryptoExchange.Net/Sockets/Default/Routing/MessageRouter.cs
View File

@ -10,7 +10,7 @@ namespace CryptoExchange.Net.Sockets.Default.Routing
/// </summary>
public class MessageRouter
{
private RoutingSubTable? _routingTable;
private ProcessorRouter? _routingTable;
/// <summary>
/// The routes registered for this router
@ -28,17 +28,26 @@ namespace CryptoExchange.Net.Sockets.Default.Routing
/// <summary>
/// Build the route mapping
/// </summary>
public void BuildRouteMap()
public void BuildQueryRouteMap()
{
_routingTable = new RoutingSubTable(Routes);
_routingTable = new QueryRouter(Routes);
}
/// <summary>
/// Get routes matching the type identifier
/// Build the route mapping
/// </summary>
internal RoutingSubTableEntry? this[string identifier]
public void BuildSubscriptionRouteMap()
{
get => (_routingTable ?? throw new InvalidOperationException("Route map not initialized before use"))[identifier];
_routingTable = new SubscriptionRouter(Routes);
}
public bool Handle(string typeIdentifier, string? topicFilter, SocketConnection connection, DateTime receiveTime, string? originalData, object data, out CallResult? result)
{
var routeCollection = _routingTable.GetRoutes(typeIdentifier);
if (routeCollection == null)
throw new InvalidOperationException($"No routes for {typeIdentifier} message type");
return routeCollection.Handle(topicFilter, connection, receiveTime, originalData, data, out result);
}
/// <summary>

103
CryptoExchange.Net/Sockets/Default/Routing/ProcessorRouter.cs Normal file
View File

@ -0,0 +1,103 @@
using CryptoExchange.Net.Objects;
using System;
using System.Collections.Generic;
using System.Text;
#if NET8_0_OR_GREATER
using System.Collections.Frozen;
#endif
namespace CryptoExchange.Net.Sockets.Default.Routing
{
internal abstract class ProcessorRouter
{
public abstract RouteCollection? GetRoutes(string identifier);
}
internal abstract class ProcessorRouter<T> : ProcessorRouter
where T : RouteCollection
{
#if NET8_0_OR_GREATER
private FrozenDictionary<string, T> _routeMap;
#else
private Dictionary<string, T> _routeMap;
#endif
public ProcessorRouter(IEnumerable<MessageRoute> routes)
{
var map = BuildFromRoutes(routes);
#if NET8_0_OR_GREATER
_routeMap = map.ToFrozenDictionary();
#else
_routeMap = map;
#endif
}
public abstract Dictionary<string, T> BuildFromRoutes(IEnumerable<MessageRoute> routes);
public override RouteCollection? GetRoutes(string identifier) => this[identifier];
/// <summary>
/// Get routes matching the type identifier
/// </summary>
public T? this[string identifier]
{
get => _routeMap.TryGetValue(identifier, out var routes) ? routes : null;
}
}
internal abstract class RouteCollection
{
protected List<MessageRoute> _routesWithoutTopicFilter;
protected Dictionary<string, List<MessageRoute>> _routesWithTopicFilter;
#if NET8_0_OR_GREATER
protected FrozenDictionary<string, List<MessageRoute>>? _routesWithTopicFilterFrozen;
#endif
public Type DeserializationType { get; }
public RouteCollection(Type routeType)
{
_routesWithoutTopicFilter = new List<MessageRoute>();
_routesWithTopicFilter = new Dictionary<string, List<MessageRoute>>();
DeserializationType = routeType;
}
public virtual void AddRoute(string? topicFilter, MessageRoute route)
{
if (string.IsNullOrEmpty(topicFilter))
{
_routesWithoutTopicFilter.Add(route);
}
else
{
if (!_routesWithTopicFilter.TryGetValue(topicFilter!, out var list))
{
list = new List<MessageRoute>();
_routesWithTopicFilter.Add(topicFilter!, list);
}
list.Add(route);
}
}
public void Build()
{
#if NET8_0_OR_GREATER
_routesWithTopicFilterFrozen = _routesWithTopicFilter.ToFrozenDictionary();
#endif
}
protected List<MessageRoute>? GetRoutesWithMatchingTopicFilter(string topicFilter)
{
#if NET8_0_OR_GREATER
_routesWithTopicFilterFrozen!.TryGetValue(topicFilter, out var matchingTopicRoutes);
#else
_routesWithTopicFilter.TryGetValue(topicFilter, out var matchingTopicRoutes);
#endif
return matchingTopicRoutes;
}
public abstract bool Handle(string? topicFilter, SocketConnection connection, DateTime receiveTime, string? originalData, object data, out CallResult? result);
}
}

95
CryptoExchange.Net/Sockets/Default/Routing/QueryRouter.cs Normal file
View File

@ -0,0 +1,95 @@
using CryptoExchange.Net.Objects;
using System;
using System.Collections.Generic;
using System.Text;
#if NET8_0_OR_GREATER
using System.Collections.Frozen;
#endif
namespace CryptoExchange.Net.Sockets.Default.Routing
{
internal class QueryRouter : ProcessorRouter<QueryRouteCollection>
{
public QueryRouter(IEnumerable<MessageRoute> routes) : base(routes)
{
}
public override Dictionary<string, QueryRouteCollection> BuildFromRoutes(IEnumerable<MessageRoute> routes)
{
var newMap = new Dictionary<string, QueryRouteCollection>();
foreach (var route in routes)
{
if (!newMap.TryGetValue(route.TypeIdentifier, out var typeMap))
{
typeMap = new QueryRouteCollection(route.DeserializationType);
newMap.Add(route.TypeIdentifier, typeMap);
}
typeMap.AddRoute(route.TopicFilter, route);
}
foreach (var subEntry in newMap.Values)
subEntry.Build();
return newMap;
}
}
internal class QueryRouteCollection : RouteCollection
{
public bool MultipleReaders { get; private set; }
public QueryRouteCollection(Type routeType) : base(routeType)
{
}
public override void AddRoute(string? topicFilter, MessageRoute route)
{
base.AddRoute(topicFilter, route);
if (route.MultipleReaders)
MultipleReaders = true;
}
public override bool Handle(string? topicFilter, SocketConnection connection, DateTime receiveTime, string? originalData, object data, out CallResult? result)
{
result = null;
// Routes without topic filter handle both when the message topic is empty and when it is not, so we always call them
var handled = false;
foreach (var route in _routesWithoutTopicFilter)
{
var thisResult = route.Handle(connection, receiveTime, originalData, data);
if (thisResult != null)
result ??= thisResult;
handled = true;
}
// Forward to routes with matching topic filter, if any
if (topicFilter != null)
{
#if NET8_0_OR_GREATER
_routesWithTopicFilterFrozen!.TryGetValue(topicFilter, out var matchingTopicRoutes);
#else
_routesWithTopicFilter.TryGetValue(topicFilter, out var matchingTopicRoutes);
#endif
foreach (var route in matchingTopicRoutes ?? [])
{
var thisResult = route.Handle(connection, receiveTime, originalData, data);
handled = true;
if (thisResult != null)
{
result ??= thisResult;
if (!MultipleReaders)
break;
}
}
}
return handled;
}
}
}

240
CryptoExchange.Net/Sockets/Default/Routing/RoutingSubTable.cs
View File

@ -1,142 +1,142 @@
using CryptoExchange.Net.Objects;
using System;
#if NET8_0_OR_GREATER
using System.Collections.Frozen;
#endif
using System.Collections.Generic;
//using CryptoExchange.Net.Objects;
//using System;
//#if NET8_0_OR_GREATER
//using System.Collections.Frozen;
//#endif
//using System.Collections.Generic;
namespace CryptoExchange.Net.Sockets.Default.Routing
{
internal class RoutingSubTable
{
#if NET8_0_OR_GREATER
// Used for mapping a type identifier to the routes matching it
private FrozenDictionary<string, RoutingSubTableEntry> _routeMap;
#else
private Dictionary<string, RoutingSubTableEntry> _routeMap;
#endif
//namespace CryptoExchange.Net.Sockets.Default.Routing
//{
// internal class RoutingSubTable
// {
//#if NET8_0_OR_GREATER
// // Used for mapping a type identifier to the routes matching it
// private FrozenDictionary<string, RoutingSubTableEntry> _routeMap;
//#else
// private Dictionary<string, RoutingSubTableEntry> _routeMap;
//#endif
public RoutingSubTable(IEnumerable<MessageRoute> routes)
{
var newMap = new Dictionary<string, RoutingSubTableEntry>();
foreach (var route in routes)
{
if (!newMap.TryGetValue(route.TypeIdentifier, out var typeMap))
{
typeMap = new RoutingSubTableEntry(route.DeserializationType);
newMap.Add(route.TypeIdentifier, typeMap);
}
// public RoutingSubTable(IEnumerable<MessageRoute> routes)
// {
// var newMap = new Dictionary<string, RoutingSubTableEntry>();
// foreach (var route in routes)
// {
// if (!newMap.TryGetValue(route.TypeIdentifier, out var typeMap))
// {
// typeMap = new RoutingSubTableEntry(route.DeserializationType);
// newMap.Add(route.TypeIdentifier, typeMap);
// }
typeMap.AddRoute(route.TopicFilter, route);
}
// typeMap.AddRoute(route.TopicFilter, route);
// }
foreach(var subEntry in newMap.Values)
subEntry.Build();
// foreach(var subEntry in newMap.Values)
// subEntry.Build();
#if NET8_0_OR_GREATER
_routeMap = newMap.ToFrozenDictionary();
#else
_routeMap = newMap;
#endif
}
//#if NET8_0_OR_GREATER
// _routeMap = newMap.ToFrozenDictionary();
//#else
// _routeMap = newMap;
//#endif
// }
/// <summary>
/// Get routes matching the type identifier
/// </summary>
public RoutingSubTableEntry? this[string identifier]
{
get => _routeMap.TryGetValue(identifier, out var routes) ? routes : null;
}
}
// /// <summary>
// /// Get routes matching the type identifier
// /// </summary>
// public RoutingSubTableEntry? this[string identifier]
// {
// get => _routeMap.TryGetValue(identifier, out var routes) ? routes : null;
// }
// }
internal record RoutingSubTableEntry
{
public Type DeserializationType { get; }
public bool MultipleReaders { get; private set; }
// internal record RoutingSubTableEntry
// {
// public Type DeserializationType { get; }
// public bool MultipleReaders { get; private set; }
private List<MessageRoute> _routesWithoutTopicFilter;
private Dictionary<string, List<MessageRoute>> _routesWithTopicFilter;
#if NET8_0_OR_GREATER
private FrozenDictionary<string, List<MessageRoute>>? _routesWithTopicFilterFrozen;
#endif
// private List<MessageRoute> _routesWithoutTopicFilter;
// private Dictionary<string, List<MessageRoute>> _routesWithTopicFilter;
//#if NET8_0_OR_GREATER
// private FrozenDictionary<string, List<MessageRoute>>? _routesWithTopicFilterFrozen;
//#endif
public RoutingSubTableEntry(Type routeType)
{
_routesWithoutTopicFilter = new List<MessageRoute>();
_routesWithTopicFilter = new Dictionary<string, List<MessageRoute>>();
// public RoutingSubTableEntry(Type routeType)
// {
// _routesWithoutTopicFilter = new List<MessageRoute>();
// _routesWithTopicFilter = new Dictionary<string, List<MessageRoute>>();
DeserializationType = routeType;
}
// DeserializationType = routeType;
// }
public void AddRoute(string? topicFilter, MessageRoute route)
{
if (string.IsNullOrEmpty(topicFilter))
{
_routesWithoutTopicFilter.Add(route);
}
else
{
if (!_routesWithTopicFilter.TryGetValue(topicFilter!, out var list))
{
list = new List<MessageRoute>();
_routesWithTopicFilter.Add(topicFilter!, list);
}
// public void AddRoute(string? topicFilter, MessageRoute route)
// {
// if (string.IsNullOrEmpty(topicFilter))
// {
// _routesWithoutTopicFilter.Add(route);
// }
// else
// {
// if (!_routesWithTopicFilter.TryGetValue(topicFilter!, out var list))
// {
// list = new List<MessageRoute>();
// _routesWithTopicFilter.Add(topicFilter!, list);
// }
list.Add(route);
}
// list.Add(route);
// }
if (route.MultipleReaders)
MultipleReaders = true;
}
// if (route.MultipleReaders)
// MultipleReaders = true;
// }
public void Build()
{
#if NET8_0_OR_GREATER
_routesWithTopicFilterFrozen = _routesWithTopicFilter.ToFrozenDictionary();
#endif
}
// public void Build()
// {
//#if NET8_0_OR_GREATER
// _routesWithTopicFilterFrozen = _routesWithTopicFilter.ToFrozenDictionary();
//#endif
// }
internal bool Handle(string? topicFilter, SocketConnection connection, DateTime receiveTime, string? originalData, object data, out CallResult? result)
{
result = null;
// internal bool Handle(string? topicFilter, SocketConnection connection, DateTime receiveTime, string? originalData, object data, out CallResult? result)
// {
// result = null;
// Routes without topic filter handle both when the message topic is empty and when it is not, so we always call them
var handled = false;
foreach (var route in _routesWithoutTopicFilter)
{
var thisResult = route.Handle(connection, receiveTime, originalData, data);
if (thisResult != null)
result ??= thisResult;
// // Routes without topic filter handle both when the message topic is empty and when it is not, so we always call them
// var handled = false;
// foreach (var route in _routesWithoutTopicFilter)
// {
// var thisResult = route.Handle(connection, receiveTime, originalData, data);
// if (thisResult != null)
// result ??= thisResult;
handled = true;
}
// handled = true;
// }
// Forward to routes with matching topic filter, if any
if (topicFilter != null)
{
#if NET8_0_OR_GREATER
_routesWithTopicFilterFrozen!.TryGetValue(topicFilter, out var matchingTopicRoutes);
#else
_routesWithTopicFilter.TryGetValue(topicFilter, out var matchingTopicRoutes);
#endif
foreach (var route in matchingTopicRoutes ?? [])
{
var thisResult = route.Handle(connection, receiveTime, originalData, data);
handled = true;
// // Forward to routes with matching topic filter, if any
// if (topicFilter != null)
// {
//#if NET8_0_OR_GREATER
// _routesWithTopicFilterFrozen!.TryGetValue(topicFilter, out var matchingTopicRoutes);
//#else
// _routesWithTopicFilter.TryGetValue(topicFilter, out var matchingTopicRoutes);
//#endif
// foreach (var route in matchingTopicRoutes ?? [])
// {
// var thisResult = route.Handle(connection, receiveTime, originalData, data);
// handled = true;
if (thisResult != null)
{
result ??= thisResult;
// if (thisResult != null)
// {
// result ??= thisResult;
#warning MultipleReaders is only for queries, subscriptions should always have multiple readers = true. Maybe create different RoutingSubTable implementations for Queries and Subscriptions?
if (!MultipleReaders)
break;
}
//#warning MultipleReaders is only for queries, subscriptions should always have multiple readers = true. Maybe create different RoutingSubTable implementations for Queries and Subscriptions?
// if (!MultipleReaders)
// break;
// }
}
}
// }
// }
return handled;
}
}
}
// return handled;
// }
// }
//}

76
CryptoExchange.Net/Sockets/Default/Routing/SubscriptionRouter.cs Normal file
View File

@ -0,0 +1,76 @@
using CryptoExchange.Net.Objects;
using System;
using System.Collections.Generic;
using System.Text;
#if NET8_0_OR_GREATER
using System.Collections.Frozen;
#endif
namespace CryptoExchange.Net.Sockets.Default.Routing
{
internal class SubscriptionRouter : ProcessorRouter<SubscriptionRouteCollection>
{
public SubscriptionRouter(IEnumerable<MessageRoute> routes) : base(routes)
{
}
public override Dictionary<string, SubscriptionRouteCollection> BuildFromRoutes(IEnumerable<MessageRoute> routes)
{
var newMap = new Dictionary<string, SubscriptionRouteCollection>();
foreach (var route in routes)
{
if (!newMap.TryGetValue(route.TypeIdentifier, out var typeMap))
{
typeMap = new SubscriptionRouteCollection(route.DeserializationType);
newMap.Add(route.TypeIdentifier, typeMap);
}
typeMap.AddRoute(route.TopicFilter, route);
}
foreach (var subEntry in newMap.Values)
subEntry.Build();
return newMap;
}
}
internal class SubscriptionRouteCollection : RouteCollection
{
public SubscriptionRouteCollection(Type routeType) : base(routeType)
{
}
public override bool Handle(string? topicFilter, SocketConnection connection, DateTime receiveTime, string? originalData, object data, out CallResult? result)
{
result = null;
// Routes without topic filter handle both when the message topic is empty and when it is not, so we always call them
var handled = false;
foreach (var route in _routesWithoutTopicFilter)
{
var thisResult = route.Handle(connection, receiveTime, originalData, data);
if (thisResult != null)
result ??= thisResult;
handled = true;
}
// Forward to routes with matching topic filter, if any
if (topicFilter != null)
{
var matchingTopicRoutes = GetRoutesWithMatchingTopicFilter(topicFilter);
foreach (var route in matchingTopicRoutes ?? [])
{
var thisResult = route.Handle(connection, receiveTime, originalData, data);
handled = true;
if (thisResult != null)
result ??= thisResult;
}
}
return handled;
}
}
}

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

@ -82,7 +82,7 @@ namespace CryptoExchange.Net.Sockets.Default
set
{
_router = value;
_router.BuildRouteMap();
_router.BuildSubscriptionRouteMap();
OnMessageRouterUpdated?.Invoke();
}
}
@ -198,7 +198,7 @@ namespace CryptoExchange.Net.Sockets.Default
SubscriptionQuery.Timeout();
}
return MessageRouter[typeIdentifier]?.Handle(topicFilter, connection, receiveTime, originalData, data, out _) ?? false;
return MessageRouter.Handle(typeIdentifier, topicFilter, connection, receiveTime, originalData, data, out _);
}
/// <summary>

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

@ -70,7 +70,7 @@ namespace CryptoExchange.Net.Sockets
set
{
_router = value;
_router.BuildRouteMap();
_router.BuildQueryRouteMap();
OnMessageRouterUpdated?.Invoke();
}
}
@ -211,7 +211,7 @@ namespace CryptoExchange.Net.Sockets
if (Result?.Success != false)
{
// If an error result is already set don't override that
MessageRouter[typeIdentifier]!.Handle(topicFilter, connection, receiveTime, originalData, message, out var result);
MessageRouter.Handle(typeIdentifier, topicFilter, connection, receiveTime, originalData, message, out var result);
Result = result;
if (Result == null)
// Null from Handle means it wasn't actually for this query