[优化]mqtt客户端采用emqx接口，根据设备订阅的me topic进行返回

config.yml

@@ -61,10 +61,10 @@ taos:
   config: ""
 
 mqtt:
-  broker: 127.0.0.1:1883
+  broker: http://127.0.0.1:18083/api
   qos: 1
-  username: pandax
-  password: pandax
+  username: dbda318b31319d49
+  password: wwqPbYu9BCsIFgW3m0aICmRrvxUHcIi44AQuG24wQARE
 
 casbin:
   model-path: './resource/rbac_model.conf'

iothub/client/mqttclient/mqtt.go

@@ -1,79 +0,0 @@
-package mqttclient
-
-import (
-	"errors"
-	"fmt"
-	mqtt "github.com/eclipse/paho.mqtt.golang"
-	"github.com/sirupsen/logrus"
-	"time"
-)
-
-const DefaultDownStreamClientId = `@panda.iothub.internal.clientId`
-
-var MqttClient *IothubMqttClient
-
-type IothubMqttClient struct {
-	Client mqtt.Client
-}
-
-var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
-	logrus.Infof("Connected")
-}
-
-var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
-	logrus.Infof("Connect lost: %v", err)
-}
-
-func InitMqtt(broker, username, password string) {
-	server := fmt.Sprintf("tcp://%s", broker)
-	client := GetMqttClinent(server, username, password)
-	MqttClient = &IothubMqttClient{
-		Client: client,
-	}
-}
-
-func GetMqttClinent(server, username, password string) mqtt.Client {
-	opts := mqtt.NewClientOptions().AddBroker(server)
-	time.Now().Unix()
-	// 默认下行ID iothub会过滤掉改Id
-	opts.SetClientID(DefaultDownStreamClientId)
-	if username != "" {
-		opts.SetUsername(username)
-	}
-	if password != "" {
-		opts.SetPassword(password)
-	}
-	opts.OnConnect = connectHandler
-	opts.OnConnectionLost = connectLostHandler
-	client := mqtt.NewClient(opts)
-	if token := client.Connect(); token.Wait() && token.Error() != nil {
-		panic(token.Error())
-	}
-	return client
-}
-
-// 订阅
-func (imc *IothubMqttClient) Sub(topic string, qos byte, handler mqtt.MessageHandler) {
-	if token := imc.Client.Subscribe(topic, qos, handler); token.Wait() && token.Error() != nil {
-		logrus.Infof(token.Error().Error())
-	}
-}
-
-// 取消订阅
-func (imc *IothubMqttClient) UnSub(topic string) {
-	if token := imc.Client.Unsubscribe(topic); token.Wait() && token.Error() != nil {
-		logrus.Infof(token.Error().Error())
-	}
-}
-
-// 发布
-func (imc *IothubMqttClient) Pub(topic string, qos byte, payload string) error {
-	token := imc.Client.Publish(topic, qos, false, payload)
-	if token.WaitTimeout(2*time.Second) == false {
-		return errors.New("推送消息超时")
-	}
-	if token.Error() != nil {
-		return token.Error()
-	}
-	return nil
-}

iothub/client/mqttclient/mqtt_api.go

@@ -0,0 +1,104 @@
+package mqttclient
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"pandax/pkg/global"
+)
+
+// key 设备id，value MQTT的clientID
+var MqttClient = make(map[string]string)
+
+const ClientsInfo string = "client"
+const SubscribeTopicsInfo string = "subscribe"
+
+// GetEmqInfo 获取emqx信息，包括连接信息，订阅信息
+func GetEmqInfo(infoType string) ([]map[string]interface{}, error) {
+	var url string
+	if infoType == ClientsInfo {
+		url = fmt.Sprintf("%s/v5/clients?_page=1&_limit=100000", global.Conf.Mqtt.Broker)
+	} else if infoType == SubscribeTopicsInfo {
+		url = fmt.Sprintf("%s/v5/subscriptions?_page=1&_limit=100000", global.Conf.Mqtt.Broker)
+	} else {
+		return nil, errors.New("invalid infoType")
+	}
+
+	req, err := http.NewRequest(http.MethodGet, url, nil)
+	if nil != err {
+		return nil, err
+	}
+	req.SetBasicAuth(global.Conf.Mqtt.Username, global.Conf.Mqtt.Password)
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+
+	defer resp.Body.Close()
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	global.Log.Debug("receive resp, ", string(body))
+	if resp.StatusCode != 200 {
+		return nil, errors.New(resp.Status)
+	}
+
+	var result interface{}
+	if err = json.Unmarshal(body, &result); nil != err {
+		global.Log.Error("body Unmarshal error", err)
+		return nil, err
+	}
+	res, ok := result.(map[string]interface{})
+	if !ok {
+		return nil, errors.New("result error")
+	}
+	data := res["data"].([]map[string]interface{})
+	return data, nil
+}
+
+// Publish 推送信息
+func Publish(topic, clientId string, payload interface{}) error {
+	global.Log.Debugf("send data to clientId: %s, topic:%s, payload: %v", clientId, topic, payload)
+	url := fmt.Sprintf("%s/v5/publish", global.Conf.Mqtt.Broker)
+	pubData := map[string]interface{}{
+		"topic":    topic,
+		"payload":  payload,
+		"qos":      global.Conf.Mqtt.Qos,
+		"retain":   false,
+		"clientid": clientId,
+	}
+	data, err := json.Marshal(pubData)
+	if nil != err {
+		global.Log.Error("error ", err)
+		return err
+	}
+	req, err := http.NewRequest(http.MethodPost, url, bytes.NewBuffer(data))
+	if nil != err {
+		return err
+	}
+	req.SetBasicAuth(global.Conf.Mqtt.Username, global.Conf.Mqtt.Password)
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		global.Log.Errorf("Publish.DefaultClient.Do data=%s error=%s", string(data), err.Error())
+		return err
+	}
+	defer resp.Body.Close()
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		global.Log.Error("error ReadAll", err)
+		return err
+	}
+	global.Log.Debug("receive resp, ", string(body))
+	if resp.StatusCode != 200 {
+		global.Log.Error("bad status ", resp.StatusCode)
+		return errors.New(resp.Status)
+	}
+	return nil
+}

iothub/client/mqttclient/rpc.go

@@ -1,9 +1,7 @@
 package mqttclient
 
 import (
-	"errors"
 	"fmt"
-	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"math/rand"
 	"time"
 )
@@ -19,52 +17,20 @@ const (
 )
 
 type RpcRequest struct {
-	Client    *IothubMqttClient
 	RequestId int
 	Mode      string //单向、双向 单项只发送不等待响应  双向需要等到响应
 	Timeout   int    // 设置双向时，等待的超时时间
 }
 
 // RequestCmd 下发指令
-func (rpc RpcRequest) RequestCmd(rpcPayload string) (respPayload string, err error) {
+func (rpc RpcRequest) RequestCmd(deviceId, rpcPayload string) error {
 	topic := fmt.Sprintf(RpcReqTopic, rpc.RequestId)
-	err = rpc.Client.Pub(topic, 0, rpcPayload)
-	if err != nil {
-		return "", err
-	}
-	if rpc.Mode == "" || rpc.Mode == SingleMode {
-		return "", nil
-	}
-	// 双向才会执行
-	repsChan := make(chan string)
-	respTopic := fmt.Sprintf(RpcRespTopic, rpc.RequestId)
-	// 订阅回调
-	mqMessagePubHandler := func(client mqtt.Client, msg mqtt.Message) {
-		if repsChan != nil {
-			repsChan <- string(msg.Payload())
-		}
-	}
-	rpc.Client.Sub(respTopic, 0, mqMessagePubHandler)
-	if rpc.Timeout == 0 {
-		rpc.Timeout = 20
-	}
-	defer func() {
-		close(repsChan)
-		rpc.Client.UnSub(respTopic)
-	}()
-	for {
-		select {
-		case <-time.After(time.Duration(rpc.Timeout) * time.Second):
-			return "", errors.New("设备指令响应超时")
-		case resp := <-repsChan:
-			return resp, nil
-		}
-	}
+	return Publish(topic, MqttClient[deviceId], rpcPayload)
 }
 
-func (rpc RpcRequest) Pub(reqPayload string) error {
+func (rpc RpcRequest) Pub(deviceId, reqPayload string) error {
 	topic := fmt.Sprintf(RpcRespTopic, rpc.RequestId)
-	return rpc.Client.Pub(topic, 0, reqPayload)
+	return Publish(topic, MqttClient[deviceId], reqPayload)
 }
 
 func (rpc *RpcRequest) GetRequestId() {

iothub/server/emqxserver/hook.go

@@ -32,8 +32,6 @@ func InitEmqxHook(addr string, hs *hook_message_work.HookService) {
 	} else {
 		global.Log.Infof("IOTHUB HOOK Start SUCCESS,Grpc Server listen: %s", addr)
 	}
-	// 初始化 MQTT客户端
-	mqttclient.InitMqtt(global.Conf.Mqtt.Broker, global.Conf.Mqtt.Username, global.Conf.Mqtt.Password)
 }
 
 func NewHookGrpcService(hs *hook_message_work.HookService) *HookGrpcService {
@@ -82,12 +80,14 @@ func (s *HookGrpcService) OnClientConnack(ctx context.Context, in *exhook2.Clien
 
 func (s *HookGrpcService) OnClientConnected(ctx context.Context, in *exhook2.ClientConnectedRequest) (*exhook2.EmptySuccess, error) {
 	global.Log.Info(fmt.Sprintf("Client %s Connected ", in.Clientinfo.GetNode()))
-	if in.Clientinfo.Clientid == mqttclient.DefaultDownStreamClientId {
-		return &exhook2.EmptySuccess{}, nil
-	}
 	token := netbase.GetUserName(in.Clientinfo)
 	etoken := &global_model.DeviceAuth{}
-	etoken.GetDeviceToken(token)
+	err := etoken.GetDeviceToken(token)
+	if err != nil {
+		return nil, err
+	}
+	//添加连接ID
+	mqttclient.MqttClient[etoken.DeviceId] = in.Clientinfo.Clientid
 	data := netbase.CreateConnectionInfo(message.ConnectMes, "mqtt", in.Clientinfo.Clientid, in.Clientinfo.Peerhost, etoken)
 	s.HookService.MessageCh <- data
 	return &exhook2.EmptySuccess{}, nil
@@ -96,14 +96,13 @@ func (s *HookGrpcService) OnClientConnected(ctx context.Context, in *exhook2.Cli
 func (s *HookGrpcService) OnClientDisconnected(ctx context.Context, in *exhook2.ClientDisconnectedRequest) (*exhook2.EmptySuccess, error) {
 	global.Log.Info(fmt.Sprintf("%s断开连接", in.Clientinfo.Username))
 	token := netbase.GetUserName(in.Clientinfo)
-	if in.Clientinfo.Clientid == mqttclient.DefaultDownStreamClientId {
-		return &exhook2.EmptySuccess{}, nil
-	}
 	etoken := &global_model.DeviceAuth{}
 	err := etoken.GetDeviceToken(token)
 	if err != nil {
 		return nil, err
 	}
+	//删除连接ID
+	delete(mqttclient.MqttClient, etoken.DeviceId)
 	data := netbase.CreateConnectionInfo(message.DisConnectMes, "mqtt", in.Clientinfo.Clientid, in.Clientinfo.Peerhost, etoken)
 	s.HookService.MessageCh <- data
 	return &exhook2.EmptySuccess{}, nil
@@ -177,10 +176,6 @@ func (s *HookGrpcService) OnMessagePublish(ctx context.Context, in *exhook2.Mess
 	res.Type = exhook2.ValuedResponse_STOP_AND_RETURN
 	res.Value = &exhook2.ValuedResponse_BoolResult{BoolResult: false}
 
-	if in.Message.From == mqttclient.DefaultDownStreamClientId {
-		res.Value = &exhook2.ValuedResponse_BoolResult{BoolResult: true}
-		return res, nil
-	}
 	etoken := &global_model.DeviceAuth{}
 	etoken.GetDeviceToken(in.Message.Headers["username"])
 	// 获取topic类型

pkg/rule_engine/nodes/action_rpc_request_from_device_node.go

@@ -52,8 +52,9 @@ func (n *rpcRequestFromDeviceNode) Handle(msg *message.Message) error {
 	if msg.Metadata.GetValue("deviceProtocol") != nil && msg.Metadata.GetValue("deviceProtocol").(string) != "" {
 		deviceProtocol = msg.Metadata.GetValue("deviceProtocol").(string)
 	}
+	deviceId := msg.Metadata.GetValue("deviceId").(string)
 	if deviceProtocol == global.MQTTProtocol {
-		rpc := &mqttclient.RpcRequest{Client: mqttclient.MqttClient}
+		rpc := &mqttclient.RpcRequest{}
 		RequestId := n.RequestId
 		if RequestId == 0 {
 			if msg.Metadata.GetValue("requestId") == nil {
@@ -64,10 +65,9 @@ func (n *rpcRequestFromDeviceNode) Handle(msg *message.Message) error {
 		} else {
 			rpc.RequestId = RequestId
 		}
-		err = rpc.Pub(result)
+		err = rpc.Pub(deviceId, result)
 	}
 	if deviceProtocol == global.TCPProtocol {
-		deviceId := msg.Metadata.GetValue("deviceId").(string)
 		err = tcpclient.Send(deviceId, result)
 	}
 	if err != nil {

pkg/rule_engine/nodes/action_rpc_request_to_device_node.go

@@ -50,13 +50,13 @@ func (n *rpcRequestToDeviceNode) Handle(msg *message.Message) error {
 		deviceProtocol = msg.Metadata.GetValue("deviceProtocol").(string)
 	}
 	var err error
+	deviceId := msg.Metadata.GetValue("deviceId").(string)
 	if deviceProtocol == global.MQTTProtocol {
-		var rpc = &mqttclient.RpcRequest{Client: mqttclient.MqttClient, Mode: mode, Timeout: n.Timeout}
+		var rpc = &mqttclient.RpcRequest{Mode: mode, Timeout: n.Timeout}
 		rpc.GetRequestId()
-		_, err = rpc.RequestCmd(string(payload))
+		err = rpc.RequestCmd(deviceId, string(payload))
 	}
 	if deviceProtocol == global.TCPProtocol {
-		deviceId := msg.Metadata.GetValue("deviceId").(string)
 		err = tcpclient.Send(deviceId, string(payload))
 	}
 	if err != nil {