When listening for messages on a specific topic, now comparing based on topic filter including wildcards

Author: sroebert

Sources/MQTTNIO/MQTTClient+Concurrency.swift

@@ -247,10 +247,10 @@ extension MQTTClient {
     }
 
     /// An async sequence for iterating over received messages from the broker to a specific topic.
-    /// - Parameter topic: The topic to receive messages for.
-    public func messages(forTopic topic: String) -> AsyncFilterSequence<AsyncStream<MQTTMessage>> {
+    /// - Parameter topicFilter: The topic filter to receive messages for.
+    public func messages(forTopic topicFilter: String) -> AsyncFilterSequence<AsyncStream<MQTTMessage>> {
         return messages.filter {
-            $0.topic == topic
+            $0.topic.matchesMqttTopicFilter(topicFilter)
         }
     }
 

Sources/MQTTNIO/MQTTClient.swift

@@ -555,13 +555,13 @@ public class MQTTClient: MQTTConnectionDelegate, MQTTSubscriptionsHandlerDelegat
 
     /// Adds an observer callback which will be called when the client has received an `MQTTMessage` for a specific topic.
     /// - Parameters:
-    ///   - topic: The topic to receive messages for.
+    ///   - topicFilter: The topic filter to receive messages for.
     ///   - callback: The observer callback to add which will be called with the received message.
     /// - Returns: An `MQTTCancellable` which can be used to cancel the observer callback.
     @discardableResult
-    public func whenMessage(forTopic topic: String, _ callback: @escaping (_ message: MQTTMessage) -> Void) -> MQTTCancellable {
+    public func whenMessage(forTopic topicFilter: String, _ callback: @escaping (_ message: MQTTMessage) -> Void) -> MQTTCancellable {
         return messageCallbacks.append { message in
-            guard message.topic == topic else {
+            guard message.topic.matchesMqttTopicFilter(topicFilter) else {
                 return
             }
             callback(message)
@@ -628,13 +628,13 @@ public class MQTTClient: MQTTConnectionDelegate, MQTTSubscriptionsHandlerDelegat
     }
 
     /// Returns a publisher for receiving MQTT messages to a specific topic.
-    /// - Parameter topic: The topic to receive messages for.
+    /// - Parameter topicFilter: The topic filter to receive messages for.
     ///
     /// This is only available on platforms where `Combine` is available.
     @available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
-    public func messagePublisher(forTopic topic: String) -> AnyPublisher<MQTTMessage, Never> {
+    public func messagePublisher(forTopic topicFilter: String) -> AnyPublisher<MQTTMessage, Never> {
         return messageSubject
-            .filter { $0.topic == topic }
+            .filter { $0.topic.matchesMqttTopicFilter(topicFilter) }
             .eraseToAnyPublisher()
     }
 

Sources/MQTTNIO/Utilities/String+Extensions.swift

@@ -0,0 +1,23 @@
+extension String {
+    
+    /// Indicates whether this string matches a given topic filter.
+    ///
+    /// This function takes into account wildcards (+ and #) in the topic filter when determining if the string matches.
+    /// - Parameter topicFilter: The topic filter to match this string with.
+    /// - Returns: `true` if the string matches the topic filter, `false` otherwise.
+    public func matchesMqttTopicFilter(_ topicFilter: String) -> Bool {
+        var filterParts = topicFilter.split(separator: "/", omittingEmptySubsequences: false)
+        var topicParts = split(separator: "/", omittingEmptySubsequences: false)
+        
+        while !filterParts.isEmpty && !topicParts.isEmpty {
+            guard filterParts[0] == topicParts[0] || filterParts[0] == "+" else {
+                return filterParts.count == 1 && filterParts[0] == "#"
+            }
+            
+            filterParts.removeFirst()
+            topicParts.removeFirst()
+        }
+        
+        return filterParts.isEmpty && topicParts.isEmpty
+    }
+}

Tests/MQTTNIOTests/Tests/TopicFilterMatchTests.swift

@@ -0,0 +1,26 @@
+@testable import MQTTNIO
+import XCTest
+
+final class TopicFilterMatchTests: XCTestCase {
+    func testTopicFilterMatches() {
+        XCTAssertTrue("one/two/three".matchesMqttTopicFilter("one/two/three"))
+        XCTAssertTrue("one/two/three".matchesMqttTopicFilter("one/+/three"))
+        XCTAssertTrue("one/two/three".matchesMqttTopicFilter("one/#"))
+        
+        XCTAssertFalse("/one/two/three".matchesMqttTopicFilter("one/two/three"))
+        XCTAssertTrue("/one/two/three".matchesMqttTopicFilter("/one/two/three"))
+        
+        XCTAssertTrue("one/two/three/four/five/six".matchesMqttTopicFilter("one/two/#"))
+        XCTAssertTrue("one/two/three/four/five/six".matchesMqttTopicFilter("one/+/three/#"))
+        
+        XCTAssertTrue("one/two/three/four/five/six".matchesMqttTopicFilter("one/+/three/#"))
+        XCTAssertTrue("one/two/three/four/five/six".matchesMqttTopicFilter("one/+/three/#"))
+        XCTAssertTrue("one/two/three/four/five/six".matchesMqttTopicFilter("one/two/+/four/+/six"))
+        
+        XCTAssertFalse("one/two/three".matchesMqttTopicFilter("one/two"))
+        XCTAssertFalse("one/two/three".matchesMqttTopicFilter("one/+"))
+        XCTAssertFalse("one/two/three".matchesMqttTopicFilter("one/two/three/four"))
+        XCTAssertFalse("one/two/three".matchesMqttTopicFilter("one/two/three/#"))
+        XCTAssertFalse("one/two/three".matchesMqttTopicFilter("one/two/three/+"))
+    }
+}