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

	topicPrefixS2C = "fwh/s2c/"
	topicPrefixC2S = "fwh/c2s/"
)

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
	ResTopic   *string // 响应 Topic
	ResCmdType *uint
	ResultCh   chan FwhMqttResult
	Ctx        context.Context
}

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

type FwhMqttClient struct {
	Client *mqttClient.MqttClient
	TaskCh chan MqttMsg
}

var fwhMqttClient = FwhMqttClient{
	Client: nil,
	TaskCh: make(chan MqttMsg, 1000),
}

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 = client

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

	go fwhMqttClient.fwhMqttWorker()

	return nil
}

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

			if !c.Client.IsConnected() {
				if err := c.Client.Connect(); err != nil {
					zap.L().Error("FWH MQTT connect failed.", zap.String("msg", fmt.Sprintf("%+v", task)), zap.String("err", err.Error()))
					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})
			// }

			var resultCh chan FwhMqttResult
			if task.ResTopic != nil {
				resultCh = make(chan FwhMqttResult)
				if err := c.Client.SubscribeMessage(*task.ResTopic, 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 task.ResCmdType != nil {
						if fwhMqttMsg.CmdType == *task.ResCmdType {
							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,
					}
					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 := 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{
					Err:  err,
					Data: nil,
				}
				break
			}

			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,
					}
				}
			} else {
				task.ResultCh <- FwhMqttResult{
					Err:  nil,
					Data: nil,
				}
			}

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

func (c *FwhMqttClient) sendMessage(ctx context.Context, topicName string, msg FwhMqttMsg, resTopicName *string, resCmdType *uint) (*FwhMqttMsg, error) {
	if c.Client == nil {
		return nil, fmt.Errorf("Fwh Mqtt Client Uninitialized.")
	}

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

	topic := topicPrefixS2C + topicName
	var resTopic *string
	if resTopicName != nil {
		resTopicStr := topicPrefixC2S + *resTopicName
		resTopic = &resTopicStr
	}

	taskMsg := MqttMsg{
		Topic:      topic,
		Msg:        msg,
		ResTopic:   resTopic,
		ResCmdType: resCmdType,
		ResultCh:   make(chan FwhMqttResult, 1),
		Ctx:        ctxSend,
	}
	c.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.")
		}
	}
}

// 仅发送不等待响应结果
func (c *FwhMqttClient) sendMessageOnce(ctx context.Context, topicName string, msg FwhMqttMsg) error {
	_, err := c.sendMessage(ctx, topicName, msg, nil, nil)
	return err
}

// 发送消息后，等待响应结果
func (c *FwhMqttClient) sendMessageWait(ctx context.Context, topicName string, msg FwhMqttMsg, resTopicName string, resCmdType uint) (*FwhMqttMsg, error) {
	res, err := c.sendMessage(ctx, topicName, msg, &resTopicName, &resCmdType)
	return res, err
}