feat:优化FwhMqtt收发模块

service/.vscode/settings.json

@@ -27,7 +27,8 @@
     "imei",
     "imsi",
     "msisdn",
-    "acceleronix"
+    "acceleronix",
+    "xlogaddr"
   ],
   "editor.tabSize": 2
 }

service/repositories/mqtt/fwh_mqtt/fwh_mqtt_client.go

@@ -38,11 +38,12 @@ type FwhMqttMsg struct {
 }
 
 type MqttMsg struct {
-	topic    string // 发送 Topic
-	msg      FwhMqttMsg
-	subTopic string // 响应 Topic
-	resultCh chan FwhMqttResult
-	ctx      context.Context
+	Topic      string // 发送 Topic
+	Msg        FwhMqttMsg
+	SubTopic   *string // 响应 Topic
+	SubCmdType *uint
+	ResultCh   chan FwhMqttResult
+	Ctx        context.Context
 }
 
 type ResultChan struct {
@@ -50,12 +51,15 @@ type ResultChan struct {
 	ctx context.Context
 }
 
-var (
-	taskCh = make(chan MqttMsg, 1000)
-	// resultChListMap = map[string][]*ResultChan{}
-)
+type FwhMqttClient struct {
+	Client *mqttClient.MqttClient
+	TaskCh chan MqttMsg
+}
 
-var fwhMqttClient *mqttClient.MqttClient
+var fwhMqttClient = FwhMqttClient{
+	Client: nil,
+	TaskCh: make(chan MqttMsg, 1000),
+}
 
 func InitFwhMqttClient(cfg *config.FwhMqttConfig) error {
 	// 证书配置
@@ -84,28 +88,28 @@ func InitFwhMqttClient(cfg *config.FwhMqttConfig) error {
 		zap.L().Error("Init FWH MQTT client failed.")
 		return err
 	}
-	fwhMqttClient = client
+	fwhMqttClient.Client = client
 
 	zap.L().Info("Init FWH MQTT client success.")
 
-	go fwhMqttWorker(fwhMqttClient)
+	go fwhMqttClient.fwhMqttWorker()
 
 	return nil
 }
 
-func fwhMqttWorker(client *mqttClient.MqttClient) {
+func (c *FwhMqttClient) fwhMqttWorker() {
 	for {
 		select {
-		case task, ok := <-taskCh:
+		case task, ok := <-c.TaskCh:
 			if !ok {
 				zap.L().Error("Channel closed, worker exiting, recreate.")
-				taskCh = make(chan MqttMsg, 1000)
+				c.TaskCh = make(chan MqttMsg, 1000)
 				break
 			}
 
-			if err := client.Connect(); err != nil {
+			if err := c.Client.Connect(); err != nil {
 				zap.L().Error("FWH MQTT connect failed.", zap.String("msg", fmt.Sprintf("%+v", task)))
-				task.resultCh <- FwhMqttResult{
+				task.ResultCh <- FwhMqttResult{
 					Err:  fmt.Errorf("FWH MQTT connect failed."),
 					Data: nil,
 				}
@@ -152,80 +156,98 @@ func fwhMqttWorker(client *mqttClient.MqttClient) {
 			// 	resultChListMap[task.subTopic] = append(resultChListMap[task.subTopic], &ResultChan{ch: task.resultCh, ctx: task.ctx})
 			// }
 
-			resultCh := make(chan FwhMqttResult)
+			var resultCh chan FwhMqttResult
+			if task.SubTopic != nil {
+				resultCh = make(chan FwhMqttResult)
+				if err := c.Client.SubscribeMessage(*task.SubTopic, func(payload []byte) {
+					var fwhMqttMsg FwhMqttMsg
+					if err := json.Unmarshal(payload, &fwhMqttMsg); err != nil {
+						zap.L().Error("Result Unmarshal failed.(FwhMqttMsg)", zap.String("topic", task.Topic), zap.String("payload", string(payload)), zap.String("err", err.Error()))
+						// resultCh <- FwhMqttResult{
+						// 	Err:  err,
+						// 	Data: nil,
+						// }
+						return
+					}
 
-			if err := client.SubscribeMessage(task.subTopic, func(payload []byte) {
-				var fwhMqttMsg FwhMqttMsg
-				if err := json.Unmarshal(payload, &fwhMqttMsg); err != nil {
-					zap.L().Error("Result Unmarshal failed.(FwhMqttMsg)", zap.String("topic", task.topic), zap.String("payload", string(payload)), zap.String("err", err.Error()))
-					resultCh <- FwhMqttResult{
+					if task.SubCmdType != nil {
+						if fwhMqttMsg.CmdType == *task.SubCmdType {
+							resultCh <- FwhMqttResult{
+								Err:  nil,
+								Data: &fwhMqttMsg,
+							}
+						}
+					} else {
+						resultCh <- FwhMqttResult{
+							Err:  nil,
+							Data: &fwhMqttMsg,
+						}
+					}
+				}); err != nil {
+					zap.L().Error("FWH MQTT subscribe message failed.", zap.String("msg", fmt.Sprintf("%+v", task)), zap.String("err", err.Error()))
+					task.ResultCh <- FwhMqttResult{
 						Err:  err,
 						Data: nil,
 					}
-					return
-				}
-
-				resultCh <- FwhMqttResult{
-					Err:  nil,
-					Data: &fwhMqttMsg,
-				}
-			}); err != nil {
-				zap.L().Error("FWH MQTT subscribe message failed.", zap.String("msg", fmt.Sprintf("%+v", task)), zap.String("err", err.Error()))
-				task.resultCh <- FwhMqttResult{
-					Err:  err,
-					Data: nil,
+					break
 				}
-				break
 			}
 
-			msgByte, err := json.Marshal(task.msg)
+			msgByte, err := json.Marshal(task.Msg)
 			if err != nil {
 				zap.L().Error("FWH MQTT message marshal failed.", zap.String("msg", fmt.Sprintf("%+v", task)), zap.String("err", err.Error()))
-				task.resultCh <- FwhMqttResult{
+				task.ResultCh <- FwhMqttResult{
 					Err:  err,
 					Data: nil,
 				}
 				break
 			}
 
-			if err := client.SendMessageOnce(task.topic, msgByte); err != nil {
+			if err := c.Client.SendMessageOnce(task.Topic, msgByte); err != nil {
 				zap.L().Error("FWH MQTT send message failed.", zap.String("msg", fmt.Sprintf("%+v", task)), zap.String("err", err.Error()))
-				task.resultCh <- FwhMqttResult{
+				task.ResultCh <- FwhMqttResult{
 					Err:  err,
 					Data: nil,
 				}
 				break
 			}
 
-			select {
-			case res, ok := <-resultCh:
-				if !ok {
-					task.resultCh <- FwhMqttResult{
-						Err:  fmt.Errorf("FWH MQTT received message channel closed."),
+			if resultCh != nil {
+				select {
+				case res, ok := <-resultCh:
+					if !ok {
+						task.ResultCh <- FwhMqttResult{
+							Err:  fmt.Errorf("FWH MQTT received message channel closed."),
+							Data: nil,
+						}
+						break
+					}
+
+					task.ResultCh <- res
+				case <-task.Ctx.Done():
+					task.ResultCh <- FwhMqttResult{
+						Err:  fmt.Errorf("FWH MQTT received message timeout."),
 						Data: nil,
 					}
-					break
 				}
-
-				task.resultCh <- res
-			case <-task.ctx.Done():
-				task.resultCh <- FwhMqttResult{
-					Err:  fmt.Errorf("FWH MQTT received message timeout."),
+			} else {
+				task.ResultCh <- FwhMqttResult{
+					Err:  nil,
 					Data: nil,
 				}
 			}
 
 		case <-time.After(sleepTimeout):
-			if client.IsConnected() {
+			if c.Client.IsConnected() {
 				zap.L().Info("FWH MQTT no task received within timeout, disconnect.")
-				client.Disconnect()
+				c.Client.Disconnect()
 			}
 		}
 	}
 }
 
-func sendMessage(ctx context.Context, topic string, msg FwhMqttMsg, subTopic string) (*FwhMqttMsg, error) {
-	if fwhMqttClient == nil {
+func (c *FwhMqttClient) sendMessage(ctx context.Context, topic string, msg FwhMqttMsg, subTopic *string, subCmdType *uint) (*FwhMqttMsg, error) {
+	if c.Client == nil {
 		return nil, fmt.Errorf("Fwh Mqtt Client Uninitialized.")
 	}
 
@@ -233,17 +255,18 @@ func sendMessage(ctx context.Context, topic string, msg FwhMqttMsg, subTopic str
 	defer cancel()
 
 	taskMsg := MqttMsg{
-		topic:    topic,
-		msg:      msg,
-		subTopic: subTopic,
-		resultCh: make(chan FwhMqttResult, 1),
-		ctx:      ctxSend,
+		Topic:      topic,
+		Msg:        msg,
+		SubTopic:   subTopic,
+		SubCmdType: subCmdType,
+		ResultCh:   make(chan FwhMqttResult, 1),
+		Ctx:        ctxSend,
 	}
-	taskCh <- taskMsg
+	c.TaskCh <- taskMsg
 
 	for {
 		select {
-		case result, ok := <-taskMsg.resultCh:
+		case result, ok := <-taskMsg.ResultCh:
 			if !ok {
 				zap.L().Error("Result channel closed, worker exiting, recreate.", zap.String("topic", topic), zap.String("msg", fmt.Sprintf("%+v", msg)))
 				return nil, fmt.Errorf("Result channel closed, worker exiting, recreate.")
@@ -260,3 +283,15 @@ func sendMessage(ctx context.Context, topic string, msg FwhMqttMsg, subTopic str
 		}
 	}
 }
+
+// 仅发送不等待响应结果
+func (c *FwhMqttClient) sendMessageOnce(ctx context.Context, topic string, msg FwhMqttMsg) error {
+	_, err := c.sendMessage(ctx, topic, msg, nil, nil)
+	return err
+}
+
+// 发送消息后，等待响应结果
+func (c *FwhMqttClient) sendMessageWait(ctx context.Context, topic string, msg FwhMqttMsg, subTopic string, subCmdType uint) (*FwhMqttMsg, error) {
+	res, err := c.sendMessage(ctx, topic, msg, &subTopic, &subCmdType)
+	return res, err
+}

service/repositories/mqtt/fwh_mqtt/fwh_mqtt_debug.go

@@ -20,7 +20,7 @@ func EnableXlog(ctx context.Context, sn string) (string, error) {
 		return "", err
 	}
 
-	res, err := sendMessage(ctx, "fwh/s2c/debug/"+sn, FwhMqttMsg{
+	res, err := fwhMqttClient.sendMessageWait(ctx, "fwh/s2c/debug/"+sn, FwhMqttMsg{
 		CmdType:   2000, // xlog请求
 		Type:      0,    // IBG
 		Snno:      123,
@@ -29,7 +29,7 @@ func EnableXlog(ctx context.Context, sn string) (string, error) {
 		DataArea:  dataArea,
 		Len:       len(dataStr),
 		Crc:       strings.ToUpper(fmt.Sprintf("%08X", crc32.ChecksumIEEE(dataStr))),
-	}, "fwh/c2s/debug/"+sn)
+	}, "fwh/c2s/debug/"+sn, 2001)
 	if err != nil {
 		return "", err
 	}

service/repositories/repositories.go

@@ -195,4 +195,9 @@ var GetAutoUpdateConfigFileInfoByPage = redis.GetAutoUpdateConfigFileInfoByPage
 /*************** MQTT ***************/
 // FwhMqtt
 var FwhMqttInit = fwh_mqtt.InitFwhMqttClient
+
+// FwhMqtt - Xlog
 var FwhMqttEnableXlog = fwh_mqtt.EnableXlog
+
+// FwhMqtt - Fault Record
+var FwhMqttFaultRecordChargeChannel = fwh_mqtt.ChargeChannel