Merge pull request eclipse-paho#214 from TheSilentForest/manually_ack_messages

Fix: The acknowledgement for incoming messages are sent even if the associated MessageHandler panics eclipse-paho#16

Author: alsm

message.go

@@ -18,6 +18,7 @@ import (
 	"net/url"
 
 	"github.com/eclipse/paho.mqtt.golang/packets"
+	"sync"
 )
 
 // Message defines the externals that a message implementation must support
@@ -30,6 +31,7 @@ type Message interface {
 	Topic() string
 	MessageID() uint16
 	Payload() []byte
+	Ack()
 }
 
 type message struct {
@@ -39,6 +41,8 @@ type message struct {
 	topic     string
 	messageID uint16
 	payload   []byte
+	once      sync.Once
+	ack       func()
 }
 
 func (m *message) Duplicate() bool {
@@ -65,14 +69,19 @@ func (m *message) Payload() []byte {
 	return m.payload
 }
 
-func messageFromPublish(p *packets.PublishPacket) Message {
+func (m *message) Ack() {
+	m.once.Do(m.ack)
+}
+
+func messageFromPublish(p *packets.PublishPacket, ack func()) Message {
 	return &message{
 		duplicate: p.Dup,
 		qos:       p.Qos,
 		retained:  p.Retain,
 		topic:     p.TopicName,
 		messageID: p.MessageID,
 		payload:   p.Payload,
+		ack:       ack,
 	}
 }
 

net.go

@@ -266,26 +266,9 @@ func alllogic(c *client) {
 				case 2:
 					c.incomingPubChan <- m
 					DEBUG.Println(NET, "done putting msg on incomingPubChan")
-					pr := packets.NewControlPacket(packets.Pubrec).(*packets.PubrecPacket)
-					pr.MessageID = m.MessageID
-					DEBUG.Println(NET, "putting pubrec msg on obound")
-					select {
-					case c.oboundP <- &PacketAndToken{p: pr, t: nil}:
-					case <-c.stop:
-					}
-					DEBUG.Println(NET, "done putting pubrec msg on obound")
 				case 1:
 					c.incomingPubChan <- m
 					DEBUG.Println(NET, "done putting msg on incomingPubChan")
-					pa := packets.NewControlPacket(packets.Puback).(*packets.PubackPacket)
-					pa.MessageID = m.MessageID
-					DEBUG.Println(NET, "putting puback msg on obound")
-					persistOutbound(c.persist, pa)
-					select {
-					case c.oboundP <- &PacketAndToken{p: pa, t: nil}:
-					case <-c.stop:
-					}
-					DEBUG.Println(NET, "done putting puback msg on obound")
 				case 0:
 					select {
 					case c.incomingPubChan <- m:
@@ -328,6 +311,34 @@ func alllogic(c *client) {
 	}
 }
 
+func (c *client) ackFunc(packet *packets.PublishPacket) func() {
+	return func() {
+		switch packet.Qos {
+		case 2:
+			pr := packets.NewControlPacket(packets.Pubrec).(*packets.PubrecPacket)
+			pr.MessageID = packet.MessageID
+			DEBUG.Println(NET, "putting pubrec msg on obound")
+			select {
+			case c.oboundP <- &PacketAndToken{p: pr, t: nil}:
+			case <-c.stop:
+			}
+			DEBUG.Println(NET, "done putting pubrec msg on obound")
+		case 1:
+			pa := packets.NewControlPacket(packets.Puback).(*packets.PubackPacket)
+			pa.MessageID = packet.MessageID
+			DEBUG.Println(NET, "putting puback msg on obound")
+			persistOutbound(c.persist, pa)
+			select {
+			case c.oboundP <- &PacketAndToken{p: pa, t: nil}:
+			case <-c.stop:
+			}
+			DEBUG.Println(NET, "done putting puback msg on obound")
+		case 0:
+			// do nothing, since there is no need to send an ack packet back
+		}
+	}
+}
+
 func errorWatch(c *client) {
 	defer c.workers.Done()
 	select {

router.go

@@ -146,13 +146,14 @@ func (r *router) matchAndDispatch(messages <-chan *packets.PublishPacket, order
 			case message := <-messages:
 				sent := false
 				r.RLock()
+				m := messageFromPublish(message, client.ackFunc(message))
 				handlers := []MessageHandler{}
 				for e := r.routes.Front(); e != nil; e = e.Next() {
 					if e.Value.(*route).match(message.TopicName) {
 						if order {
 							handlers = append(handlers, e.Value.(*route).callback)
 						} else {
-							go e.Value.(*route).callback(client, messageFromPublish(message))
+							go e.Value.(*route).callback(client, m)
 						}
 						sent = true
 					}
@@ -161,12 +162,12 @@ func (r *router) matchAndDispatch(messages <-chan *packets.PublishPacket, order
 					if order {
 						handlers = append(handlers, r.defaultHandler)
 					} else {
-						go r.defaultHandler(client, messageFromPublish(message))
+						go r.defaultHandler(client, m)
 					}
 				}
 				r.RUnlock()
 				for _, handler := range handlers {
-					handler(client, messageFromPublish(message))
+					handler(client, m)
 				}
 			case <-r.stop:
 				return