espat: add built-in support for MQTT publish using the Paho library packets, alongside some modifications needed for the AT protocol.

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

espat/mqtt/mqtt.go

@@ -0,0 +1,186 @@
+package mqtt
+
+import (
+	"errors"
+	"strings"
+	"time"
+
+	"github.com/eclipse/paho.mqtt.golang/packets"
+	"tinygo.org/x/drivers/espat"
+)
+
+// NewClient will create an MQTT v3.1.1 client with all of the options specified
+// in the provided ClientOptions. The client must have the Connect method called
+// on it before it may be used. This is to make sure resources (such as a net
+// connection) are created before the application is actually ready.
+func NewClient(o *ClientOptions) Client {
+	c := &mqttclient{opts: o, adaptor: o.Adaptor}
+	return c
+}
+
+type mqttclient struct {
+	adaptor   *espat.Device
+	conn      espat.Conn
+	connected bool
+	opts      *ClientOptions
+	mid       uint16
+}
+
+// AddRoute allows you to add a handler for messages on a specific topic
+// without making a subscription. For example having a different handler
+// for parts of a wildcard subscription
+func (c *mqttclient) AddRoute(topic string, callback MessageHandler) {
+	return
+}
+
+// IsConnected returns a bool signifying whether
+// the client is connected or not.
+func (c *mqttclient) IsConnected() bool {
+	return c.connected
+}
+
+// IsConnectionOpen return a bool signifying whether the client has an active
+// connection to mqtt broker, i.e not in disconnected or reconnect mode
+func (c *mqttclient) IsConnectionOpen() bool {
+	return c.connected
+}
+
+// Connect will create a connection to the message broker.
+func (c *mqttclient) Connect() Token {
+	var err error
+
+	// make connection
+	if strings.Contains(c.opts.Servers, "ssl://") {
+		url := strings.TrimPrefix(c.opts.Servers, "ssl://")
+		c.conn, err = c.adaptor.DialTLS("tcp", url, nil)
+		if err != nil {
+			return &mqtttoken{err: err}
+		}
+	} else if strings.Contains(c.opts.Servers, "tcp://") {
+		url := strings.TrimPrefix(c.opts.Servers, "tcp://")
+		c.conn, err = c.adaptor.Dial("tcp", url)
+		if err != nil {
+			return &mqtttoken{err: err}
+		}
+	} else {
+		// invalid protocol
+		return &mqtttoken{err: errors.New("invalid protocol")}
+	}
+
+	// send the MQTT connect message
+	connectPkt := packets.NewControlPacket(packets.Connect).(*packets.ConnectPacket)
+	connectPkt.Qos = 0
+	if c.opts.Username != "" {
+		connectPkt.Username = c.opts.Username
+		connectPkt.UsernameFlag = true
+	}
+
+	if c.opts.Password != "" {
+		connectPkt.Password = []byte(c.opts.Password)
+		connectPkt.PasswordFlag = true
+	}
+
+	connectPkt.ClientIdentifier = c.opts.ClientID //"tinygo-client-" + randomString(10)
+	connectPkt.ProtocolVersion = byte(c.opts.ProtocolVersion)
+	connectPkt.ProtocolName = "MQTT"
+	connectPkt.Keepalive = 30
+
+	err = connectPkt.Write(c.conn)
+	if err != nil {
+		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
+				return &mqtttoken{err: errors.New(packet.String())}
+			}
+		}
+
+		time.Sleep(100 * time.Millisecond)
+	}
+
+	c.connected = true
+	return &mqtttoken{}
+}
+
+// Disconnect will end the connection with the server, but not before waiting
+// the specified number of milliseconds to wait for existing work to be
+// completed.
+func (c *mqttclient) Disconnect(quiesce uint) {
+	c.conn.Close()
+	return
+}
+
+// Publish will publish a message with the specified QoS and content
+// 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 {
+	pub := packets.NewControlPacket(packets.Publish).(*packets.PublishPacket)
+	pub.Qos = qos
+	pub.TopicName = topic
+	switch payload.(type) {
+	case string:
+		pub.Payload = []byte(payload.(string))
+	case []byte:
+		pub.Payload = payload.([]byte)
+	default:
+		return &mqtttoken{err: errors.New("Unknown payload type")}
+	}
+	pub.MessageID = c.mid
+	c.mid++
+
+	err := pub.Write(c.conn)
+	return &mqtttoken{err: err}
+}
+
+// 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 {
+	return &mqtttoken{}
+}
+
+// SubscribeMultiple starts a new subscription for multiple topics. Provide a MessageHandler to
+// be executed when a message is published on one of the topics provided.
+func (c *mqttclient) SubscribeMultiple(filters map[string]byte, callback MessageHandler) Token {
+	return &mqtttoken{}
+}
+
+// Unsubscribe will end the subscription from each of the topics provided.
+// Messages published to those topics from other clients will no longer be
+// received.
+func (c *mqttclient) Unsubscribe(topics ...string) Token {
+	return &mqtttoken{}
+}
+
+// OptionsReader returns a ClientOptionsReader which is a copy of the clientoptions
+// in use by the client.
+func (c *mqttclient) OptionsReader() ClientOptionsReader {
+	r := ClientOptionsReader{}
+	return r
+}
+
+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
+}

espat/mqtt/paho.go

@@ -0,0 +1,245 @@
+// 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
+ */
+
+// Portions copyright © 2018 TIBCO Software Inc.
+
+package mqtt
+
+import (
+	"strings"
+	"time"
+
+	"tinygo.org/x/drivers/espat"
+)
+
+const (
+	disconnected uint32 = iota
+	connecting
+	reconnecting
+	connected
+)
+
+// Client is the interface definition for a Client as used by this
+// library, the interface is primarily to allow mocking tests.
+//
+// It is an MQTT v3.1.1 client for communicating
+// with an MQTT server using non-blocking methods that allow work
+// to be done in the background.
+// An application may connect to an MQTT server using:
+//   A plain TCP socket
+//   A secure SSL/TLS socket
+//   A websocket
+// To enable ensured message delivery at Quality of Service (QoS) levels
+// described in the MQTT spec, a message persistence mechanism must be
+// used. This is done by providing a type which implements the Store
+// interface. For convenience, FileStore and MemoryStore are provided
+// implementations that should be sufficient for most use cases. More
+// information can be found in their respective documentation.
+// Numerous connection options may be specified by configuring a
+// and then supplying a ClientOptions type.
+type Client interface {
+	// IsConnected returns a bool signifying whether
+	// the client is connected or not.
+	IsConnected() bool
+	// IsConnectionOpen return a bool signifying wether the client has an active
+	// connection to mqtt broker, i.e not in disconnected or reconnect mode
+	IsConnectionOpen() bool
+	// Connect will create a connection to the message broker, by default
+	// it will attempt to connect at v3.1.1 and auto retry at v3.1 if that
+	// fails
+	Connect() Token
+	// Disconnect will end the connection with the server, but not before waiting
+	// the specified number of milliseconds to wait for existing work to be
+	// completed.
+	Disconnect(quiesce uint)
+	// Publish will publish a message with the specified QoS and content
+	// to the specified topic.
+	// Returns a token to track delivery of the message to the broker
+	Publish(topic string, qos byte, retained bool, payload interface{}) Token
+	// Subscribe starts a new subscription. Provide a MessageHandler to be executed when
+	// a message is published on the topic provided, or nil for the default handler
+	Subscribe(topic string, qos byte, callback MessageHandler) Token
+	// SubscribeMultiple starts a new subscription for multiple topics. Provide a MessageHandler to
+	// be executed when a message is published on one of the topics provided, or nil for the
+	// default handler
+	SubscribeMultiple(filters map[string]byte, callback MessageHandler) Token
+	// Unsubscribe will end the subscription from each of the topics provided.
+	// Messages published to those topics from other clients will no longer be
+	// received.
+	Unsubscribe(topics ...string) Token
+	// AddRoute allows you to add a handler for messages on a specific topic
+	// without making a subscription. For example having a different handler
+	// for parts of a wildcard subscription
+	AddRoute(topic string, callback MessageHandler)
+	// OptionsReader returns a ClientOptionsReader which is a copy of the clientoptions
+	// in use by the client.
+	OptionsReader() ClientOptionsReader
+}
+
+// Token defines the interface for the tokens used to indicate when
+// actions have completed.
+type Token interface {
+	Wait() bool
+	WaitTimeout(time.Duration) bool
+	Error() error
+}
+
+// MessageHandler is a callback type which can be set to be
+// executed upon the arrival of messages published to topics
+// to which the client is subscribed.
+type MessageHandler func(Client, Message)
+
+// Message defines the externals that a message implementation must support
+// these are received messages that are passed to the callbacks, not internal
+// messages
+type Message interface {
+	Duplicate() bool
+	Qos() byte
+	Retained() bool
+	Topic() string
+	MessageID() uint16
+	Payload() []byte
+	Ack()
+}
+
+type message struct {
+	duplicate bool
+	qos       byte
+	retained  bool
+	topic     string
+	messageID uint16
+	payload   []byte
+	ack       func()
+}
+
+func (m *message) Duplicate() bool {
+	return m.duplicate
+}
+
+func (m *message) Qos() byte {
+	return m.qos
+}
+
+func (m *message) Retained() bool {
+	return m.retained
+}
+
+func (m *message) Topic() string {
+	return m.topic
+}
+
+func (m *message) MessageID() uint16 {
+	return m.messageID
+}
+
+func (m *message) Payload() []byte {
+	return m.payload
+}
+
+func (m *message) Ack() {
+	return
+}
+
+// ClientOptionsReader provides an interface for reading ClientOptions after the client has been initialized.
+type ClientOptionsReader struct {
+	options *ClientOptions
+}
+
+// ClientOptions contains configurable options for an MQTT Client.
+type ClientOptions struct {
+	Adaptor *espat.Device
+
+	//Servers                 []*url.URL
+	Servers  string
+	ClientID string
+	Username string
+	Password string
+	//CredentialsProvider     CredentialsProvider
+	CleanSession            bool
+	Order                   bool
+	WillEnabled             bool
+	WillTopic               string
+	WillPayload             []byte
+	WillQos                 byte
+	WillRetained            bool
+	ProtocolVersion         uint
+	protocolVersionExplicit bool
+	//TLSConfig               *tls.Config
+	KeepAlive            int64
+	PingTimeout          time.Duration
+	ConnectTimeout       time.Duration
+	MaxReconnectInterval time.Duration
+	AutoReconnect        bool
+	//Store                   Store
+	//DefaultPublishHandler   MessageHandler
+	//OnConnect               OnConnectHandler
+	//OnConnectionLost        ConnectionLostHandler
+	WriteTimeout        time.Duration
+	MessageChannelDepth uint
+	ResumeSubs          bool
+	//HTTPHeaders             http.Header
+}
+
+// NewClientOptions returns a new ClientOptions struct.
+func NewClientOptions(adaptor *espat.Device) *ClientOptions {
+	return &ClientOptions{Adaptor: adaptor, ProtocolVersion: 4}
+}
+
+// AddBroker adds a broker URI to the list of brokers to be used. 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.
+//
+// Default values for hostname is "127.0.0.1", for schema is "tcp://".
+//
+// An example broker URI would look like: tcp://foobar.com:1883
+func (o *ClientOptions) AddBroker(server string) *ClientOptions {
+	if len(server) > 0 && server[0] == ':' {
+		server = "127.0.0.1" + server
+	}
+	if !strings.Contains(server, "://") {
+		server = "tcp://" + server
+	}
+
+	o.Servers = server
+	return o
+}
+
+// SetClientID will set the client id to be used by this client when
+// connecting to the MQTT broker. According to the MQTT v3.1 specification,
+// a client id mus be no longer than 23 characters.
+func (o *ClientOptions) SetClientID(id string) *ClientOptions {
+	o.ClientID = id
+	return o
+}
+
+// SetUsername will set the username to be used by this client when connecting
+// to the MQTT broker. Note: without the use of SSL/TLS, this information will
+// be sent in plaintext accross the wire.
+func (o *ClientOptions) SetUsername(u string) *ClientOptions {
+	o.Username = u
+	return o
+}
+
+// SetPassword will set the password to be used by this client when connecting
+// to the MQTT broker. Note: without the use of SSL/TLS, this information will
+// be sent in plaintext accross the wire.
+func (o *ClientOptions) SetPassword(p string) *ClientOptions {
+	o.Password = p
+	return o
+}

examples/espat/mqttclient/main.go

@@ -7,7 +7,7 @@
 //
 // You must install the Paho MQTT package to build this program:
 //
-// 		go get github.com/eclipse/paho.mqtt.golang
+// 		go get -u github.com/eclipse/paho.mqtt.golang
 //
 package main
 
@@ -16,18 +16,17 @@ import (
 	"math/rand"
 	"time"
 
-	"github.com/eclipse/paho.mqtt.golang/packets"
 	"tinygo.org/x/drivers/espat"
+	"tinygo.org/x/drivers/espat/mqtt"
 )
 
 // access point info
 const ssid = "YOURSSID"
 const pass = "YOURPASS"
-const useSSL = true
 
 // IP address of the MQTT broker to use. Replace with your own info.
-//const server = "test.mosquitto.org:1883"
+//const server = "tcp://test.mosquitto.org:1883"
-const server = "test.mosquitto.org:8883"
+const server = "ssl://test.mosquitto.org:8883"
 
 // change these to connect to a different UART or pins for the ESP8266/ESP32
 var (
@@ -38,9 +37,6 @@ var (
 	console = machine.UART0
 
 	adaptor *espat.Device
-	conn    espat.Conn
-	err     error
-	mid     uint16
 	topic   = "tinygo"
 )
 
@@ -62,40 +58,34 @@ func main() {
 		connectToAP()
 	} else {
 		println("")
-		println("Unable to connect to wifi adaptor.")
+		failMessage("Unable to connect to wifi adaptor.")
 		return
 	}
 
-	// make connection
+	opts := mqtt.NewClientOptions(adaptor)
-	if useSSL {
+	opts.AddBroker(server).SetClientID("tinygo-client-" + randomString(10))
-		println("Dialing SSL connection...")
+
-		conn, err = adaptor.DialTLS("tcp", server, nil)
+	println("Connectng to MQTT...")
-		if err != nil {
+	cl := mqtt.NewClient(opts)
-			println("SSL connect error")
+	if token := cl.Connect(); token.Wait() && token.Error() != nil {
-			println(err)
+		failMessage(token.Error().Error())
-			return
-		}
-	} else {
-		println("Dialing TCP connection...")
-		conn, err = adaptor.Dial("tcp", server)
-		if err != nil {
-			println("TCP connect error")
-			println(err)
-			return
-		}
 	}
 
-	err = connectToMQTTServer()
-
 	for {
-		publishToMQTT()
+		println("Publishing MQTT message...")
+		data := []byte("{\"e\":[{ \"n\":\"hello\", \"v\":101 }]}")
+		token := cl.Publish(topic, 0, false, data)
+		token.Wait()
+		if token.Error() != nil {
+			println(token.Error().Error())
+		}
 
 		time.Sleep(1000 * time.Millisecond)
 	}
 
 	// Right now this code is never reached. Need a way to trigger it...
-	println("Disconnecting TCP...")
+	println("Disconnecting MQTT...")
-	conn.Close()
+	cl.Disconnect(100)
 
 	println("Done.")
 }
@@ -123,56 +113,6 @@ func connectToAP() {
 	println(adaptor.GetClientIP())
 }
 
-func connectToMQTTServer() error {
-	// send the MQTT connect message
-	connectPkt := packets.NewControlPacket(packets.Connect).(*packets.ConnectPacket)
-	connectPkt.Qos = 0
-	// connectPkt.Username = "tinygo"
-	// connectPkt.Password = []byte("1234")
-	connectPkt.ClientIdentifier = "tinygo-client-" + randomString(10)
-	connectPkt.ProtocolVersion = 4
-	connectPkt.ProtocolName = "MQTT"
-	connectPkt.Keepalive = 30
-
-	println("Sending MQTT connect...")
-	err := connectPkt.Write(conn)
-	if err != nil {
-		println("mqtt connect error")
-		println(err.Error())
-		return err
-	}
-
-	println("Waiting for MQTT connect...")
-	// TODO: handle timeout
-	for {
-		packet, _ := packets.ReadPacket(conn)
-
-		if packet != nil {
-			_, ok := packet.(*packets.ConnackPacket)
-			if ok {
-				println("Connected to MQTT server.")
-				println(packet.String())
-				return nil
-			}
-		}
-
-		time.Sleep(100 * time.Millisecond)
-	}
-}
-
-func publishToMQTT() error {
-	println("Publishing MQTT message...")
-
-	publish := packets.NewControlPacket(packets.Publish).(*packets.PublishPacket)
-	publish.Qos = 0
-	publish.TopicName = topic
-	publish.Payload = []byte("Hello, mqtt\r\n")
-	publish.MessageID = mid
-	mid++
-
-	return publish.Write(conn)
-}
-
 // Returns an int >= min, < max
 func randomInt(min, max int) int {
 	return min + rand.Intn(max-min)
@@ -186,3 +126,10 @@ func randomString(len int) string {
 	}
 	return string(bytes)
 }
+
+func failMessage(msg string) {
+	for {
+		println(msg)
+		time.Sleep(1 * time.Second)
+	}
+}