MQTT topic matcher (#32)

v2/emitter.go

@@ -66,7 +66,7 @@ func NewClient(options ...func(*Client)) *Client {
 		opts:     mqtt.NewClientOptions(),
 		timeout:  60 * time.Second,
 		store:    new(store),
-		handlers: newTrie(),
+		handlers: NewTrie(),
 	}
 
 	// Set handlers
@@ -301,8 +301,11 @@ func (c *Client) Subscribe(key string, channel string, optionalHandler MessageHa
 		c.handlers.AddHandler(channel, optionalHandler)
 	}
 
+	// https://github.com/eclipse/paho.mqtt.golang/blob/master/topic.go#L78
+	topic := strings.ReplaceAll(formatTopic(key, channel, options), "#/", "#")
+
 	// Issue subscribe
-	token := c.conn.Subscribe(formatTopic(key, channel, options), 0, nil)
+	token := c.conn.Subscribe(topic, 0, nil)
 	return c.do(token)
 }
 

v2/options.go

@@ -7,6 +7,15 @@ import (
 	"time"
 )
 
+// WithMatcher If "mqtt", then topic matching would follow MQTT specification.
+func WithMatcher(matcher string) func(*Client) {
+	return func(c *Client) {
+		if "mqtt" == matcher {
+			c.handlers = NewTrieMQTT()
+		}
+	}
+}
+
 // WithBrokers configures broker URIs to connect to. The format should be scheme://host:port
 // Where "scheme" is one of "tcp", "ssl", or "ws", "host" is the ip-address (or hostname)
 // and "port" is the port on which the broker is accepting connections.

v2/subtrie.go

@@ -46,9 +46,10 @@ func (n *node) orphan() {
 type trie struct {
 	sync.RWMutex
 	root *node // The root node of the tree.
+	lookup func(query []string, result *[]MessageHandler, node *node)
 }
 
-// newTrie creates a new matcher for the subscriptions.
+// newTrie creates a new trie without a lookup function.
 func newTrie() *trie {
 	return &trie{
 		root: &node{
@@ -57,6 +58,20 @@ func newTrie() *trie {
 	}
 }
 
+// NewTrie creates a new subscriptions matcher using standard emitter strategy.
+func NewTrie() *trie {
+	t := newTrie()
+	t.lookup = t.lookupEmitter
+	return t
+}
+
+// NewTrieMQTT creates a new subscriptions matcher using standard MQTT strategy.
+func NewTrieMQTT() *trie {
+	t := newTrie()
+	t.lookup = t.lookupMqtt
+	return t
+}
+
 // AddHandler adds a message handler to a topic.
 func (t *trie) AddHandler(topic string, handler MessageHandler) error {
 	query, rt := newRoute(topic, handler)
@@ -120,7 +135,7 @@ func (t *trie) Lookup(topic string) []MessageHandler {
 	return result
 }
 
-func (t *trie) lookup(query []string, result *[]MessageHandler, node *node) {
+func (t *trie) lookupEmitter(query []string, result *[]MessageHandler, node *node) {
 
 	// Add routes from the current branch
 	for _, route := range node.routes {
@@ -132,12 +147,41 @@ func (t *trie) lookup(query []string, result *[]MessageHandler, node *node) {
 
 		// Go through the exact match branch
 		if n, ok := node.children[query[0]]; ok {
-			t.lookup(query[1:], result, n)
+			t.lookupEmitter(query[1:], result, n)
 		}
 
 		// Go through wildcard match branch
 		if n, ok := node.children["+"]; ok {
-			t.lookup(query[1:], result, n)
+			t.lookupEmitter(query[1:], result, n)
+		}
+	}
+}
+
+func (t *trie) lookupMqtt(query []string, result *[]MessageHandler, node *node) {
+	if len(query) == 0 {
+		// Add routes from the current branch
+		for _, route := range node.routes {
+			*result = append(*result, route.Action)
+		}
+	}
+
+	// If we're not yet done, continue
+	if len(query) > 0 {
+		// Go through the exact match branch
+		if n, ok := node.children[query[0]]; ok {
+			t.lookupMqtt(query[1:], result, n)
+		}
+
+		// Go through wildcard match branch
+		if n, ok := node.children["+"]; ok {
+			t.lookupMqtt(query[1:], result, n)
+		}
+
+		// Go through wildcard match branch
+		if n, ok := node.children["#"]; ok {
+			for _, route := range n.routes {
+				*result = append(*result, route.Action)
+			}
 		}
 	}
 }

v2/subtrie_test.go

@@ -12,7 +12,7 @@ func TestRoute(t *testing.T) {
 }
 
 func TestTrieMatch(t *testing.T) {
-	m := newTrie()
+	m := NewTrie()
 	testPopulateWithStrings(m, []string{
 		"a/",
 		"a/b/c/",