Merge branch 'master' into patch-1

Author: Al S-M

README.md

@@ -22,11 +22,11 @@ This client is designed to work with the standard Go tools, so installation is a
 go get github.com/eclipse/paho.mqtt.golang
 ```
 
-The client depends on Google's [websockets](https://godoc.org/golang.org/x/net/websocket) and [proxy](https://godoc.org/golang.org/x/net/proxy) package, 
+The client depends on Google's [proxy](https://godoc.org/golang.org/x/net/proxy) package and the [websockets](https://godoc.org/github.com/gorilla/websocket) package, 
 also easily installed with the commands:
 
 ```
-go get golang.org/x/net/websocket
+go get github.com/gorilla/websocket
 go get golang.org/x/net/proxy
 ```
 
@@ -44,6 +44,10 @@ import "github.com/eclipse/paho.mqtt.golang"
 
 Samples are available in the `cmd` directory for reference.
 
+Note:
+
+The library also supports using MQTT over websockets by using the `ws://` (unsecure) or `wss://` (secure) prefix in the URI. If the client is running behind a corporate http/https proxy then the following environment variables `HTTP_PROXY`, `HTTPS_PROXY` and `NO_PROXY` are taken into account when establishing the connection.
+
 
 Runtime tracing
 ---------------

client.go

@@ -18,6 +18,7 @@
 package mqtt
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"net"
@@ -95,8 +96,8 @@ type Client interface {
 
 // client implements the Client interface
 type client struct {
-	lastSent        int64
-	lastReceived    int64
+	lastSent        atomic.Value
+	lastReceived    atomic.Value
 	pingOutstanding int32
 	status          uint32
 	sync.RWMutex
@@ -140,9 +141,7 @@ func NewClient(o *ClientOptions) Client {
 	c.messageIds = messageIds{index: make(map[uint16]tokenCompletor)}
 	c.msgRouter, c.stopRouter = newRouter()
 	c.msgRouter.setDefaultHandler(c.options.DefaultPublishHandler)
-	if !c.options.AutoReconnect {
-		c.options.MessageChannelDepth = 0
-	}
+
 	return c
 }
 
@@ -157,6 +156,8 @@ func (c *client) AddRoute(topic string, callback MessageHandler) {
 
 // IsConnected returns a bool signifying whether
 // the client is connected or not.
+// connected means that the connection is up now OR it will
+// be established/reestablished automatically when possible
 func (c *client) IsConnected() bool {
 	c.RLock()
 	defer c.RUnlock()
@@ -166,6 +167,8 @@ func (c *client) IsConnected() bool {
 		return true
 	case c.options.AutoReconnect && status > connecting:
 		return true
+	case c.options.ConnectRetry && status == connecting:
+		return true
 	default:
 		return false
 	}
@@ -210,14 +213,27 @@ func (c *client) Connect() Token {
 	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)
+	if c.options.ConnectRetry && atomic.LoadUint32(&c.status) != disconnected {
+		// if in any state other than disconnected and ConnectRetry is
+		// enabled then the connection will come up automatically
+		// client can assume connection is up
+		WARN.Println(CLI, "Connect() called but not disconnected")
+		t.returnCode = packets.Accepted
+		t.flowComplete()
+		return t
+	}
+
+	c.obound = make(chan *PacketAndToken)
+	c.oboundP = make(chan *PacketAndToken)
 	c.ibound = make(chan packets.ControlPacket)
 
-	go func() {
-		c.persist.Open()
+	c.persist.Open()
+	if c.options.ConnectRetry {
+		c.reserveStoredPublishIDs() // Reserve IDs to allow publish before connect complete
+	}
+	c.setConnected(connecting)
 
-		c.setConnected(connecting)
+	go func() {
 		c.errors = make(chan error, 1)
 		c.stop = make(chan struct{})
 
@@ -229,12 +245,16 @@ func (c *client) Connect() Token {
 			return
 		}
 
+	RETRYCONN:
 		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.options.HTTPHeaders)
+			c.Lock()
+			c.conn, err = openConnection(broker, c.options.TLSConfig, c.options.ConnectTimeout,
+				c.options.HTTPHeaders)
+			c.Unlock()
 			if err == nil {
 				DEBUG.Println(CLI, "socket connected to broker")
 				switch c.options.ProtocolVersion {
@@ -260,10 +280,12 @@ func (c *client) Connect() Token {
 
 				rc, t.sessionPresent = c.connect()
 				if rc != packets.Accepted {
+					c.Lock()
 					if c.conn != nil {
 						c.conn.Close()
 						c.conn = nil
 					}
+					c.Unlock()
 					//if the protocol version was explicitly set don't do any fallback
 					if c.options.protocolVersionExplicit {
 						ERROR.Println(CLI, "Connecting to", broker, "CONNACK was not CONN_ACCEPTED, but rather", packets.ConnackReturnCodes[rc])
@@ -284,6 +306,14 @@ func (c *client) Connect() Token {
 		}
 
 		if c.conn == nil {
+			if c.options.ConnectRetry {
+				DEBUG.Println(CLI, "Connect failed, sleeping for", int(c.options.ConnectRetryInterval.Seconds()), "seconds and will then retry")
+				time.Sleep(c.options.ConnectRetryInterval)
+
+				if atomic.LoadUint32(&c.status) == connecting {
+					goto RETRYCONN
+				}
+			}
 			ERROR.Println(CLI, "Failed to connect to a broker")
 			c.setConnected(disconnected)
 			c.persist.Close()
@@ -300,13 +330,13 @@ func (c *client) Connect() Token {
 
 		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.lastReceived.Store(time.Now())
+			c.lastSent.Store(time.Now())
 			c.workers.Add(1)
 			go keepalive(c)
 		}
 
-		c.incomingPubChan = make(chan *packets.PublishPacket, c.options.MessageChannelDepth)
+		c.incomingPubChan = make(chan *packets.PublishPacket)
 		c.msgRouter.matchAndDispatch(c.incomingPubChan, c.options.Order, c)
 
 		c.setConnected(connected)
@@ -322,7 +352,7 @@ func (c *client) Connect() Token {
 		go incoming(c)
 
 		// Take care of any messages in the store
-		if c.options.CleanSession == false {
+		if !c.options.CleanSession {
 			c.resume(c.options.ResumeSubs)
 		} else {
 			c.persist.Reset()
@@ -375,8 +405,10 @@ func (c *client) reconnect() {
 
 				rc, _ = c.connect()
 				if rc != packets.Accepted {
-					c.conn.Close()
-					c.conn = nil
+					if c.conn != nil {
+						c.conn.Close()
+						c.conn = nil
+					}
 					//if the protocol version was explicitly set don't do any fallback
 					if c.options.protocolVersionExplicit {
 						ERROR.Println(CLI, "Connecting to", broker, "CONNACK was not Accepted, but rather", packets.ConnackReturnCodes[rc])
@@ -412,8 +444,8 @@ func (c *client) reconnect() {
 
 	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.lastReceived.Store(time.Now())
+		c.lastSent.Store(time.Now())
 		c.workers.Add(1)
 		go keepalive(c)
 	}
@@ -580,11 +612,13 @@ func (c *client) Publish(topic string, qos byte, retained bool, payload interfac
 	pub.Qos = qos
 	pub.TopicName = topic
 	pub.Retain = retained
-	switch payload.(type) {
+	switch p := payload.(type) {
 	case string:
-		pub.Payload = []byte(payload.(string))
+		pub.Payload = []byte(p)
 	case []byte:
-		pub.Payload = payload.([]byte)
+		pub.Payload = p
+	case bytes.Buffer:
+		pub.Payload = p.Bytes()
 	default:
 		token.setError(fmt.Errorf("Unknown payload type"))
 		return token
@@ -595,11 +629,22 @@ func (c *client) Publish(topic string, qos byte, retained bool, payload interfac
 		token.messageID = pub.MessageID
 	}
 	persistOutbound(c.persist, pub)
-	if c.connectionStatus() == reconnecting {
+	switch c.connectionStatus() {
+	case connecting:
+		DEBUG.Println(CLI, "storing publish message (connecting), topic:", topic)
+	case reconnecting:
 		DEBUG.Println(CLI, "storing publish message (reconnecting), topic:", topic)
-	} else {
+	default:
 		DEBUG.Println(CLI, "sending publish message, topic:", topic)
-		c.obound <- &PacketAndToken{p: pub, t: token}
+		publishWaitTimeout := c.options.WriteTimeout
+		if publishWaitTimeout == 0 {
+			publishWaitTimeout = time.Second * 30
+		}
+		select {
+		case c.obound <- &PacketAndToken{p: pub, t: token}:
+		case <-time.After(publishWaitTimeout):
+			token.setError(errors.New("publish was broken by timeout"))
+		}
 	}
 	return token
 }
@@ -664,27 +709,58 @@ func (c *client) SubscribeMultiple(filters map[string]byte, callback MessageHand
 	return token
 }
 
+// reserveStoredPublishIDs reserves the ids for publish packets in the persistant store to ensure these are not duplicated
+func (c *client) reserveStoredPublishIDs() {
+	// The resume function sets the stored id for publish packets only (some other packets
+	// will get new ids in net code). This means that the only keys we need to ensure are
+	// unique are the publish ones (and these will completed/replaced in resume() )
+	if c.options.CleanSession == false {
+		storedKeys := c.persist.All()
+		for _, key := range storedKeys {
+			packet := c.persist.Get(key)
+			if packet == nil {
+				continue
+			}
+			switch packet.(type) {
+			case *packets.PublishPacket:
+				details := packet.Details()
+				token := &PlaceHolderToken{id: details.MessageID}
+				c.claimID(token, details.MessageID)
+			}
+		}
+	}
+}
+
 // 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)
+		if packet == nil {
+			continue
+		}
 		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}
+					select {
+					case c.oboundP <- &PacketAndToken{p: packet, t: token}:
+					case <-c.stop:
+					}
 				}
 			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}
+					select {
+					case c.oboundP <- &PacketAndToken{p: packet, t: token}:
+					case <-c.stop:
+					}
 				}
 			case *packets.PubrelPacket:
 				DEBUG.Println(STR, fmt.Sprintf("loaded pending pubrel (%d)", details.MessageID))
@@ -698,7 +774,10 @@ func (c *client) resume(subscription bool) {
 				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}
+				select {
+				case c.obound <- &PacketAndToken{p: packet, t: token}:
+				case <-c.stop:
+				}
 			default:
 				ERROR.Println(STR, "invalid message type in store (discarded)")
 				c.persist.Del(key)

cmd/ssl/main.go

@@ -113,7 +113,7 @@ func main() {
 	c.Subscribe("/go-mqtt/sample", 0, nil)
 
 	i := 0
-	for _ = range time.Tick(time.Duration(1) * time.Second) {
+	for range time.Tick(time.Duration(1) * time.Second) {
 		if i == 5 {
 			break
 		}

filestore.go

@@ -166,7 +166,7 @@ func (store *FileStore) all() []string {
 	for _, f := range files {
 		DEBUG.Println(STR, "file in All():", f.Name())
 		name := f.Name()
-		if name[len(name)-4:len(name)] != msgExt {
+		if name[len(name)-4:] != msgExt {
 			DEBUG.Println(STR, "skipping file, doesn't have right extension: ", name)
 			continue
 		}