=gomod

Author: liangdas

go.mod

@@ -0,0 +1,8 @@
+module github.com/liangdas/armyant
+
+go 1.13
+
+require (
+	github.com/eclipse/paho.mqtt.golang v1.2.0
+	golang.org/x/net v0.0.0-20191126235420-ef20fe5d7933
+)

go.sum

@@ -0,0 +1,8 @@
+github.com/eclipse/paho.mqtt.golang v1.2.0 h1:1F8mhG9+aO5/xpdtFkW4SxOJB67ukuDC3t2y2qayIX0=
+github.com/eclipse/paho.mqtt.golang v1.2.0/go.mod h1:H9keYFcgq3Qr5OUJm/JZI/i6U7joQ8SYLhZwfeOo6Ts=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/net v0.0.0-20191126235420-ef20fe5d7933 h1:e6HwijUxhDe+hPNjZQQn9bA5PW3vNmnN64U2ZW759Lk=
+golang.org/x/net v0.0.0-20191126235420-ef20fe5d7933/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=

work/mqttwork.go

@@ -17,18 +17,21 @@ import (
 	"fmt"
 	MQTT "github.com/eclipse/paho.mqtt.golang"
 	"strings"
+	"sync"
 	_ "time"
 	"net/url"
 )
 
 type MqttWork struct {
 	client        MQTT.Client
 	waiting_queue map[string]func(client MQTT.Client, msg MQTT.Message)
+	lock			*sync.Mutex
 	curr_id       int64
 }
 
 func (this *MqttWork) GetDefaultOptions(addrURI string) *MQTT.ClientOptions {
 	this.curr_id = 0
+	this.lock=new(sync.Mutex)
 	this.waiting_queue = make(map[string]func(client MQTT.Client, msg MQTT.Message))
 	opts := MQTT.NewClientOptions()
 	opts.AddBroker(addrURI)
@@ -40,6 +43,7 @@ func (this *MqttWork) GetDefaultOptions(addrURI string) *MQTT.ClientOptions {
 	opts.SetAutoReconnect(false)
 	opts.SetDefaultPublishHandler(func(client MQTT.Client, msg MQTT.Message) {
 		//收到消息
+		this.lock.Lock()
 		if callback, ok := this.waiting_queue[msg.Topic()]; ok {
 			//有等待消息的callback 还缺一个信息超时的处理机制
 			_, err := url.Parse(msg.Topic())
@@ -52,6 +56,7 @@ func (this *MqttWork) GetDefaultOptions(addrURI string) *MQTT.ClientOptions {
 			}
 			go callback(client, msg)
 		}
+		this.lock.Unlock()
 	})
 	return opts
 }
@@ -147,5 +152,7 @@ func (this *MqttWork) RequestNR(topic string, body []byte) {
  */
 func (this *MqttWork) On(topic string, callback func(client MQTT.Client, msg MQTT.Message)) {
 	////服务器不会返回结果
+	this.lock.Lock()
 	this.waiting_queue[topic] = callback //添加这条消息到等待队列
+	this.lock.Unlock()
 }