espat: implement MQTT subscribe functionality via blocking select/channels.

also refactor response processing for greater speed and efficiency.

Signed-off-by: Ron Evans <ron@hybridgroup.com>

espat/mqtt/mqtt.go

@@ -19,15 +19,21 @@ import (
 // connection) are created before the application is actually ready.
 func NewClient(o *ClientOptions) Client {
 	c := &mqttclient{opts: o, adaptor: o.Adaptor}
+	c.msgRouter, c.stopRouter = newRouter()
 	return c
 }
 
 type mqttclient struct {
-	adaptor   *espat.Device
-	conn      net.Conn
-	connected bool
-	opts      *ClientOptions
-	mid       uint16
+	adaptor         *espat.Device
+	conn            net.Conn
+	connected       bool
+	opts            *ClientOptions
+	mid             uint16
+	inbound         chan packets.ControlPacket
+	stop            chan struct{}
+	msgRouter       *router
+	stopRouter      chan bool
+	incomingPubChan chan *packets.PublishPacket
 }
 
 // AddRoute allows you to add a handler for messages on a specific topic
@@ -71,6 +77,12 @@ func (c *mqttclient) Connect() Token {
 		return &mqtttoken{err: errors.New("invalid protocol")}
 	}
 
+	c.mid = 1
+	c.inbound = make(chan packets.ControlPacket)
+	c.stop = make(chan struct{})
+	c.incomingPubChan = make(chan *packets.PublishPacket)
+	c.msgRouter.matchAndDispatch(c.incomingPubChan, c.opts.Order, c)
+
 	// send the MQTT connect message
 	connectPkt := packets.NewControlPacket(packets.Connect).(*packets.ConnectPacket)
 	connectPkt.Qos = 0
@@ -84,7 +96,7 @@ func (c *mqttclient) Connect() Token {
 		connectPkt.PasswordFlag = true
 	}
 
-	connectPkt.ClientIdentifier = c.opts.ClientID //"tinygo-client-" + randomString(10)
+	connectPkt.ClientIdentifier = c.opts.ClientID
 	connectPkt.ProtocolVersion = byte(c.opts.ProtocolVersion)
 	connectPkt.ProtocolName = "MQTT"
 	connectPkt.Keepalive = 30
@@ -94,26 +106,25 @@ func (c *mqttclient) Connect() Token {
 		return &mqtttoken{err: err}
 	}
 
-	// TODO: handle timeout
-	for {
-		packet, _ := packets.ReadPacket(c.conn)
-
-		if packet != nil {
-			ack, ok := packet.(*packets.ConnackPacket)
-			if ok {
-				if ack.ReturnCode == 0 {
-					// success
-					return &mqtttoken{}
-				}
-				// otherwise something went wrong
+	// TODO: handle timeout as ReadPacket blocks until it gets a packet.
+	// CONNECT response.
+	packet, err := packets.ReadPacket(c.conn)
+	if err != nil {
+		return &mqtttoken{err: err}
+	}
+	if packet != nil {
+		ack, ok := packet.(*packets.ConnackPacket)
+		if ok {
+			if ack.ReturnCode != 0 {
 				return &mqtttoken{err: errors.New(packet.String())}
 			}
+			c.connected = true
 		}
-
-		time.Sleep(100 * time.Millisecond)
 	}
 
-	c.connected = true
+	go readMessages(c)
+	go processInbound(c)
+
 	return &mqtttoken{}
 }
 
@@ -129,6 +140,10 @@ func (c *mqttclient) Disconnect(quiesce uint) {
 // to the specified topic.
 // Returns a token to track delivery of the message to the broker
 func (c *mqttclient) Publish(topic string, qos byte, retained bool, payload interface{}) Token {
+	if !c.IsConnected() {
+		return &mqtttoken{err: errors.New("MQTT client not connected")}
+	}
+
 	pub := packets.NewControlPacket(packets.Publish).(*packets.PublishPacket)
 	pub.Qos = qos
 	pub.TopicName = topic
@@ -144,12 +159,37 @@ func (c *mqttclient) Publish(topic string, qos byte, retained bool, payload inte
 	c.mid++
 
 	err := pub.Write(c.conn)
-	return &mqtttoken{err: err}
+	if err != nil {
+		return &mqtttoken{err: err}
+	}
+
+	return &mqtttoken{}
 }
 
 // Subscribe starts a new subscription. Provide a MessageHandler to be executed when
 // a message is published on the topic provided.
 func (c *mqttclient) Subscribe(topic string, qos byte, callback MessageHandler) Token {
+	if !c.IsConnected() {
+		return &mqtttoken{err: errors.New("MQTT client not connected")}
+	}
+
+	sub := packets.NewControlPacket(packets.Subscribe).(*packets.SubscribePacket)
+	sub.Topics = append(sub.Topics, topic)
+	sub.Qoss = append(sub.Qoss, qos)
+
+	if callback != nil {
+		c.msgRouter.addRoute(topic, callback)
+	}
+
+	sub.MessageID = c.mid
+	c.mid++
+
+	// drop in the channel to send
+	err := sub.Write(c.conn)
+	if err != nil {
+		return &mqtttoken{err: err}
+	}
+
 	return &mqtttoken{}
 }
 
@@ -173,18 +213,92 @@ func (c *mqttclient) OptionsReader() ClientOptionsReader {
 	return r
 }
 
-type mqtttoken struct {
-	err error
+func processInbound(c *mqttclient) {
+	for {
+		select {
+		case msg := <-c.inbound:
+			switch m := msg.(type) {
+			case *packets.PingrespPacket:
+				// TODO: handle this
+			case *packets.SubackPacket:
+				// TODO: handle this
+			case *packets.UnsubackPacket:
+				// TODO: handle this
+			case *packets.PublishPacket:
+				// TODO: handle Qos
+				c.incomingPubChan <- m
+			case *packets.PubackPacket:
+				// TODO: handle this
+			case *packets.PubrecPacket:
+				// TODO: handle this
+			case *packets.PubrelPacket:
+				// TODO: handle this
+			case *packets.PubcompPacket:
+				// TODO: handle this
+			}
+		case <-c.stop:
+			return
+		}
+	}
 }
 
-func (t *mqtttoken) Wait() bool {
-	return true
+// readMessages reads incoming messages off the wire.
+// incoming messages are then send into inbound channel.
+func readMessages(c *mqttclient) {
+	var err error
+	var cp packets.ControlPacket
+
+PROCESS:
+	for {
+		if cp, err = c.ReadPacket(); err != nil {
+			break PROCESS
+		}
+		if cp != nil {
+			c.inbound <- cp
+			// TODO: Notify keepalive logic that we recently received a packet
+		}
+
+		time.Sleep(100 * time.Millisecond)
+	}
+
+	// TODO: handle if we received an error on read.
+	// If disconnect is in progress, swallow error and return
 }
 
-func (t *mqtttoken) WaitTimeout(time.Duration) bool {
-	return true
+func (c *mqttclient) ackFunc(packet *packets.PublishPacket) func() {
+	return func() {
+		switch packet.Qos {
+		case 2:
+			// pr := packets.NewControlPacket(packets.Pubrec).(*packets.PubrecPacket)
+			// pr.MessageID = packet.MessageID
+			// DEBUG.Println(NET, "putting pubrec msg on obound")
+			// select {
+			// case c.oboundP <- &PacketAndToken{p: pr, t: nil}:
+			// case <-c.stop:
+			// }
+			// DEBUG.Println(NET, "done putting pubrec msg on obound")
+		case 1:
+			// pa := packets.NewControlPacket(packets.Puback).(*packets.PubackPacket)
+			// pa.MessageID = packet.MessageID
+			// DEBUG.Println(NET, "putting puback msg on obound")
+			// persistOutbound(c.persist, pa)
+			// select {
+			// case c.oboundP <- &PacketAndToken{p: pa, t: nil}:
+			// case <-c.stop:
+			// }
+			// DEBUG.Println(NET, "done putting puback msg on obound")
+		case 0:
+			// do nothing, since there is no need to send an ack packet back
+		}
+	}
 }
 
-func (t *mqtttoken) Error() error {
-	return t.err
+// ReadPacket tries to read the next incoming packet from the MQTT broker.
+// If there is no data yet but also is no error, it returns nil for both values.
+func (c *mqttclient) ReadPacket() (packets.ControlPacket, error) {
+	// check for data first...
+	if espat.ActiveDevice.IsSocketDataAvailable() {
+		return packets.ReadPacket(c.conn)
+	}
+	return nil, nil
 }

espat/mqtt/paho.go

@@ -24,6 +24,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/eclipse/paho.mqtt.golang/packets"
 	"tinygo.org/x/drivers/espat"
 )
 
@@ -155,6 +156,18 @@ func (m *message) Ack() {
 	return
 }
 
+func messageFromPublish(p *packets.PublishPacket, ack func()) Message {
+	return &message{
+		duplicate: p.Dup,
+		qos:       p.Qos,
+		retained:  p.Retain,
+		topic:     p.TopicName,
+		messageID: p.MessageID,
+		payload:   p.Payload,
+		ack:       ack,
+	}
+}
+
 // ClientOptionsReader provides an interface for reading ClientOptions after the client has been initialized.
 type ClientOptionsReader struct {
 	options *ClientOptions

espat/mqtt/router.go

@@ -0,0 +1,182 @@
+// The following code is a slightly modified version of code taken from the Paho MQTT library.
+// It is here until TinyGo can compile the "net" package from the standard library, at which time
+// it can be removed.
+
+/*
+ * Copyright (c) 2013 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ *    Seth Hoenig
+ *    Allan Stockdill-Mander
+ *    Mike Robertson
+ */
+
+package mqtt
+
+import (
+	"container/list"
+	"strings"
+
+	"github.com/eclipse/paho.mqtt.golang/packets"
+)
+
+// route is a type which associates MQTT Topic strings with a
+// callback to be executed upon the arrival of a message associated
+// with a subscription to that topic.
+type route struct {
+	topic    string
+	callback MessageHandler
+}
+
+// match takes a slice of strings which represent the route being tested having been split on '/'
+// separators, and a slice of strings representing the topic string in the published message, similarly
+// split.
+// The function determines if the topic string matches the route according to the MQTT topic rules
+// and returns a boolean of the outcome
+func match(route []string, topic []string) bool {
+	if len(route) == 0 {
+		if len(topic) == 0 {
+			return true
+		}
+		return false
+	}
+
+	if len(topic) == 0 {
+		if route[0] == "#" {
+			return true
+		}
+		return false
+	}
+
+	if route[0] == "#" {
+		return true
+	}
+
+	if (route[0] == "+") || (route[0] == topic[0]) {
+		return match(route[1:], topic[1:])
+	}
+	return false
+}
+
+func routeIncludesTopic(route, topic string) bool {
+	return match(routeSplit(route), strings.Split(topic, "/"))
+}
+
+// removes $share and sharename when splitting the route to allow
+// shared subscription routes to correctly match the topic
+func routeSplit(route string) []string {
+	var result []string
+	if strings.HasPrefix(route, "$share") {
+		result = strings.Split(route, "/")[2:]
+	} else {
+		result = strings.Split(route, "/")
+	}
+	return result
+}
+
+// match takes the topic string of the published message and does a basic compare to the
+// string of the current Route, if they match it returns true
+func (r *route) match(topic string) bool {
+	return r.topic == topic || routeIncludesTopic(r.topic, topic)
+}
+
+type router struct {
+	//sync.RWMutex
+	routes         *list.List
+	defaultHandler MessageHandler
+	messages       chan *packets.PublishPacket
+	stop           chan bool
+}
+
+// newRouter returns a new instance of a Router and channel which can be used to tell the Router
+// to stop
+func newRouter() (*router, chan bool) {
+	router := &router{routes: list.New(), messages: make(chan *packets.PublishPacket), stop: make(chan bool)}
+	stop := router.stop
+	return router, stop
+}
+
+// addRoute takes a topic string and MessageHandler callback. It looks in the current list of
+// routes to see if there is already a matching Route. If there is it replaces the current
+// callback with the new one. If not it add a new entry to the list of Routes.
+func (r *router) addRoute(topic string, callback MessageHandler) {
+	for e := r.routes.Front(); e != nil; e = e.Next() {
+		if e.Value.(*route).match(topic) {
+			r := e.Value.(*route)
+			r.callback = callback
+			return
+		}
+	}
+	r.routes.PushBack(&route{topic: topic, callback: callback})
+}
+
+// deleteRoute takes a route string, looks for a matching Route in the list of Routes. If
+// found it removes the Route from the list.
+func (r *router) deleteRoute(topic string) {
+	for e := r.routes.Front(); e != nil; e = e.Next() {
+		if e.Value.(*route).match(topic) {
+			r.routes.Remove(e)
+			return
+		}
+	}
+}
+
+// setDefaultHandler assigns a default callback that will be called if no matching Route
+// is found for an incoming Publish.
+func (r *router) setDefaultHandler(handler MessageHandler) {
+	r.defaultHandler = handler
+}
+
+// matchAndDispatch takes a channel of Message pointers as input and starts a go routine that
+// takes messages off the channel, matches them against the internal route list and calls the
+// associated callback (or the defaultHandler, if one exists and no other route matched). If
+// anything is sent down the stop channel the function will end.
+func (r *router) matchAndDispatch(messages <-chan *packets.PublishPacket, order bool, client *mqttclient) {
+	go func() {
+		for {
+			select {
+			case message := <-messages:
+				sent := false
+				m := messageFromPublish(message, client.ackFunc(message))
+				handlers := []MessageHandler{}
+				for e := r.routes.Front(); e != nil; e = e.Next() {
+					if e.Value.(*route).match(message.TopicName) {
+						if order {
+							handlers = append(handlers, e.Value.(*route).callback)
+						} else {
+							hd := e.Value.(*route).callback
+							go func() {
+								hd(client, m)
+								//TODO: m.Ack()
+							}()
+						}
+						sent = true
+					}
+				}
+				if !sent && r.defaultHandler != nil {
+					if order {
+						handlers = append(handlers, r.defaultHandler)
+					} else {
+						go func() {
+							r.defaultHandler(client, m)
+							//TODO: m.Ack()
+						}()
+					}
+				}
+				for _, handler := range handlers {
+					func() {
+						handler(client, m)
+						//TODO: m.Ack()
+					}()
+				}
+			case <-r.stop:
+				return
+			}
+		}
+	}()
+}

espat/mqtt/token.go

@@ -0,0 +1,19 @@
+package mqtt
+
+import "time"
+
+type mqtttoken struct {
+	err error
+}
+
+func (t *mqtttoken) Wait() bool {
+	return true
+}
+
+func (t *mqtttoken) WaitTimeout(time.Duration) bool {
+	return true
+}
+
+func (t *mqtttoken) Error() error {
+	return t.err
+}