Improve handling of SUBSCRIBE/UNSUBSCRIBE over disconnect/reconnect.

client.go

@@ -514,7 +514,9 @@ func (c *client) internalConnLost(err error) {
 			reconnect := c.options.AutoReconnect && c.connectionStatus() > connecting
 
 			if c.options.CleanSession && !reconnect {
-				c.messageIds.cleanUp()
+				c.messageIds.cleanUp() // completes PUB/SUB/UNSUB tokens
+			} else if !c.options.ResumeSubs {
+				c.messageIds.cleanUpSubscribe() // completes SUB/UNSUB tokens
 			}
 			if reconnect {
 				c.setConnected(reconnecting)
@@ -811,7 +813,9 @@ func (c *client) Subscribe(topic string, qos byte, callback MessageHandler) Toke
 	}
 	DEBUG.Println(CLI, sub.String())
 
-	persistOutbound(c.persist, sub)
+	if c.options.ResumeSubs { // Only persist if we need this to resume subs after a disconnection
+		persistOutbound(c.persist, sub)
+	}
 	switch c.connectionStatus() {
 	case connecting:
 		DEBUG.Println(CLI, "storing subscribe message (connecting), topic:", topic)
@@ -883,7 +887,9 @@ func (c *client) SubscribeMultiple(filters map[string]byte, callback MessageHand
 		sub.MessageID = mID
 		token.messageID = mID
 	}
-	persistOutbound(c.persist, sub)
+	if c.options.ResumeSubs { // Only persist if we need this to resume subs after a disconnection
+		persistOutbound(c.persist, sub)
+	}
 	switch c.connectionStatus() {
 	case connecting:
 		DEBUG.Println(CLI, "storing subscribe message (connecting), topics:", sub.Topics)
@@ -1096,7 +1102,9 @@ func (c *client) Unsubscribe(topics ...string) Token {
 		token.messageID = mID
 	}
 
-	persistOutbound(c.persist, unsub)
+	if c.options.ResumeSubs { // Only persist if we need this to resume subs after a disconnection
+		persistOutbound(c.persist, unsub)
+	}
 
 	switch c.connectionStatus() {
 	case connecting:

messageids.go

@@ -42,6 +42,7 @@ const (
 	midMax uint16 = 65535
 )
 
+// cleanup clears the message ID map; completes all token types and sets error on PUB, SUB and UNSUB tokens.
 func (mids *messageIds) cleanUp() {
 	mids.Lock()
 	for _, token := range mids.index {
@@ -52,7 +53,7 @@ func (mids *messageIds) cleanUp() {
 			token.setError(fmt.Errorf("connection lost before Subscribe completed"))
 		case *UnsubscribeToken:
 			token.setError(fmt.Errorf("connection lost before Unsubscribe completed"))
-		case nil:
+		case nil: // should not be any nil entries
 			continue
 		}
 		token.flowComplete()
@@ -62,6 +63,24 @@ func (mids *messageIds) cleanUp() {
 	DEBUG.Println(MID, "cleaned up")
 }
 
+// cleanUpSubscribe removes all SUBSCRIBE and UNSUBSCRIBE tokens (setting error)
+// This may be called when the connection is lost, and we will not be resending SUB/UNSUB packets
+func (mids *messageIds) cleanUpSubscribe() {
+	mids.Lock()
+	for mid, token := range mids.index {
+		switch token.(type) {
+		case *SubscribeToken:
+			token.setError(fmt.Errorf("connection lost before Subscribe completed"))
+			delete(mids.index, mid)
+		case *UnsubscribeToken:
+			token.setError(fmt.Errorf("connection lost before Unsubscribe completed"))
+			delete(mids.index, mid)
+		}
+	}
+	mids.Unlock()
+	DEBUG.Println(MID, "cleaned up subs")
+}
+
 func (mids *messageIds) freeID(id uint16) {
 	mids.Lock()
 	delete(mids.index, id)