diff --git a/CryptoExchange.Net.UnitTests/SocketRoutingTests/SubscriptionTests.cs b/CryptoExchange.Net.UnitTests/SocketRoutingTests/SubscriptionTests.cs new file mode 100644 index 00000000..1d89c323 --- /dev/null +++ b/CryptoExchange.Net.UnitTests/SocketRoutingTests/SubscriptionTests.cs @@ -0,0 +1,130 @@ +using CryptoExchange.Net.Objects; +using CryptoExchange.Net.Sockets; +using CryptoExchange.Net.Sockets.Default; +using CryptoExchange.Net.Sockets.Default.Routing; +using CryptoExchange.Net.UnitTests.Implementations; +using Microsoft.Extensions.Logging.Abstractions; +using NUnit.Framework; +using System; + +namespace CryptoExchange.Net.UnitTests.SocketRoutingTests +{ + [TestFixture] + public class SubscriptionTests + { + [Test] + public void Handle_Should_OnlyCompleteSubscriptionQuery_ForMatchingTopic() + { + // arrange + var topicASubscription = new TopicSubscription("topic-a"); + var topicBSubscription = new TopicSubscription("topic-b"); + var topicAQuery = topicASubscription.CreateSubscriptionQuery(null!)!; + var topicBQuery = topicBSubscription.CreateSubscriptionQuery(null!)!; + + // act + var topicAHandled = topicASubscription.Handle("type", "topic-a", null!, DateTime.UtcNow, "original", "data"); + var topicBHandled = topicBSubscription.Handle("type", "topic-a", null!, DateTime.UtcNow, "original", "data"); + + // assert + Assert.That(topicAHandled, Is.True); + Assert.That(topicBHandled, Is.False); + Assert.That(topicAQuery.Completed, Is.True); + Assert.That(topicAQuery.Success, Is.True); + Assert.That(topicBQuery.Completed, Is.False); + } + + [Test] + public void Handle_Should_CompleteSubscriptionQuery_BeforeInvokingMatchingHandler() + { + // arrange + Query? query = null; + var queryCompletedWhenHandlerInvoked = false; + var subscription = new TopicSubscription("topic", () => queryCompletedWhenHandlerInvoked = query!.Completed); + query = subscription.CreateSubscriptionQuery(null!)!; + + // act + subscription.Handle("type", "topic", null!, DateTime.UtcNow, "original", "data"); + + // assert + Assert.That(queryCompletedWhenHandlerInvoked, Is.True); + } + + [TestCase(null)] + [TestCase("topic")] + public void Handle_Should_CompleteSubscriptionQuery_ForUnfilteredRoute(string? topicFilter) + { + // arrange + var subscription = new TopicSubscription(topic: null); + var query = subscription.CreateSubscriptionQuery(null!)!; + + // act + var handled = subscription.Handle("type", topicFilter, null!, DateTime.UtcNow, "original", "data"); + + // assert + Assert.That(handled, Is.True); + Assert.That(query.Completed, Is.True); + Assert.That(query.Success, Is.True); + } + + [Test] + public void Handle_Should_TreatEmptyTopicFilterAsUnfilteredRoute() + { + // arrange + var subscription = new TopicSubscription(string.Empty); + var query = subscription.CreateSubscriptionQuery(null!)!; + + // act + var handled = subscription.Handle("type", "topic", null!, DateTime.UtcNow, "original", "data"); + + // assert + Assert.That(handled, Is.True); + Assert.That(query.Completed, Is.True); + Assert.That(query.Success, Is.True); + } + + [Test] + public void Handle_Should_CompleteSubscriptionQuery_ForAnyMatchingTopic() + { + // arrange + var subscription = new TopicSubscription(["topic-a", "topic-b"]); + var query = subscription.CreateSubscriptionQuery(null!)!; + + // act + var handled = subscription.Handle("type", "topic-b", null!, DateTime.UtcNow, "original", "data"); + + // assert + Assert.That(handled, Is.True); + Assert.That(query.Completed, Is.True); + Assert.That(query.Success, Is.True); + } + + private sealed class TopicSubscription : Subscription + { + public TopicSubscription(string? topic, Action? handler = null) + : base(NullLogger.Instance, false) + { + MessageRouter = MessageRouter.CreateForEvent("type", topic, (_, _, _, _) => + { + handler?.Invoke(); + return CallResult.Ok(); + }); + } + + public TopicSubscription(string[] topics) + : base(NullLogger.Instance, false) + { + MessageRouter = MessageRouter.CreateForEvent("type", topics, (_, _, _, _) => CallResult.Ok()); + } + + protected override Query? GetSubQuery(SocketConnection connection) + { + return new TestQuery(new TestSocketMessage { Id = 1, Data = "Sub" }, false) + { + TimeoutBehavior = TimeoutBehavior.Succeed + }; + } + + protected override Query? GetUnsubQuery(SocketConnection connection) => null; + } + } +} diff --git a/CryptoExchange.Net/Sockets/Default/Routing/MessageRouter.cs b/CryptoExchange.Net/Sockets/Default/Routing/MessageRouter.cs index 01e5ee73..10332242 100644 --- a/CryptoExchange.Net/Sockets/Default/Routing/MessageRouter.cs +++ b/CryptoExchange.Net/Sockets/Default/Routing/MessageRouter.cs @@ -25,6 +25,9 @@ namespace CryptoExchange.Net.Sockets.Default.Routing Routes = routes; } + private RouteCollection? GetRoutes(string typeIdentifier) + => (_routingTable ?? throw new NullReferenceException("Routing table not build before handling")).GetRoutes(typeIdentifier); + /// /// Build the route mapping /// @@ -46,13 +49,21 @@ namespace CryptoExchange.Net.Sockets.Default.Routing /// public bool Handle(string typeIdentifier, string? topicFilter, SocketConnection connection, DateTime receiveTime, string? originalData, object data, out CallResult? result) { - var routeCollection = (_routingTable ?? throw new NullReferenceException("Routing table not build before handling")).GetRoutes(typeIdentifier); + var routeCollection = GetRoutes(typeIdentifier); if (routeCollection == null) throw new InvalidOperationException($"No routes for {typeIdentifier} message type"); return routeCollection.Handle(topicFilter, connection, receiveTime, originalData, data, out result); } + /// + /// Whether the router has a matching route for the specified message type and topic + /// + public bool CanHandle(string typeIdentifier, string? topicFilter) + { + return GetRoutes(typeIdentifier)?.CanHandle(topicFilter) == true; + } + /// /// Create a void handler /// diff --git a/CryptoExchange.Net/Sockets/Default/Routing/RouteCollection.cs b/CryptoExchange.Net/Sockets/Default/Routing/RouteCollection.cs index 48df1979..90a7f3f4 100644 --- a/CryptoExchange.Net/Sockets/Default/Routing/RouteCollection.cs +++ b/CryptoExchange.Net/Sockets/Default/Routing/RouteCollection.cs @@ -60,6 +60,15 @@ namespace CryptoExchange.Net.Sockets.Default.Routing return matchingTopicRoutes; } + public bool CanHandle(string? topicFilter) + { + if (_routesWithoutTopicFilter.Count > 0) + return true; + + return topicFilter != null + && GetRoutesWithMatchingTopicFilter(topicFilter) != null; + } + public abstract bool Handle(string? topicFilter, SocketConnection connection, DateTime receiveTime, string? originalData, object data, out CallResult? result); } } diff --git a/CryptoExchange.Net/Sockets/Default/Subscription.cs b/CryptoExchange.Net/Sockets/Default/Subscription.cs index 7c025ea3..0ed13739 100644 --- a/CryptoExchange.Net/Sockets/Default/Subscription.cs +++ b/CryptoExchange.Net/Sockets/Default/Subscription.cs @@ -219,9 +219,13 @@ namespace CryptoExchange.Net.Sockets.Default ConnectionInvocations++; TotalInvocations++; - if (SubscriptionQuery != null && !SubscriptionQuery.Completed && SubscriptionQuery.TimeoutBehavior == TimeoutBehavior.Succeed) + if (SubscriptionQuery != null + && !SubscriptionQuery.Completed + && SubscriptionQuery.TimeoutBehavior == TimeoutBehavior.Succeed + && MessageRouter.CanHandle(typeIdentifier, topicFilter)) { // The subscription query is one where it is successful if there is no error returned + // The connection routes by type, so only a matching subscription route proves the query was successful // Since we've received a data update for the subscription we can assume the subscribe query was successful // Call timeout to complete SubscriptionQuery.Timeout();