Persist offline mqtt events in the queueDir and replay (#7037)

vendor/github.com/eclipse/paho.mqtt.golang/client.go

@@ -12,6 +12,8 @@
  *    Mike Robertson
  */
 
+// Portions copyright © 2018 TIBCO Software Inc.
+
 // Package mqtt provides an MQTT v3.1.1 client library.
 package mqtt
 
@@ -19,16 +21,16 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/eclipse/paho.mqtt.golang/packets"
 )
 
-type connStatus uint
-
 const (
-	disconnected connStatus = iota
+	disconnected uint32 = iota
 	connecting
 	reconnecting
 	connected
@@ -53,18 +55,50 @@ const (
 // 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
 }
 
 // client implements the Client interface
 type client struct {
+	lastSent        int64
+	lastReceived    int64
+	pingOutstanding int32
+	status          uint32
 	sync.RWMutex
 	messageIds
 	conn            net.Conn
@@ -78,10 +112,6 @@ type client struct {
 	stop            chan struct{}
 	persist         Store
 	options         ClientOptions
-	pingResp        chan struct{}
-	packetResp      chan struct{}
-	keepaliveReset  chan struct{}
-	status          connStatus
 	workers         sync.WaitGroup
 }
 
@@ -99,21 +129,26 @@ func NewClient(o *ClientOptions) Client {
 	switch c.options.ProtocolVersion {
 	case 3, 4:
 		c.options.protocolVersionExplicit = true
+	case 0x83, 0x84:
+		c.options.protocolVersionExplicit = true
 	default:
 		c.options.ProtocolVersion = 4
 		c.options.protocolVersionExplicit = false
 	}
 	c.persist = c.options.Store
 	c.status = disconnected
-	c.messageIds = messageIds{index: make(map[uint16]Token)}
+	c.messageIds = messageIds{index: make(map[uint16]tokenCompletor)}
 	c.msgRouter, c.stopRouter = newRouter()
-	c.msgRouter.setDefaultHandler(c.options.DefaultPublishHander)
+	c.msgRouter.setDefaultHandler(c.options.DefaultPublishHandler)
 	if !c.options.AutoReconnect {
 		c.options.MessageChannelDepth = 0
 	}
 	return c
 }
 
+// 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 *client) AddRoute(topic string, callback MessageHandler) {
 	if callback != nil {
 		c.msgRouter.addRoute(topic, callback)
@@ -125,54 +160,81 @@ func (c *client) AddRoute(topic string, callback MessageHandler) {
 func (c *client) IsConnected() bool {
 	c.RLock()
 	defer c.RUnlock()
+	status := atomic.LoadUint32(&c.status)
 	switch {
-	case c.status == connected:
+	case status == connected:
 		return true
-	case c.options.AutoReconnect && c.status > disconnected:
+	case c.options.AutoReconnect && status > connecting:
 		return true
 	default:
 		return false
 	}
 }
 
-func (c *client) connectionStatus() connStatus {
+// 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 *client) IsConnectionOpen() bool {
 	c.RLock()
 	defer c.RUnlock()
-	return c.status
+	status := atomic.LoadUint32(&c.status)
+	switch {
+	case status == connected:
+		return true
+	default:
+		return false
+	}
 }
 
-func (c *client) setConnected(status connStatus) {
+func (c *client) connectionStatus() uint32 {
+	c.RLock()
+	defer c.RUnlock()
+	status := atomic.LoadUint32(&c.status)
+	return status
+}
+
+func (c *client) setConnected(status uint32) {
 	c.Lock()
 	defer c.Unlock()
-	c.status = status
+	atomic.StoreUint32(&c.status, uint32(status))
 }
 
 //ErrNotConnected is the error returned from function calls that are
 //made when the client is not connected to a broker
 var ErrNotConnected = errors.New("Not Connected")
 
-// Connect will create a connection to the message broker
-// If clean session is false, then a slice will
-// be returned containing Receipts for all messages
-// that were in-flight at the last disconnect.
-// If clean session is true, then any existing client
-// state will be removed.
+// 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
 func (c *client) Connect() Token {
 	var err error
 	t := newToken(packets.Connect).(*ConnectToken)
 	DEBUG.Println(CLI, "Connect()")
 
+	c.obound = make(chan *PacketAndToken, c.options.MessageChannelDepth)
+	c.oboundP = make(chan *PacketAndToken, c.options.MessageChannelDepth)
+	c.ibound = make(chan packets.ControlPacket)
+
 	go func() {
 		c.persist.Open()
 
 		c.setConnected(connecting)
+		c.errors = make(chan error, 1)
+		c.stop = make(chan struct{})
+
 		var rc byte
-		cm := newConnectMsgFromOptions(&c.options)
+		protocolVersion := c.options.ProtocolVersion
+
+		if len(c.options.Servers) == 0 {
+			t.setError(fmt.Errorf("No servers defined to connect to"))
+			return
+		}
 
 		for _, broker := range c.options.Servers {
+			cm := newConnectMsgFromOptions(&c.options, broker)
+			c.options.ProtocolVersion = protocolVersion
 		CONN:
 			DEBUG.Println(CLI, "about to write new connect msg")
-			c.conn, err = openConnection(broker, &c.options.TLSConfig, c.options.ConnectTimeout)
+			c.conn, err = openConnection(broker, c.options.TLSConfig, c.options.ConnectTimeout, c.options.HTTPHeaders)
 			if err == nil {
 				DEBUG.Println(CLI, "socket connected to broker")
 				switch c.options.ProtocolVersion {
@@ -180,6 +242,14 @@ func (c *client) Connect() Token {
 					DEBUG.Println(CLI, "Using MQTT 3.1 protocol")
 					cm.ProtocolName = "MQIsdp"
 					cm.ProtocolVersion = 3
+				case 0x83:
+					DEBUG.Println(CLI, "Using MQTT 3.1b protocol")
+					cm.ProtocolName = "MQIsdp"
+					cm.ProtocolVersion = 0x83
+				case 0x84:
+					DEBUG.Println(CLI, "Using MQTT 3.1.1b protocol")
+					cm.ProtocolName = "MQTT"
+					cm.ProtocolVersion = 0x84
 				default:
 					DEBUG.Println(CLI, "Using MQTT 3.1.1 protocol")
 					c.options.ProtocolVersion = 4
@@ -188,7 +258,7 @@ func (c *client) Connect() Token {
 				}
 				cm.Write(c.conn)
 
-				rc = c.connect()
+				rc, t.sessionPresent = c.connect()
 				if rc != packets.Accepted {
 					if c.conn != nil {
 						c.conn.Close()
@@ -215,32 +285,27 @@ func (c *client) Connect() Token {
 
 		if c.conn == nil {
 			ERROR.Println(CLI, "Failed to connect to a broker")
-			t.returnCode = rc
-			if rc != packets.ErrNetworkError {
-				t.err = packets.ConnErrors[rc]
-			} else {
-				t.err = fmt.Errorf("%s : %s", packets.ConnErrors[rc], err)
-			}
 			c.setConnected(disconnected)
 			c.persist.Close()
-			t.flowComplete()
+			t.returnCode = rc
+			if rc != packets.ErrNetworkError {
+				t.setError(packets.ConnErrors[rc])
+			} else {
+				t.setError(fmt.Errorf("%s : %s", packets.ConnErrors[rc], err))
+			}
 			return
 		}
 
+		c.options.protocolVersionExplicit = true
+
 		if c.options.KeepAlive != 0 {
+			atomic.StoreInt32(&c.pingOutstanding, 0)
+			atomic.StoreInt64(&c.lastReceived, time.Now().Unix())
+			atomic.StoreInt64(&c.lastSent, time.Now().Unix())
 			c.workers.Add(1)
 			go keepalive(c)
 		}
 
-		c.obound = make(chan *PacketAndToken, c.options.MessageChannelDepth)
-		c.oboundP = make(chan *PacketAndToken, c.options.MessageChannelDepth)
-		c.ibound = make(chan packets.ControlPacket)
-		c.errors = make(chan error, 1)
-		c.stop = make(chan struct{})
-		c.pingResp = make(chan struct{}, 1)
-		c.packetResp = make(chan struct{}, 1)
-		c.keepaliveReset = make(chan struct{}, 1)
-
 		c.incomingPubChan = make(chan *packets.PublishPacket, c.options.MessageChannelDepth)
 		c.msgRouter.matchAndDispatch(c.incomingPubChan, c.options.Order, c)
 
@@ -250,22 +315,19 @@ func (c *client) Connect() Token {
 			go c.options.OnConnect(c)
 		}
 
-		// Take care of any messages in the store
-		//var leftovers []Receipt
-		if c.options.CleanSession == false {
-			//leftovers = c.resume()
-		} else {
-			c.persist.Reset()
-		}
-
+		c.workers.Add(4)
 		go errorWatch(c)
-
-		// Do not start incoming until resume has completed
-		c.workers.Add(3)
 		go alllogic(c)
 		go outgoing(c)
 		go incoming(c)
 
+		// Take care of any messages in the store
+		if c.options.CleanSession == false {
+			c.resume(c.options.ResumeSubs)
+		} else {
+			c.persist.Reset()
+		}
+
 		DEBUG.Println(CLI, "exit startClient")
 		t.flowComplete()
 	}()
@@ -283,28 +345,33 @@ func (c *client) reconnect() {
 	)
 
 	for rc != 0 && c.status != disconnected {
-		cm := newConnectMsgFromOptions(&c.options)
-
 		for _, broker := range c.options.Servers {
-		CONN:
+			cm := newConnectMsgFromOptions(&c.options, broker)
 			DEBUG.Println(CLI, "about to write new connect msg")
-			c.conn, err = openConnection(broker, &c.options.TLSConfig, c.options.ConnectTimeout)
+			c.conn, err = openConnection(broker, c.options.TLSConfig, c.options.ConnectTimeout, c.options.HTTPHeaders)
 			if err == nil {
 				DEBUG.Println(CLI, "socket connected to broker")
 				switch c.options.ProtocolVersion {
+				case 0x83:
+					DEBUG.Println(CLI, "Using MQTT 3.1b protocol")
+					cm.ProtocolName = "MQIsdp"
+					cm.ProtocolVersion = 0x83
+				case 0x84:
+					DEBUG.Println(CLI, "Using MQTT 3.1.1b protocol")
+					cm.ProtocolName = "MQTT"
+					cm.ProtocolVersion = 0x84
 				case 3:
 					DEBUG.Println(CLI, "Using MQTT 3.1 protocol")
 					cm.ProtocolName = "MQIsdp"
 					cm.ProtocolVersion = 3
 				default:
 					DEBUG.Println(CLI, "Using MQTT 3.1.1 protocol")
-					c.options.ProtocolVersion = 4
 					cm.ProtocolName = "MQTT"
 					cm.ProtocolVersion = 4
 				}
 				cm.Write(c.conn)
 
-				rc = c.connect()
+				rc, _ = c.connect()
 				if rc != packets.Accepted {
 					c.conn.Close()
 					c.conn = nil
@@ -313,11 +380,6 @@ func (c *client) reconnect() {
 						ERROR.Println(CLI, "Connecting to", broker, "CONNACK was not Accepted, but rather", packets.ConnackReturnCodes[rc])
 						continue
 					}
-					if c.options.ProtocolVersion == 4 {
-						DEBUG.Println(CLI, "Trying reconnect using MQTT 3.1 protocol")
-						c.options.ProtocolVersion = 3
-						goto CONN
-					}
 				}
 				break
 			} else {
@@ -339,64 +401,69 @@ func (c *client) reconnect() {
 		}
 	}
 	// Disconnect() must have been called while we were trying to reconnect.
-	if c.status == disconnected {
+	if c.connectionStatus() == disconnected {
 		DEBUG.Println(CLI, "Client moved to disconnected state while reconnecting, abandoning reconnect")
 		return
 	}
 
+	c.stop = make(chan struct{})
+
 	if c.options.KeepAlive != 0 {
+		atomic.StoreInt32(&c.pingOutstanding, 0)
+		atomic.StoreInt64(&c.lastReceived, time.Now().Unix())
+		atomic.StoreInt64(&c.lastSent, time.Now().Unix())
 		c.workers.Add(1)
 		go keepalive(c)
 	}
 
-	c.stop = make(chan struct{})
-
 	c.setConnected(connected)
 	DEBUG.Println(CLI, "client is reconnected")
 	if c.options.OnConnect != nil {
 		go c.options.OnConnect(c)
 	}
 
+	c.workers.Add(4)
 	go errorWatch(c)
-
-	c.workers.Add(3)
 	go alllogic(c)
 	go outgoing(c)
 	go incoming(c)
+
+	c.resume(false)
 }
 
 // This function is only used for receiving a connack
 // when the connection is first started.
 // This prevents receiving incoming data while resume
 // is in progress if clean session is false.
-func (c *client) connect() byte {
+func (c *client) connect() (byte, bool) {
 	DEBUG.Println(NET, "connect started")
 
 	ca, err := packets.ReadPacket(c.conn)
 	if err != nil {
 		ERROR.Println(NET, "connect got error", err)
-		return packets.ErrNetworkError
+		return packets.ErrNetworkError, false
 	}
 	if ca == nil {
 		ERROR.Println(NET, "received nil packet")
-		return packets.ErrNetworkError
+		return packets.ErrNetworkError, false
 	}
 
 	msg, ok := ca.(*packets.ConnackPacket)
 	if !ok {
 		ERROR.Println(NET, "received msg that was not CONNACK")
-		return packets.ErrNetworkError
+		return packets.ErrNetworkError, false
 	}
 
 	DEBUG.Println(NET, "received connack")
-	return msg.ReturnCode
+	return msg.ReturnCode, msg.SessionPresent
 }
 
 // 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 *client) Disconnect(quiesce uint) {
-	if c.status == connected {
+	status := atomic.LoadUint32(&c.status)
+	if status == connected {
 		DEBUG.Println(CLI, "disconnecting")
 		c.setConnected(disconnected)
 
@@ -409,6 +476,7 @@ func (c *client) Disconnect(quiesce uint) {
 	} else {
 		WARN.Println(CLI, "Disconnect() called but not connected (disconnected/reconnecting)")
 		c.setConnected(disconnected)
+		return
 	}
 
 	c.disconnect()
@@ -456,7 +524,9 @@ func (c *client) closeStop() {
 	case <-c.stop:
 		DEBUG.Println("In disconnect and stop channel is already closed")
 	default:
-		close(c.stop)
+		if c.stop != nil {
+			close(c.stop)
+		}
 	}
 }
 
@@ -472,6 +542,7 @@ func (c *client) disconnect() {
 	c.closeStop()
 	c.closeConn()
 	c.workers.Wait()
+	c.messageIds.cleanUp()
 	close(c.stopRouter)
 	DEBUG.Println(CLI, "disconnected")
 	c.persist.Close()
@@ -507,13 +578,17 @@ func (c *client) Publish(topic string, qos byte, retained bool, payload interfac
 		return token
 	}
 
-	DEBUG.Println(CLI, "sending publish message, topic:", topic)
 	if pub.Qos != 0 && pub.MessageID == 0 {
 		pub.MessageID = c.getID(token)
 		token.messageID = pub.MessageID
 	}
 	persistOutbound(c.persist, pub)
-	c.obound <- &PacketAndToken{p: pub, t: token}
+	if c.connectionStatus() == reconnecting {
+		DEBUG.Println(CLI, "storing publish message (reconnecting), topic:", topic)
+	} else {
+		DEBUG.Println(CLI, "sending publish message, topic:", topic)
+		c.obound <- &PacketAndToken{p: pub, t: token}
+	}
 	return token
 }
 
@@ -536,6 +611,10 @@ func (c *client) Subscribe(topic string, qos byte, callback MessageHandler) Toke
 	sub.Qoss = append(sub.Qoss, qos)
 	DEBUG.Println(CLI, sub.String())
 
+	if strings.HasPrefix(topic, "$share") {
+		topic = strings.Join(strings.Split(topic, "/")[2:], "/")
+	}
+
 	if callback != nil {
 		c.msgRouter.addRoute(topic, callback)
 	}
@@ -575,6 +654,61 @@ func (c *client) SubscribeMultiple(filters map[string]byte, callback MessageHand
 	return token
 }
 
+// Load all stored messages and resend them
+// Call this to ensure QOS > 1,2 even after an application crash
+func (c *client) resume(subscription bool) {
+
+	storedKeys := c.persist.All()
+	for _, key := range storedKeys {
+		packet := c.persist.Get(key)
+		details := packet.Details()
+		if isKeyOutbound(key) {
+			switch packet.(type) {
+			case *packets.SubscribePacket:
+				if subscription {
+					DEBUG.Println(STR, fmt.Sprintf("loaded pending subscribe (%d)", details.MessageID))
+					token := newToken(packets.Subscribe).(*SubscribeToken)
+					c.oboundP <- &PacketAndToken{p: packet, t: token}
+				}
+			case *packets.UnsubscribePacket:
+				if subscription {
+					DEBUG.Println(STR, fmt.Sprintf("loaded pending unsubscribe (%d)", details.MessageID))
+					token := newToken(packets.Unsubscribe).(*UnsubscribeToken)
+					c.oboundP <- &PacketAndToken{p: packet, t: token}
+				}
+			case *packets.PubrelPacket:
+				DEBUG.Println(STR, fmt.Sprintf("loaded pending pubrel (%d)", details.MessageID))
+				select {
+				case c.oboundP <- &PacketAndToken{p: packet, t: nil}:
+				case <-c.stop:
+				}
+			case *packets.PublishPacket:
+				token := newToken(packets.Publish).(*PublishToken)
+				token.messageID = details.MessageID
+				c.claimID(token, details.MessageID)
+				DEBUG.Println(STR, fmt.Sprintf("loaded pending publish (%d)", details.MessageID))
+				DEBUG.Println(STR, details)
+				c.obound <- &PacketAndToken{p: packet, t: token}
+			default:
+				ERROR.Println(STR, "invalid message type in store (discarded)")
+				c.persist.Del(key)
+			}
+		} else {
+			switch packet.(type) {
+			case *packets.PubrelPacket, *packets.PublishPacket:
+				DEBUG.Println(STR, fmt.Sprintf("loaded pending incomming (%d)", details.MessageID))
+				select {
+				case c.ibound <- packet:
+				case <-c.stop:
+				}
+			default:
+				ERROR.Println(STR, "invalid message type in store (discarded)")
+				c.persist.Del(key)
+			}
+		}
+	}
+}
+
 // Unsubscribe will end the subscription from each of the topics provided.
 // Messages published to those topics from other clients will no longer be
 // received.
@@ -599,6 +733,8 @@ func (c *client) Unsubscribe(topics ...string) Token {
 	return token
 }
 
+// OptionsReader returns a ClientOptionsReader which is a copy of the clientoptions
+// in use by the client.
 func (c *client) OptionsReader() ClientOptionsReader {
 	r := ClientOptionsReader{options: &c.options}
 	return r