Fix subscription query completion for topic-filtered routes (#288)

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<string>("type", topic, (_, _, _, _) =>
+                {
+                    handler?.Invoke();
+                    return CallResult.Ok();
+                });
+            }
+
+            public TopicSubscription(string[] topics)
+                : base(NullLogger.Instance, false)
+            {
+                MessageRouter = MessageRouter.CreateForEvent<string>("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;
+        }
+    }
+}

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);
+
         /// <summary>
         /// Build the route mapping
         /// </summary>
@@ -46,13 +49,21 @@ namespace CryptoExchange.Net.Sockets.Default.Routing
         /// </summary>
         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);
         }
 
+        /// <summary>
+        /// Whether the router has a matching route for the specified message type and topic
+        /// </summary>
+        public bool CanHandle(string typeIdentifier, string? topicFilter)
+        {
+            return GetRoutes(typeIdentifier)?.CanHandle(topicFilter) == true;
+        }
+
         /// <summary>
         /// Create a void handler
         /// </summary>

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);
     }
 }

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();