FranklinWH-Toolbox/service/repositories/mqtt/fwh_mqtt/fwh_mqtt_client.go

package fwh_mqtt

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"fmt"
	"fwh-toolbox-server/config"
	"fwh-toolbox-server/utils/mqttClient"
	"os"
	"strconv"
	"time"

	"go.uber.org/zap"
)

const (
	fwhClientIdPrefix  = "fwh_xlog_mqtt_client_"
	sleepTimeout       = 5 * time.Minute
	sendMessageTimeout = 1 * time.Minute
)

type FwhMqttResult struct {
	Err  error
	Data *FwhMqttMsg
}

type FwhMqttMsg struct {
	CmdType   uint        `json:"cmdType"`   // 业务类型标识
	Crc       string      `json:"crc"`       // dataArea 数据字符串的CRC
	DataArea  interface{} `json:"dataArea"`  // 数据域
	EquipNo   string      `json:"equipNo"`   // 设备编号 SN
	Len       int         `json:"len"`       // dataArea 数据长度
	Snno      uint        `json:"snno"`      // 序号
	TimeStamp int64       `json:"timeStamp"` // 时间戳
	Type      uint        `json:"type"`      // 设备类型 0：设备网关IBG
}

type MqttMsg struct {
	topic    string // 发送 Topic
	msg      FwhMqttMsg
	subTopic string // 响应 Topic
	resultCh chan FwhMqttResult
	ctx      context.Context
}

type ResultChan struct {
	ch  chan FwhMqttMsg
	ctx context.Context
}

var (
	taskCh = make(chan MqttMsg, 1000)
	// resultChListMap = map[string][]*ResultChan{}
)

var fwhMqttClient *mqttClient.MqttClient

func InitFwhMqttClient(cfg *config.FwhMqttConfig) error {
	// 证书配置
	caCert, err := os.ReadFile(cfg.FwhMqttCaUrl)
	if err != nil {
		panic(err)
	}

	caCertPool := x509.NewCertPool()
	caCertPool.AppendCertsFromPEM(caCert)

	cert, err := tls.LoadX509KeyPair(cfg.FwhMqttCertUrl, cfg.FwhMqttKeyUrl)
	if err != nil {
		panic(err)
	}

	client, err := mqttClient.CreateMqttClient(mqttClient.MqttClientOpt{
		Broker:   cfg.FwhMqttBroker,
		ClientID: fwhClientIdPrefix + strconv.FormatInt(time.Now().Unix(), 10),
		TLS: &tls.Config{
			Certificates: []tls.Certificate{cert},
			RootCAs:      caCertPool,
		},
	})
	if err != nil {
		zap.L().Error("Init FWH MQTT client failed.")
		return err
	}
	fwhMqttClient = client

	zap.L().Info("Init FWH MQTT client success.")

	go fwhMqttWorker(fwhMqttClient)

	return nil
}

func fwhMqttWorker(client *mqttClient.MqttClient) {
	for {
		select {
		case task, ok := <-taskCh:
			if !ok {
				zap.L().Error("Channel closed, worker exiting, recreate.")
				taskCh = make(chan MqttMsg, 1000)
				break
			}

			if err := client.Connect(); err != nil {
				zap.L().Error("FWH MQTT connect failed.", zap.String("msg", fmt.Sprintf("%+v", task)))
				task.resultCh <- FwhMqttResult{
					Err:  fmt.Errorf("FWH MQTT connect failed."),
					Data: nil,
				}
				break
			}

			// 订阅响应消息 - 请求响应无法一一对应，不支持并发请求，改成简单的队列请求
			// if _, ok := resultChListMap[task.subTopic]; !ok {
			// 	resultChListMap[task.subTopic] = []*ResultChan{}
			// 	resultChListMap[task.subTopic] = append(resultChListMap[task.subTopic], &ResultChan{ch: task.resultCh, ctx: task.ctx})
			// 	if err := client.SubscribeMessage(task.subTopic, func(payload []byte) {
			// 		var newList []*ResultChan
			// 		for _, ch := range resultChListMap[task.subTopic] {
			// 			if err := ch.ctx.Err(); err != nil {
			// 				switch err {
			// 				case context.DeadlineExceeded:
			// 					zap.L().Warn("Waiting for FWH MQTT message timeout.", zap.String("topic", task.topic))
			// 				case context.Canceled:
			// 					zap.L().Warn("Waiting for FWH MQTT message cancel.", zap.String("topic", task.topic))
			// 				}
			// 			} else {
			// 				var fwhMqttMsg FwhMqttMsg
			// 				err := json.Unmarshal(payload, &fwhMqttMsg)
			// 				if err != nil {
			// 					zap.L().Error("Result Unmarshal failed.(FwhMqttMsg)", zap.String("topic", task.topic), zap.String("payload", string(payload)), zap.String("err", err.Error()))
			// 					newList = append(newList, ch)
			// 					continue
			// 				}

			// 				if task.msg.Snno != fwhMqttMsg.Snno {
			// 					newList = append(newList, ch)
			// 					continue
			// 				}

			// 				ch.ch <- fwhMqttMsg
			// 			}
			// 		}
			// 		resultChListMap[task.subTopic] = newList
			// 	}); err != nil {
			// 		zap.L().Error("FWH MQTT subscribe message failed.", zap.String("msg", fmt.Sprintf("%+v", task)))
			// 		break
			// 	}
			// } else {
			// 	resultChListMap[task.subTopic] = append(resultChListMap[task.subTopic], &ResultChan{ch: task.resultCh, ctx: task.ctx})
			// }

			resultCh := make(chan FwhMqttResult)

			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{
						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
			}

			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{
					Err:  err,
					Data: nil,
				}
				break
			}

			if err := 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{
					Err:  err,
					Data: nil,
				}
				break
			}

			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,
				}
			}

		case <-time.After(sleepTimeout):
			if client.IsConnected() {
				zap.L().Info("FWH MQTT no task received within timeout, disconnect.")
				client.Disconnect()
			}
		}
	}
}

func sendMessage(ctx context.Context, topic string, msg FwhMqttMsg, subTopic string) (*FwhMqttMsg, error) {
	if fwhMqttClient == nil {
		return nil, fmt.Errorf("Fwh Mqtt Client Uninitialized.")
	}

	ctxSend, cancel := context.WithTimeout(ctx, sendMessageTimeout)
	defer cancel()

	taskMsg := MqttMsg{
		topic:    topic,
		msg:      msg,
		subTopic: subTopic,
		resultCh: make(chan FwhMqttResult, 1),
		ctx:      ctxSend,
	}
	taskCh <- taskMsg

	for {
		select {
		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.")
			}

			if result.Err != nil {
				return nil, result.Err
			}

			return result.Data, nil
		case <-ctxSend.Done():
			zap.L().Info("Result received timeout.", zap.String("topic", topic), zap.String("msg", fmt.Sprintf("%+v", msg)))
			return nil, fmt.Errorf("Result received timeout.")
		}
	}
}