Bugfixes: Cleaning up Go-Routines Left Over in case of Bad Username and Password during login.

Adding ability to stop the messageid object when a connection is torn down, releasing the go-routine. Also, adding ability to stop the outgoing channel when no message is being sent (in the
case of error during initial connection, like bad username/password). Added a timeout to all attempts to send on outgoing via oboung and oboundP, to prevent them blocking forever if outgoing
is no longer running.

Change-Id: I0f9c3b4154fdeda1d552f8e7c9c14a6f7620b368
Signed-off-by: Christie Deiters <cdeiters@lutron.com>

Author: cdeiters

client.go

@@ -123,7 +123,15 @@ func (c *MqttClient) Start() ([]Receipt, error) {
 
 	cm := newConnectMsgFromOptions(c.options)
 	DEBUG.Println(CLI, "about to write new connect msg")
-	c.oboundP <- cm
+
+	if err := sendMessageWithTimeout(c.oboundP, cm); err != nil {
+		select {
+		case c.errors <- err:
+		default:
+			// c.errors is a buffer of one, so there must already be an error closing this connection.
+		}
+		return nil, err
+	}
 
 	rc := connect(c)
 	if chkrc(rc) != nil {
@@ -200,12 +208,11 @@ func (c *MqttClient) disconnect() {
 	c.connected = false
 	dm := newDisconnectMsg()
 
-	// Stop all go routines except outgoing
+	// send disconnect message
+	sendMessageWithTimeout(c.oboundP, dm)
+	// Stop all go routines
 	close(c.stop)
 
-	// Send disconnect message and stop outgoing
-	c.oboundP <- dm
-
 	DEBUG.Println(CLI, "disconnected")
 	c.persist.Close()
 }
@@ -228,9 +235,7 @@ func (c *MqttClient) Publish(qos QoS, topic string, payload interface{}) <-chan
 	r := make(chan Receipt, 1)
 	DEBUG.Println(CLI, "sending publish message, topic:", topic)
 
-	select {
-	case c.obound <- sendable{pub, r}:
-	case <-time.After(time.Second):
+	if err := sendSendableWithTimeout(c.obound, sendable{pub, r}); err != nil {
 		close(r)
 	}
 	return r
@@ -249,13 +254,10 @@ func (c *MqttClient) PublishMessage(topic string, message *Message) <-chan Recei
 
 	DEBUG.Println(CLI, "sending publish message, topic:", topic)
 
-	select {
-	case c.obound <- sendable{pub, r}:
-		return r
-	case <-time.After(time.Second):
+	if err := sendSendableWithTimeout(c.obound, sendable{pub, r}); err != nil {
 		close(r)
-		return nil
 	}
+	return r
 }
 
 // Start a new subscription. Provide a MessageHandler to be executed when
@@ -275,8 +277,9 @@ func (c *MqttClient) StartSubscription(callback MessageHandler, filters ...*Topi
 	}
 
 	r := make(chan Receipt, 1)
-
-	c.obound <- sendable{submsg, r}
+	if err := sendSendableWithTimeout(c.obound, sendable{submsg, r}); err != nil {
+		close(r)
+	}
 
 	DEBUG.Println(CLI, "exit StartSubscription")
 	return r, nil
@@ -293,8 +296,10 @@ func (c *MqttClient) EndSubscription(topics ...string) (<-chan Receipt, error) {
 	usmsg := newUnsubscribeMsg(topics...)
 
 	r := make(chan Receipt, 1)
+	if err := sendSendableWithTimeout(c.obound, sendable{usmsg, r}); err != nil {
+		close(r)
+	}
 
-	c.obound <- sendable{usmsg, r}
 	for _, topic := range topics {
 		c.options.msgRouter.deleteRoute(topic)
 	}

messageids.go

@@ -15,6 +15,7 @@
 package mqtt
 
 import (
+	"errors"
 	"sync"
 )
 
@@ -25,30 +26,41 @@ type MId uint16
 
 type messageIds struct {
 	sync.Mutex
-	idChan chan MId
-	index  map[MId]bool
+	idChan   chan MId
+	index    map[MId]bool
+	stopChan chan struct{}
 }
 
 const (
 	MId_MAX MId = 65535
 	MId_MIN MId = 1
 )
 
+func (mids *messageIds) stop() {
+	close(mids.stopChan)
+}
+
 func (mids *messageIds) generateMsgIds() {
+
 	mids.idChan = make(chan MId, 10)
-	go func() {
+	mids.stopChan = make(chan struct{})
+	go func(mid *messageIds) {
 		for {
-			mids.Lock()
+			mid.Lock()
 			for i := MId_MIN; i < MId_MAX; i++ {
-				if !mids.index[i] {
-					mids.index[i] = true
-					mids.Unlock()
-					mids.idChan <- i
+				if !mid.index[i] {
+					mid.index[i] = true
+					mid.Unlock()
+					select {
+					case mid.idChan <- i:
+					case <-mid.stopChan:
+						return
+					}
 					break
 				}
 			}
 		}
-	}()
+	}(mids)
 }
 
 func (mids *messageIds) freeId(id MId) {
@@ -58,6 +70,11 @@ func (mids *messageIds) freeId(id MId) {
 	mids.index[id] = false
 }
 
-func (mids *messageIds) getId() MId {
-	return <-mids.idChan
+func (mids *messageIds) getId() (MId, error) {
+	select {
+	case i := <-mids.idChan:
+		return i, nil
+	case <-mids.stopChan:
+	}
+	return 0, errors.New("Failed to get next message id.")
 }

net.go

@@ -17,6 +17,7 @@ package mqtt
 import (
 	"code.google.com/p/go.net/websocket"
 	"crypto/tls"
+	"errors"
 	"io"
 	"net"
 	"net/url"
@@ -133,6 +134,32 @@ func incoming(c *MqttClient) {
 	}
 }
 
+func sendSendableWithTimeout(sChan chan sendable, s sendable) error {
+
+	t := time.NewTimer(time.Second)
+	select {
+	case sChan <- s:
+		// stop timer so we don't leak it.
+		t.Stop()
+	case <-t.C:
+		return errors.New("Timed out sending message")
+	}
+	return nil
+}
+
+func sendMessageWithTimeout(mChan chan *Message, m *Message) error {
+
+	t := time.NewTimer(time.Second)
+	select {
+	case mChan <- m:
+		// stop timer so we don't leak it.
+		t.Stop()
+	case <-t.C:
+		return errors.New("Timed out sending message")
+	}
+	return nil
+}
+
 // receive a Message object on obound, and then
 // actually send outgoing message to the wire
 func outgoing(c *MqttClient) {
@@ -142,12 +169,24 @@ func outgoing(c *MqttClient) {
 	for {
 		DEBUG.Println(NET, "outgoing waiting for an outbound message")
 		select {
+		case <-c.stop:
+			// The connection has been closed, we won't be receiving any more requests.
+			return
 		case out := <-c.obound:
 			msg := out.m
 			msgtype := msg.msgType()
 			DEBUG.Println(NET, "obound got msg to write, type:", msgtype)
 			if msg.QoS() != QOS_ZERO && msg.MsgId() == 0 {
-				msg.setMsgId(c.options.mids.getId())
+				i, err := c.options.mids.getId()
+				if err != nil {
+					select {
+					case c.errors <- err:
+					default:
+						// c.errors is a buffer of one, so there must already be an error closing this connection.
+					}
+					return
+				}
+				msg.setMsgId(i)
 			}
 			if out.r != nil {
 				c.receipts.put(msg.MsgId(), out.r)
@@ -283,18 +322,12 @@ func alllogic(c *MqttClient) {
 				id := msg.MsgId()
 				pubrelMsg := newPubRelMsg()
 				pubrelMsg.setMsgId(id)
-				select {
-				case c.obound <- sendable{pubrelMsg, nil}:
-				case <-time.After(time.Second):
-				}
+				sendSendableWithTimeout(c.obound, sendable{pubrelMsg, nil})
 			case PUBREL:
 				DEBUG.Println(NET, "received pubrel, id:", msg.MsgId())
 				pubcompMsg := newPubCompMsg()
 				pubcompMsg.setMsgId(msg.MsgId())
-				select {
-				case c.obound <- sendable{pubcompMsg, nil}:
-				case <-time.After(time.Second):
-				}
+				sendSendableWithTimeout(c.obound, sendable{pubcompMsg, nil})
 			case PUBCOMP:
 				DEBUG.Println(NET, "received pubcomp, id:", msg.MsgId())
 				c.receipts.get(msg.MsgId()) <- Receipt{}
@@ -312,6 +345,7 @@ func alllogic(c *MqttClient) {
 			// but let it know to stop anyways.
 			close(c.options.stopRouter)
 			close(c.stop)
+			c.options.mids.stop()
 			c.conn.Close()
 
 			// Call onConnectionLost or default error handler

ping.go

@@ -59,8 +59,13 @@ func keepalive(c *MqttClient) {
 				if !c.pingOutstanding {
 					DEBUG.Println(PNG, "keepalive sending ping")
 					ping := newPingReqMsg()
-					c.oboundP <- ping
-					c.pingOutstanding = true
+					if err := sendMessageWithTimeout(c.oboundP, ping); err != nil {
+						DEBUG.Println(PNG, "unable to send ping, disconnecting")
+						go c.options.onconnlost(c, errors.New("unable to send ping, disconnecting"))
+						c.disconnect()
+					} else {
+						c.pingOutstanding = true
+					}
 				} else {
 					CRITICAL.Println(PNG, "pingresp not received, disconnecting")
 					go c.options.onconnlost(c, errors.New("pingresp not received, disconnecting"))

state.go

@@ -69,7 +69,7 @@ func (c *MqttClient) resume() []Receipt {
 			if c.receipts.get(m.MsgId()) == nil { // will be nil if client crashed
 				c.receipts.put(m.MsgId(), make(chan Receipt, 1))
 			}
-			c.obound <- sendable{m, c.receipts.get(m.MsgId())}
+			sendSendableWithTimeout(c.obound, sendable{m, c.receipts.get(m.MsgId())})
 		}
 	}
 
@@ -81,7 +81,7 @@ func (c *MqttClient) resume() []Receipt {
 			if c.receipts.get(m.MsgId()) == nil { // will be nil if client crashed
 				c.receipts.put(m.MsgId(), make(chan Receipt, 1))
 			}
-			c.obound <- sendable{m, c.receipts.get(m.MsgId())}
+			sendSendableWithTimeout(c.obound, sendable{m, c.receipts.get(m.MsgId())})
 		}
 	}
 

unit_messageids_test.go

@@ -23,20 +23,29 @@ func Test_getId(t *testing.T) {
 	mids := &messageIds{index: make(map[MId]bool)}
 	mids.generateMsgIds()
 
-	i1 := mids.getId()
+	i1, err := mids.getId()
+	if err != nil {
+		t.Fatalf("Failed to get id. %s", err.Error())
+	}
 
 	if i1 != MId(1) {
 		t.Fatalf("i1 was wrong: %v", i1)
 	}
 
-	i2 := mids.getId()
+	i2, err := mids.getId()
+	if err != nil {
+		t.Fatalf("Failed to get id. %s", err.Error())
+	}
 
 	if i2 != MId(2) {
 		t.Fatalf("i2 was wrong: %v", i2)
 	}
 
 	for i := 3; i < 100; i++ {
-		id := mids.getId()
+		id, err := mids.getId()
+		if err != nil {
+			t.Fatalf("Failed to get id. %s", err.Error())
+		}
 		if id != MId(i) {
 			t.Fatalf("id was wrong expected %v got %v", i, id)
 		}
@@ -47,14 +56,20 @@ func Test_freeId(t *testing.T) {
 	mids := &messageIds{index: make(map[MId]bool)}
 	mids.generateMsgIds()
 
-	i1 := mids.getId()
+	i1, err := mids.getId()
+	if err != nil {
+		t.Fatalf("Failed to get id. %s", err.Error())
+	}
 	mids.freeId(i1)
 
 	if i1 != MId(1) {
 		t.Fatalf("i1 was wrong: %v", i1)
 	}
 
-	i2 := mids.getId()
+	i2, err := mids.getId()
+	if err != nil {
+		t.Fatalf("Failed to get id. %s", err.Error())
+	}
 	fmt.Printf("i2: %v\n", i2)
 }
 
@@ -69,23 +84,38 @@ func Test_messageids_mix(t *testing.T) {
 
 	go func() {
 		for i := 0; i < 10000; i++ {
-			a <- mids.getId()
+			id, err := mids.getId()
+			if err != nil {
+				t.Fatalf("Failed to get id. %s", err.Error())
+			}
+			a <- id
+
 			mids.freeId(<-b)
 		}
 		done <- true
 	}()
 
 	go func() {
 		for i := 0; i < 10000; i++ {
-			b <- mids.getId()
+			id, err := mids.getId()
+			if err != nil {
+				t.Fatalf("Failed to get id. %s", err.Error())
+			}
+			b <- id
+
 			mids.freeId(<-c)
 		}
 		done <- true
 	}()
 
 	go func() {
 		for i := 0; i < 10000; i++ {
-			c <- mids.getId()
+			id, err := mids.getId()
+			if err != nil {
+				t.Fatalf("Failed to get id. %s", err.Error())
+			}
+			c <- id
+
 			mids.freeId(<-a)
 		}
 		done <- true