Merge branch 'master' into master

Author: Al S-M

client.go

@@ -18,6 +18,7 @@
 package mqtt
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"net"
@@ -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
 }
 
@@ -224,8 +223,8 @@ func (c *client) Connect() Token {
 		return t
 	}
 
-	c.obound = make(chan *PacketAndToken, c.options.MessageChannelDepth)
-	c.oboundP = make(chan *PacketAndToken, c.options.MessageChannelDepth)
+	c.obound = make(chan *PacketAndToken)
+	c.oboundP = make(chan *PacketAndToken)
 	c.ibound = make(chan packets.ControlPacket)
 
 	c.persist.Open()
@@ -337,7 +336,7 @@ func (c *client) Connect() Token {
 			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)
@@ -353,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()
@@ -611,11 +610,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
@@ -633,9 +634,13 @@ func (c *client) Publish(topic string, qos byte, retained bool, payload interfac
 		DEBUG.Println(CLI, "storing publish message (reconnecting), topic:", topic)
 	default:
 		DEBUG.Println(CLI, "sending publish message, topic:", topic)
+		publishWaitTimeout := c.options.WriteTimeout
+		if publishWaitTimeout == 0 {
+			publishWaitTimeout = time.Second * 30
+		}
 		select {
 		case c.obound <- &PacketAndToken{p: pub, t: token}:
-		case <-time.After(c.options.WriteTimeout):
+		case <-time.After(publishWaitTimeout):
 			token.setError(errors.New("publish was broken by timeout"))
 		}
 	}

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
 		}

fvt_client_test.go

@@ -144,6 +144,24 @@ func Test_Publish_3(t *testing.T) {
 	c.Disconnect(250)
 }
 
+func Test_Publish_BytesBuffer(t *testing.T) {
+	ops := NewClientOptions()
+	ops.AddBroker(FVTTCP)
+	ops.SetClientID("Publish_BytesBuffer")
+
+	c := NewClient(ops)
+	token := c.Connect()
+	if token.Wait() && token.Error() != nil {
+		t.Fatalf("Error on Client.Connect(): %v", token.Error())
+	}
+
+	payload := bytes.NewBufferString("Publish qos0")
+
+	c.Publish("test/Publish", 0, false, payload)
+
+	c.Disconnect(250)
+}
+
 func Test_Subscribe(t *testing.T) {
 	pops := NewClientOptions()
 	pops.AddBroker(FVTTCP)
@@ -1028,7 +1046,7 @@ func Test_cleanUpMids(t *testing.T) {
 	c.(*client).messageIds.Unlock()
 	c.(*client).internalConnLost(fmt.Errorf("cleanup test"))
 
-	time.Sleep(5 * time.Second)
+	time.Sleep(1 * time.Second)
 	if !c.IsConnected() {
 		t.Fail()
 	}

options.go

@@ -10,6 +10,7 @@
  *    Seth Hoenig
  *    Allan Stockdill-Mander
  *    Mike Robertson
+ *    Måns Ansgariusson
  */
 
 // Portions copyright © 2018 TIBCO Software Inc.
@@ -20,6 +21,7 @@ import (
 	"crypto/tls"
 	"net/http"
 	"net/url"
+	"regexp"
 	"strings"
 	"time"
 )
@@ -113,7 +115,6 @@ func NewClientOptions() *ClientOptions {
 		OnConnect:               nil,
 		OnConnectionLost:        DefaultConnectionLostHandler,
 		WriteTimeout:            0, // 0 represents timeout disabled
-		MessageChannelDepth:     100,
 		ResumeSubs:              false,
 		HTTPHeaders:             make(map[string][]string),
 	}
@@ -129,12 +130,14 @@ func NewClientOptions() *ClientOptions {
 //
 // An example broker URI would look like: tcp://foobar.com:1883
 func (o *ClientOptions) AddBroker(server string) *ClientOptions {
+	re := regexp.MustCompile(`%(25)?`)
 	if len(server) > 0 && server[0] == ':' {
 		server = "127.0.0.1" + server
 	}
 	if !strings.Contains(server, "://") {
 		server = "tcp://" + server
 	}
+	server = re.ReplaceAllLiteralString(server, "%25")
 	brokerURI, err := url.Parse(server)
 	if err != nil {
 		ERROR.Println(CLI, "Failed to parse %q broker address: %s", server, err)
@@ -344,10 +347,8 @@ func (o *ClientOptions) SetConnectRetry(a bool) *ClientOptions {
 	return o
 }
 
-// SetMessageChannelDepth sets the size of the internal queue that holds messages while the
-// client is temporairily offline, allowing the application to publish when the client is
-// reconnecting. This setting is only valid if AutoReconnect is set to true, it is otherwise
-// ignored.
+// SetMessageChannelDepth DEPRECATED The value set here no longer has any effect, this function
+// remains so the API is not altered.
 func (o *ClientOptions) SetMessageChannelDepth(s uint) *ClientOptions {
 	o.MessageChannelDepth = s
 	return o

router.go

@@ -37,17 +37,11 @@ type route struct {
 // 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
+		return len(topic) == 0
 	}
 
 	if len(topic) == 0 {
-		if route[0] == "#" {
-			return true
-		}
-		return false
+		return route[0] == "#"
 	}
 
 	if route[0] == "#" {

topic.go

@@ -75,7 +75,7 @@ func validateTopicAndQos(topic string, qos byte) error {
 		}
 	}
 
-	if qos < 0 || qos > 2 {
+	if qos > 2 {
 		return ErrInvalidQos
 	}
 	return nil

websocket.go

@@ -27,7 +27,7 @@ func NewWebsocket(host string, tlsc *tls.Config, timeout time.Duration, requestH
 	ws, _, err := dialer.Dial(host, requestHeader)
 
 	if err != nil {
-		panic(err)
+		return nil, err
 	}
 
 	wrapper := &websocketConnector{