91 lines
2.3 KiB
Go
91 lines
2.3 KiB
Go
|
package service
|
||
|
|
|
||
|
|
import (
|
||
|
|
"errors"
|
||
|
|
"fmt"
|
||
|
|
"time"
|
||
|
|
"encoding/json"
|
||
|
|
"zhaoCaiMao/vo"
|
||
|
|
|
||
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
||
|
|
"github.com/gogf/gf/v2/frame/g"
|
||
|
|
"github.com/gogf/gf/v2/util/guid"
|
||
|
|
)
|
||
|
|
|
||
|
|
type MQTT struct {
|
||
|
|
client mqtt.Client
|
||
|
|
}
|
||
|
|
|
||
|
|
var MQTTService MQTT
|
||
|
|
|
||
|
|
func (t *MQTT) InitMQTT() error {
|
||
|
|
ip := g.Cfg().MustGet(nil, "MQTT.ip") //订单MQTT
|
||
|
|
port := g.Cfg().MustGet(nil, "MQTT.port")
|
||
|
|
|
||
|
|
/*
|
||
|
|
user := g.Cfg().MustGet(nil, "MQTT.user")
|
||
|
|
passwd := g.Cfg().MustGet(nil, "MQTT.passwd")
|
||
|
|
*/
|
||
|
|
opts := mqtt.NewClientOptions()
|
||
|
|
mqttAddress := fmt.Sprintf("mqtt://%s:%d", ip.String(), port.Int())
|
||
|
|
g.Log().Line().Print(nil, "订单MQTT连接地址:", mqttAddress)
|
||
|
|
opts.AddBroker(mqttAddress)
|
||
|
|
opts.SetKeepAlive(time.Hour * 24)
|
||
|
|
opts.SetPingTimeout(time.Second * 30)
|
||
|
|
opts.SetClientID(guid.S())
|
||
|
|
//opts.SetUsername(user.String()) //使用用户名和密码
|
||
|
|
//opts.SetPassword(passwd.String())
|
||
|
|
opts.SetDefaultPublishHandler(t.reciveOrderMQTTHandler) // 设置消息回调处理函数
|
||
|
|
//opts.OnConnect = connectHandler
|
||
|
|
opts.OnConnectionLost = t.connectLostHandler
|
||
|
|
t.client = mqtt.NewClient(opts)
|
||
|
|
token := t.client.Connect()
|
||
|
|
if token.Wait() && token.Error() != nil {
|
||
|
|
err := errors.New("订单MQTT初始化失败:" + token.Error().Error())
|
||
|
|
g.Log().Line().Error(nil, err)
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
payTopic := g.Cfg().MustGet(nil, "MQTT.topic").String()
|
||
|
|
//订阅支付订单消息
|
||
|
|
token = t.client.Subscribe(payTopic, 2, nil)
|
||
|
|
if !token.Wait() {
|
||
|
|
return token.Error()
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
func (t *MQTT) reciveOrderMQTTHandler(client mqtt.Client, msg mqtt.Message) {
|
||
|
|
logMsg := fmt.Sprintf("Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())
|
||
|
|
fmt.Println(logMsg)
|
||
|
|
g.Log().Line().Print(nil, logMsg)
|
||
|
|
mqttOrder := vo.MQTTOrder{}
|
||
|
|
err := json.Unmarshal(msg.Payload(),&mqttOrder)
|
||
|
|
if err != nil {
|
||
|
|
g.Log().Line().Print(nil, err)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
t.createMsgToCat(&mqttOrder)
|
||
|
|
|
||
|
|
|
||
|
|
//to-----发送到招财猫设备
|
||
|
|
|
||
|
|
}
|
||
|
|
|
||
|
|
func (t *MQTT) connectLostHandler(client mqtt.Client, err error) {
|
||
|
|
token := t.client.Connect()
|
||
|
|
g.Log().Line().Print(nil, "订单MQTT 重新连接成功")
|
||
|
|
token.Wait()
|
||
|
|
}
|
||
|
|
//组织发送招财猫的报文
|
||
|
|
func (t *MQTT) createMsgToCat(mqttOrder *vo.MQTTOrder) {
|
||
|
|
mqttOrder.
|
||
|
|
mqttDevice := vo.MQTTDevice{}
|
||
|
|
mqttDevice.msgId = mqttOrder.msgId
|
||
|
|
if mqttOrder.Channel == 1 { //微信渠道
|
||
|
|
mqttdevice.Type = 1
|
||
|
|
}else { //支付宝渠道
|
||
|
|
mqttdevice.Type = 2
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
}
|