From 7e78e2c998e7a6647b619b27c44bcf08081da2d1 Mon Sep 17 00:00:00 2001 From: Ron Evans Date: Tue, 2 Jul 2019 15:57:46 +0200 Subject: [PATCH] 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 --- espat/mqtt/mqtt.go | 186 +++++++++++++++++++++++ espat/mqtt/paho.go | 245 ++++++++++++++++++++++++++++++ examples/espat/mqttclient/main.go | 109 ++++--------- 3 files changed, 459 insertions(+), 81 deletions(-) create mode 100644 espat/mqtt/mqtt.go create mode 100644 espat/mqtt/paho.go diff --git a/espat/mqtt/mqtt.go b/espat/mqtt/mqtt.go new file mode 100644 index 0000000..bc16bf3 --- /dev/null +++ b/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 +} diff --git a/espat/mqtt/paho.go b/espat/mqtt/paho.go new file mode 100644 index 0000000..7f0a99b --- /dev/null +++ b/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 +} diff --git a/examples/espat/mqttclient/main.go b/examples/espat/mqttclient/main.go index 13cecd0..a990100 100644 --- a/examples/espat/mqttclient/main.go +++ b/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 = "test.mosquitto.org:8883" +//const server = "tcp://test.mosquitto.org:1883" +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 - if useSSL { - println("Dialing SSL connection...") - conn, err = adaptor.DialTLS("tcp", server, nil) - if err != nil { - println("SSL connect error") - println(err) - return - } - } else { - println("Dialing TCP connection...") - conn, err = adaptor.Dial("tcp", server) - if err != nil { - println("TCP connect error") - println(err) - return - } + opts := mqtt.NewClientOptions(adaptor) + opts.AddBroker(server).SetClientID("tinygo-client-" + randomString(10)) + + println("Connectng to MQTT...") + cl := mqtt.NewClient(opts) + if token := cl.Connect(); token.Wait() && token.Error() != nil { + failMessage(token.Error().Error()) } - 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...") - conn.Close() + println("Disconnecting MQTT...") + 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) + } +}