add mqtt

庄敏学
go.mod
go.sum
main.go
pkg/constant/mqtt.go
pkg/infrastructure/mqtt/produce_client.go
pkg/infrastructure/mqtt/subscribe_client.go
pkg/port/mqtt/mqtt.go
--- a/go.mod
查看文件 @bc85917
+++ b/go.mod
查看文件 @bc85917
@@ -5,6 +5,7 @@ go 1.16
 require (
 	github.com/ajg/form v1.5.1 // indirect
 	github.com/beego/beego/v2 v2.0.1
+	github.com/eclipse/paho.mqtt.golang v1.3.5
 	github.com/fasthttp-contrib/websocket v0.0.0-20160511215533-1f3b11f56072 // indirect
 	github.com/fatih/structs v1.1.0 // indirect
 	github.com/gavv/httpexpect v2.0.0+incompatible
--- a/go.sum
查看文件 @bc85917
+++ b/go.sum
查看文件 @bc85917
@@ -68,6 +68,8 @@ github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 h1:YEetp8
 github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1:+020luEh2TKB4/GOp8oxxtq0Daoen/Cii55CzbTV6DU=
 github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc=
 github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
+github.com/eclipse/paho.mqtt.golang v1.3.5 h1:sWtmgNxYM9P2sP+xEItMozsR3w0cqZFlqnNN1bdl41Y=
+github.com/eclipse/paho.mqtt.golang v1.3.5/go.mod h1:eTzb4gxwwyWpqBUHGQZ4ABAV7+Jgm1PklsYT/eo8Hcc=
 github.com/edsrzf/mmap-go v0.0.0-20170320065105-0bce6a688712/go.mod h1:YO35OhQPt3KJa3ryjFM5Bs14WD66h8eGKpfaBNrHW5M=
 github.com/elastic/go-elasticsearch/v6 v6.8.5/go.mod h1:UwaDJsD3rWLM5rKNFzv9hgox93HoX8utj1kxD9aFUcI=
 github.com/elazarl/go-bindata-assetfs v1.0.0 h1:G/bYguwHIzWq9ZoyUQqrjTmJbbYn3j3CKKpKinvZLFk=
@@ -474,6 +476,7 @@ golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191002035440-2ec189313ef0/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200425230154-ff2c4b7c35a0/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20201006153459-a7d1128ccaa0/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
--- a/main.go
查看文件 @bc85917
+++ b/main.go
查看文件 @bc85917
@@ -7,6 +7,7 @@ import (
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/constant"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/redis"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/log"
+	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/port/mqtt"
 	_ "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/constant"
 	_ "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/pg"
@@ -30,7 +31,7 @@ func main() {
 	})
 	log.Logger.AddHook(bw)
 	redis.InitRedis()
-
+	go mqtt.Start()
 	log.Logger.Info("server start!")
 	web.Run()
 }
--- a/pkg/constant/mqtt.go 0 → 100644
查看文件 @bc85917
+++ b/pkg/constant/mqtt.go 0 → 100644
查看文件 @bc85917
+package constant
+
+import "os"
+
+//设备商提供的测试地址
+var MQTT_HOST = "175.24.122.87"
+//内网测试地址
+//var MQTT_HOST = "192.168.100.222"
+var MQTT_PORT = "1883"
+
+var MQTT_USER = ""
+
+var MQTT_PASSWORD = ""
+
+func init(){
+	if os.Getenv("MQTT_HOST") != "" {
+		MQTT_HOST = os.Getenv("MQTT_HOST")
+	}
+	if os.Getenv("MQTT_PORT") != "" {
+		MQTT_PORT = os.Getenv("MQTT_PORT")
+	}
+	if os.Getenv("MQTT_USER") != "" {
+		MQTT_USER = os.Getenv("MQTT_USER")
+	}
+	if os.Getenv("MQTT_PASSWORD") != "" {
+		MQTT_PASSWORD = os.Getenv("MQTT_PASSWORD")
+	}
+}
--- a/pkg/infrastructure/mqtt/produce_client.go 0 → 100644
查看文件 @bc85917
+++ b/pkg/infrastructure/mqtt/produce_client.go 0 → 100644
查看文件 @bc85917
+package mqtt
+
+import (
+	"fmt"
+	pahomqtt "github.com/eclipse/paho.mqtt.golang"
+	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/constant"
+	"time"
+)
+
+type ProduceClient struct {
+	client  pahomqtt.Client
+}
+
+func NewProduceClient() *ProduceClient {
+	return &ProduceClient{}
+}
+
+func (produceClient *ProduceClient) options() *pahomqtt.ClientOptions {
+	opts := pahomqtt.NewClientOptions().AddBroker(fmt.Sprintf("tcp://%v:%v", constant.MQTT_HOST, constant.MQTT_PORT))
+	opts.SetUsername(constant.MQTT_USER)
+	opts.SetPassword(constant.MQTT_PASSWORD)
+	opts.SetKeepAlive(2 * time.Second)
+	opts.SetPingTimeout(1 * time.Second)
+	return opts
+}
+
+func (produceClient *ProduceClient) connect(){
+	opts := produceClient.options()
+	produceClient.client = pahomqtt.NewClient(opts)
+	token := produceClient.client.Connect()
+	token.Wait()
+}
+
+func (produceClient *ProduceClient) Publish(topic string,data interface{}) error {
+	produceClient.connect()
+	token := produceClient.client.Publish(topic,0,false,data)
+	token.Wait()
+	return token.Error()
+}
--- a/pkg/infrastructure/mqtt/subscribe_client.go 0 → 100644
查看文件 @bc85917
+++ b/pkg/infrastructure/mqtt/subscribe_client.go 0 → 100644
查看文件 @bc85917
+package mqtt
+
+import (
+	"fmt"
+	pahomqtt "github.com/eclipse/paho.mqtt.golang"
+	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/constant"
+	"time"
+)
+
+type SubscribeClient struct {
+	topic   string
+	handler pahomqtt.MessageHandler
+	client  pahomqtt.Client
+}
+
+type MessageHandler pahomqtt.MessageHandler
+
+func NewSubscribeClient() *SubscribeClient {
+	return &SubscribeClient{}
+}
+
+func (subscribeClient *SubscribeClient) options() *pahomqtt.ClientOptions {
+	opts := pahomqtt.NewClientOptions().AddBroker(fmt.Sprintf("tcp://%v:%v", constant.MQTT_HOST, constant.MQTT_PORT))
+	opts.SetUsername(constant.MQTT_USER)
+	opts.SetPassword(constant.MQTT_PASSWORD)
+	opts.SetKeepAlive(2 * time.Second)
+	opts.SetPingTimeout(1 * time.Second)
+	return opts
+}
+
+func (subscribeClient *SubscribeClient) Connect() *SubscribeClient{
+	opts := subscribeClient.options()
+	fmt.Println("start connect......")
+	opts.OnConnectionLost = func(c pahomqtt.Client, err error) {
+		fmt.Println("Connect error:", err)
+		for {
+			fmt.Println("reconnect server")
+			token := subscribeClient.client.Connect()
+			token.Wait()
+			fmt.Println("server Connect status:",subscribeClient.client.IsConnectionOpen())
+			if subscribeClient.client.IsConnectionOpen() {
+				break
+			}
+			time.Sleep(3 * time.Second)
+		}
+	}
+	opts.OnConnect = func(c pahomqtt.Client) {
+		c.Subscribe(subscribeClient.topic,0,subscribeClient.handler)
+	}
+	subscribeClient.client = pahomqtt.NewClient(opts)
+	token := subscribeClient.client.Connect()
+	token.Wait()
+	return subscribeClient
+}
+
+func (subscribeClient *SubscribeClient) Subscribe(topic string, messageHandler pahomqtt.MessageHandler){
+	subscribeClient.topic = topic
+	subscribeClient.handler = messageHandler
+	token := subscribeClient.client.Subscribe(topic,0,messageHandler)
+	token.Wait()
+	token.Done()
+}
+
+func StartSubscribe(topic string,handler MessageHandler){
+	defer func() {
+		if err := recover();err != nil {
+			fmt.Println(err)
+			StartSubscribe(topic,handler)
+		}
+	}()
+	fmt.Println("start subscribe...")
+	NewSubscribeClient().Connect().Subscribe(topic,pahomqtt.MessageHandler(handler))
+}
--- a/pkg/port/mqtt/mqtt.go 0 → 100644
查看文件 @bc85917
+++ b/pkg/port/mqtt/mqtt.go 0 → 100644
查看文件 @bc85917
+package mqtt
+
+import (
+	"fmt"
+	pahomqtt "github.com/eclipse/paho.mqtt.golang"
+	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/mqtt"
+	"strconv"
+	"time"
+)
+
+func Start(){
+	mqtt.StartSubscribe("test", func(client pahomqtt.Client, message pahomqtt.Message) {
+		fmt.Println(time.Now(),"Topic:"+message.Topic()+" MessageId:"+ strconv.Itoa(int(message.MessageID()))+" Message:"+ string(message.Payload()))
+	})
+}