feat:把fwh_mqtt封装成一个通用的操作模块，生产服和测试服同时使用相同的模块

service/dto/faultRecordDTO.go

@@ -12,6 +12,7 @@ type FaultRecordSubParam struct {
 }
 
 type FaultRecordParamInfo struct {
-	Sn   string                `json:"sn"` // aGate SN
+	Env  string                `json:"env"` // 'prod'（默认）or 'test'
+	Sn   string                `json:"sn"`  // aGate SN
 	List []FaultRecordSubParam `json:"list"`
 }

service/main.go

@@ -94,9 +94,14 @@ func main() {
 	// 初始化Casdoor
 	casdoor.InitAuthConfig()
 
-	// 初始化FWH MQTT客户端
-	repositories.FwhMqttInit(&cfg.FwhMqtt)
-	// repositories.FwhMqttFaultRecordChargeChannel(context.Background(), "10060005A03X25160385")
+	// 初始化FWH MQTT客户端（正式服、测试服各一个实例，单个环境失败不影响服务启动）
+	if err := repositories.FwhMqttInit(repositories.FwhMqttEnvProd, &cfg.FwhMqtt); err != nil {
+		zap.L().Error("FWH MQTT init failed.", zap.String("env", "prod"), zap.String("err", err.Error()))
+	}
+	if err := repositories.FwhMqttInit(repositories.FwhMqttEnvTest, &cfg.FwhTestMqtt); err != nil {
+		zap.L().Error("FWH MQTT init failed.", zap.String("env", "test"), zap.String("err", err.Error()))
+	}
+	// repositories.FwhMqttFaultRecordChargeChannel(context.Background(), repositories.FwhMqttEnvProd, "10060005A03X25160385")
 
 	app := fiber.New()
 	app.Use(func(c *fiber.Ctx) error {

service/repositories/mqtt/fwh_mqtt/fwh_mqtt_client.go

@@ -10,6 +10,7 @@ import (
 	"fwh-toolbox-server/utils/mqttClient"
 	"os"
 	"strconv"
+	"sync"
 	"time"
 
 	"go.uber.org/zap"
@@ -59,16 +60,43 @@ type FwhMqttClient struct {
 	TaskCh chan MqttMsg
 }
 
-var fwhMqttClient = FwhMqttClient{
-	Client: nil,
-	TaskCh: make(chan MqttMsg, 1000),
+// Env FWH 服务器环境，每个环境使用独立的 broker、证书和客户端实例
+type Env string
+
+const (
+	EnvProd Env = "prod" // 正式服
+	EnvTest Env = "test" // 测试服
+)
+
+// ParseEnv 解析请求中的环境参数，空值默认为正式服
+func ParseEnv(s string) (Env, error) {
+	switch Env(s) {
+	case "", EnvProd:
+		return EnvProd, nil
+	case EnvTest:
+		return EnvTest, nil
+	default:
+		return "", fmt.Errorf("unknown FWH MQTT env: %s", s)
+	}
 }
 
-func InitFwhMqttClient(cfg *config.FwhMqttConfig) error {
+var (
+	fwhMqttClients   = map[Env]*FwhMqttClient{}
+	fwhMqttClientsMu sync.RWMutex
+)
+
+// InitFwhMqttClient 初始化指定环境的 MQTT 客户端；失败时该环境不可用，不影响其他环境
+func InitFwhMqttClient(env Env, cfg *config.FwhMqttConfig) error {
+	if cfg.FwhMqttBroker == "" {
+		zap.L().Warn("FWH MQTT broker not configured, skip init.", zap.String("env", string(env)))
+		return fmt.Errorf("FWH MQTT broker not configured (env=%s)", env)
+	}
+
 	// 证书配置
 	caCert, err := os.ReadFile(cfg.FwhMqttCaUrl)
 	if err != nil {
-		panic(err)
+		zap.L().Error("Read FWH MQTT CA failed.", zap.String("env", string(env)), zap.String("err", err.Error()))
+		return err
 	}
 
 	caCertPool := x509.NewCertPool()
@@ -76,7 +104,8 @@ func InitFwhMqttClient(cfg *config.FwhMqttConfig) error {
 
 	cert, err := tls.LoadX509KeyPair(cfg.FwhMqttCertUrl, cfg.FwhMqttKeyUrl)
 	if err != nil {
-		panic(err)
+		zap.L().Error("Load FWH MQTT cert failed.", zap.String("env", string(env)), zap.String("err", err.Error()))
+		return err
 	}
 
 	client, err := mqttClient.CreateMqttClient(mqttClient.MqttClientOpt{
@@ -88,18 +117,42 @@ func InitFwhMqttClient(cfg *config.FwhMqttConfig) error {
 		},
 	})
 	if err != nil {
-		zap.L().Error("Init FWH MQTT client failed.")
+		zap.L().Error("Init FWH MQTT client failed.", zap.String("env", string(env)))
 		return err
 	}
-	fwhMqttClient.Client = client
 
-	zap.L().Info("Init FWH MQTT client success.")
+	c := &FwhMqttClient{
+		Client: client,
+		TaskCh: make(chan MqttMsg, 1000),
+	}
+
+	fwhMqttClientsMu.Lock()
+	if _, exists := fwhMqttClients[env]; exists {
+		fwhMqttClientsMu.Unlock()
+		return fmt.Errorf("FWH MQTT client already initialized (env=%s)", env)
+	}
+	fwhMqttClients[env] = c
+	fwhMqttClientsMu.Unlock()
+
+	zap.L().Info("Init FWH MQTT client success.", zap.String("env", string(env)))
 
-	go fwhMqttClient.fwhMqttWorker()
+	go c.fwhMqttWorker()
 
 	return nil
 }
 
+// getFwhMqttClient 获取指定环境的 MQTT 客户端
+func getFwhMqttClient(env Env) (*FwhMqttClient, error) {
+	fwhMqttClientsMu.RLock()
+	defer fwhMqttClientsMu.RUnlock()
+
+	c, ok := fwhMqttClients[env]
+	if !ok {
+		return nil, fmt.Errorf("Fwh Mqtt Client Uninitialized (env=%s).", env)
+	}
+	return c, nil
+}
+
 func (c *FwhMqttClient) fwhMqttWorker() {
 	idleTimer := time.NewTimer(sleepTimeout)
 	defer idleTimer.Stop()

service/repositories/mqtt/fwh_mqtt/fwh_mqtt_client_test.go

@@ -0,0 +1,44 @@
+package fwh_mqtt
+
+import (
+	"context"
+	"fwh-toolbox-server/config"
+	"fwh-toolbox-server/dto"
+	"testing"
+)
+
+func TestParseEnv(t *testing.T) {
+	cases := []struct {
+		in      string
+		want    Env
+		wantErr bool
+	}{
+		{"", EnvProd, false},
+		{"prod", EnvProd, false},
+		{"test", EnvTest, false},
+		{"TEST", "", true},
+		{"dev", "", true},
+	}
+	for _, c := range cases {
+		got, err := ParseEnv(c.in)
+		if (err != nil) != c.wantErr || got != c.want {
+			t.Errorf("ParseEnv(%q) = %q, %v; want %q, err=%v", c.in, got, err, c.want, c.wantErr)
+		}
+	}
+}
+
+func TestUninitializedEnvReturnsError(t *testing.T) {
+	// 未初始化的环境不应 panic，应返回错误
+	if err := FaultRecordParamSetting(context.Background(), EnvTest, dto.FaultRecordParamInfo{Sn: "X"}); err == nil {
+		t.Fatal("expected error for uninitialized env")
+	}
+}
+
+func TestInitWithoutBrokerFails(t *testing.T) {
+	if err := InitFwhMqttClient(EnvTest, &config.FwhMqttConfig{}); err == nil {
+		t.Fatal("expected error when broker not configured")
+	}
+	if _, err := getFwhMqttClient(EnvTest); err == nil {
+		t.Fatal("client should not be registered after failed init")
+	}
+}

service/repositories/mqtt/fwh_mqtt/fwh_mqtt_debug.go

@@ -20,7 +20,13 @@ func EnableXlog(ctx context.Context, sn string) (string, error) {
 		return "", err
 	}
 
-	res, err := fwhMqttClient.sendMessageWait(ctx, "debug/"+sn, FwhMqttMsg{
+	// Xlog 只在正式服使用
+	client, err := getFwhMqttClient(EnvProd)
+	if err != nil {
+		return "", err
+	}
+
+	res, err := client.sendMessageWait(ctx, "debug/"+sn, FwhMqttMsg{
 		CmdType:   2000, // xlog请求
 		Type:      0,    // IBG
 		Snno:      123,

service/repositories/mqtt/fwh_mqtt/fwh_mqtt_fault_record.go

@@ -47,7 +47,12 @@ type FaultRecordResponse struct {
 	Result int `json:"result"` // 0:成功；-1:失败
 }
 
-func ChargeChannel(ctx context.Context, sn string) error {
+func ChargeChannel(ctx context.Context, env Env, sn string) error {
+	client, err := getFwhMqttClient(env)
+	if err != nil {
+		return err
+	}
+
 	dataArea := map[string]interface{}{
 		"cmd": 1,
 		"data": map[string]interface{}{
@@ -87,7 +92,7 @@ func ChargeChannel(ctx context.Context, sn string) error {
 		return err
 	}
 
-	err = fwhMqttClient.sendMessageOnce(ctx, "mng/"+sn, FwhMqttMsg{
+	err = client.sendMessageOnce(ctx, "mng/"+sn, FwhMqttMsg{
 		CmdType:   377, // 故障录波请求
 		Type:      0,   // IBG
 		Snno:      123,
@@ -104,7 +109,12 @@ func ChargeChannel(ctx context.Context, sn string) error {
 	return nil
 }
 
-func FaultRecordParamSetting(ctx context.Context, data dto.FaultRecordParamInfo) error {
+func FaultRecordParamSetting(ctx context.Context, env Env, data dto.FaultRecordParamInfo) error {
+	client, err := getFwhMqttClient(env)
+	if err != nil {
+		return err
+	}
+
 	siteSn := data.Sn
 
 	peParaList := []ParamSettingInfoItem{}
@@ -163,7 +173,7 @@ func FaultRecordParamSetting(ctx context.Context, data dto.FaultRecordParamInfo)
 
 	fmt.Println(string(dataStr))
 
-	msg, err := fwhMqttClient.sendMessageWait(ctx, "mng/"+siteSn, FwhMqttMsg{
+	msg, err := client.sendMessageWait(ctx, "mng/"+siteSn, FwhMqttMsg{
 		CmdType:   FaultRecordReqCmdType, // 故障录波请求
 		Type:      0,                     // IBG
 		Snno:      123,

service/repositories/mqtt/fwh_mqtt_test/test/main.go → service/repositories/mqtt/fwh_mqtt/test/main.go

@@ -6,12 +6,12 @@ import (
 	"fwh-toolbox-server/config"
 	"fwh-toolbox-server/dto"
 	"fwh-toolbox-server/logger"
-	"fwh-toolbox-server/repositories/mqtt/fwh_mqtt_test"
+	"fwh-toolbox-server/repositories/mqtt/fwh_mqtt"
 	"time"
 )
 
 func main() {
-	err := logger.Init("dev")
+	err := logger.Init("dev", "debug")
 	if err != nil {
 		panic(err)
 	}
@@ -23,11 +23,13 @@ func main() {
 		FwhMqttCertUrl: "/var/www/fwh-toolbox/server/cert/test/01.certificate.pem.crt",
 		FwhMqttKeyUrl:  "/var/www/fwh-toolbox/server/cert/test/01.private.pem.key",
 	}
-	fwh_mqtt_test.InitFwhMqttClient(&mqttConfig)
+	if err := fwh_mqtt.InitFwhMqttClient(fwh_mqtt.EnvTest, &mqttConfig); err != nil {
+		panic(err)
+	}
 
-	// fwh_mqtt_test.ChargeChannel(context.Background(), "EamonAux")
+	// fwh_mqtt.ChargeChannel(context.Background(), fwh_mqtt.EnvTest, "EamonAux")
 	// time.Sleep(2 * time.Minute)
-	// fwh_mqtt_test.ChargeChannel(context.Background(), "EamonAux")
+	// fwh_mqtt.ChargeChannel(context.Background(), fwh_mqtt.EnvTest, "EamonAux")
 	// time.Sleep(2 * time.Minute)
 
 	data := dto.FaultRecordParamInfo{
@@ -58,12 +60,12 @@ func main() {
 		},
 	}
 
-	err = fwh_mqtt_test.FaultRecordParamSetting(context.Background(), data)
+	err = fwh_mqtt.FaultRecordParamSetting(context.Background(), fwh_mqtt.EnvTest, data)
 	if err != nil {
 		fmt.Println(err.Error())
 	}
 	time.Sleep(2 * time.Minute)
-	err = fwh_mqtt_test.FaultRecordParamSetting(context.Background(), data)
+	err = fwh_mqtt.FaultRecordParamSetting(context.Background(), fwh_mqtt.EnvTest, data)
 	if err != nil {
 		fmt.Println(err.Error())
 	}

service/repositories/mqtt/fwh_mqtt_test/fwh_mqtt_client.go

@@ -1,307 +0,0 @@
-package fwh_mqtt_test
-
-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       = 1 * 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 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
-}

service/repositories/mqtt/fwh_mqtt_test/fwh_mqtt_debug.go

@@ -1,43 +0,0 @@
-// debug 设备调试数据
-package fwh_mqtt_test
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"hash/crc32"
-	"strings"
-	"time"
-)
-
-func EnableXlog(ctx context.Context, sn string) (string, error) {
-	dataArea := map[string]string{
-		"xlogaddr": "18.188.130.67:9081",
-	}
-
-	dataStr, err := json.Marshal(dataArea)
-	if err != nil {
-		return "", err
-	}
-
-	res, err := fwhMqttClient.sendMessageWait(ctx, "debug/"+sn, FwhMqttMsg{
-		CmdType:   2000, // xlog请求
-		Type:      0,    // IBG
-		Snno:      123,
-		TimeStamp: time.Now().Unix(),
-		EquipNo:   sn,
-		DataArea:  dataArea,
-		Len:       len(dataStr),
-		Crc:       strings.ToUpper(fmt.Sprintf("%08X", crc32.ChecksumIEEE(dataStr))),
-	}, "debug/"+sn, 2001)
-	if err != nil {
-		return "", err
-	}
-
-	b, err := json.Marshal(res.DataArea)
-	if err != nil {
-		return "", err
-	}
-
-	return string(b), nil
-}

service/repositories/mqtt/fwh_mqtt_test/fwh_mqtt_fault_record.go

@@ -1,202 +0,0 @@
-// fault record 故障录波请求
-package fwh_mqtt_test
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"fwh-toolbox-server/dto"
-	"hash/crc32"
-	"strings"
-	"time"
-
-	"go.uber.org/zap"
-)
-
-const (
-	FaultRecordReqCmdType = 377 // 请求协议类型
-	FaultRecordResCmdType = 378 // 应答协议类型
-)
-
-type ParamSettingParaListItem struct {
-	Id     int    `json:"id"`
-	SetVal string `json:"setVal"`
-}
-
-type ParamSettingInfoItem struct {
-	Sn       string                     `json:"sn"` // aGate/aPower SN
-	ParaList []ParamSettingParaListItem `json:"paraList"`
-}
-
-type ParamSettingListItem struct {
-	Type string                 `json:"type"` // 'pe' or 'aGate' or 'mppt'
-	Info []ParamSettingInfoItem `json:"info"`
-}
-
-type ParamSettingData struct {
-	List []ParamSettingListItem `json:"list"`
-}
-
-type FaultRecordDataArea struct {
-	Cmd  int         `json:"cmd"` // 0:实时数据获取；1:参数设置；2:文件获取
-	Data interface{} `json:"data"`
-}
-
-type FaultRecordResponse struct {
-	Cmd    int `json:"cmd"`    // 0:实时数据获取；1:参数设置；2:文件获取
-	Result int `json:"result"` // 0:成功；-1:失败
-}
-
-func ChargeChannel(ctx context.Context, sn string) error {
-	dataArea := map[string]interface{}{
-		"cmd": 1,
-		"data": map[string]interface{}{
-			"list": []interface{}{
-				map[string]interface{}{
-					"type": "pe",
-					"info": []interface{}{
-						map[string]interface{}{
-							"sn": "10070025A05X25310280",
-							"paraList": []interface{}{
-								map[string]interface{}{
-									"id":     6,
-									"setVal": "010100015984",
-								},
-								map[string]interface{}{
-									"id":     6,
-									"setVal": "020100015981",
-								},
-								map[string]interface{}{
-									"id":     6,
-									"setVal": "030100015983",
-								},
-								map[string]interface{}{
-									"id":     6,
-									"setVal": "04030003FD34",
-								},
-							},
-						},
-					},
-				},
-			},
-		},
-	}
-
-	dataStr, err := json.Marshal(dataArea)
-	if err != nil {
-		return err
-	}
-
-	fmt.Println(string(dataStr))
-
-	err = fwhMqttClient.sendMessageOnce(ctx, "mng/"+sn, FwhMqttMsg{
-		CmdType:   377, // 故障录波请求
-		Type:      0,   // IBG
-		Snno:      123,
-		TimeStamp: time.Now().Unix(),
-		EquipNo:   sn,
-		DataArea:  dataArea,
-		Len:       len(dataStr),
-		Crc:       strings.ToUpper(fmt.Sprintf("%08X", crc32.ChecksumIEEE(dataStr))),
-	})
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func FaultRecordParamSetting(ctx context.Context, data dto.FaultRecordParamInfo) error {
-	siteSn := data.Sn
-
-	peParaList := []ParamSettingInfoItem{}
-	agateParaList := []ParamSettingInfoItem{}
-
-	for _, value := range data.List {
-		var paraSettingList []ParamSettingParaListItem
-		for _, v := range value.ParaList {
-			paraSettingList = append(paraSettingList, ParamSettingParaListItem{
-				Id:     v.Id,
-				SetVal: v.Val,
-			})
-		}
-
-		switch value.Type {
-		case "pe":
-			peParaList = append(peParaList, ParamSettingInfoItem{
-				Sn:       value.Sn,
-				ParaList: paraSettingList,
-			})
-		case "agate":
-			agateParaList = append(agateParaList, ParamSettingInfoItem{
-				Sn:       value.Sn,
-				ParaList: paraSettingList,
-			})
-		}
-	}
-
-	paraData := ParamSettingData{
-		List: []ParamSettingListItem{},
-	}
-
-	if len(agateParaList) > 0 {
-		paraData.List = append(paraData.List, ParamSettingListItem{
-			Type: "aGate",
-			Info: agateParaList,
-		})
-	}
-
-	if len(peParaList) > 0 {
-		paraData.List = append(paraData.List, ParamSettingListItem{
-			Type: "pe",
-			Info: peParaList,
-		})
-	}
-
-	dataArea := FaultRecordDataArea{
-		Cmd:  1, // 参数设置
-		Data: paraData,
-	}
-
-	dataStr, err := json.Marshal(dataArea)
-	if err != nil {
-		return err
-	}
-
-	fmt.Println(string(dataStr))
-
-	msg, err := fwhMqttClient.sendMessageWait(ctx, "mng/"+siteSn, FwhMqttMsg{
-		CmdType:   FaultRecordReqCmdType, // 故障录波请求
-		Type:      0,                     // IBG
-		Snno:      123,
-		TimeStamp: time.Now().Unix(),
-		EquipNo:   siteSn,
-		DataArea:  dataArea,
-		Len:       len(dataStr),
-		Crc:       strings.ToUpper(fmt.Sprintf("%08X", crc32.ChecksumIEEE(dataStr))),
-	}, "mng/"+siteSn, FaultRecordResCmdType)
-	if err != nil {
-		return err
-	}
-
-	if msg == nil {
-		return fmt.Errorf("FaultRecordParamSetting response data is empty.")
-	}
-
-	var res FaultRecordResponse
-	b, err := json.Marshal(msg.DataArea)
-	if err != nil {
-		return err
-	}
-
-	if err := json.Unmarshal(b, &res); err != nil {
-		zap.L().Error("FaultRecordParamSetting response data type unmarshal error.", zap.String("DataArea", string(b)), zap.String("err", err.Error()))
-		return err
-	}
-
-	if res.Result != 0 {
-		return fmt.Errorf("FaultRecord param setting failed.")
-	}
-
-	return nil
-}

service/repositories/repositories.go

@@ -505,6 +505,14 @@ var DeleteAiChatSession = mongo.DeleteAiChatSession
 
 /*************** MQTT ***************/
 // FwhMqtt
+type FwhMqttEnv = fwh_mqtt.Env
+
+const (
+	FwhMqttEnvProd = fwh_mqtt.EnvProd // 正式服
+	FwhMqttEnvTest = fwh_mqtt.EnvTest // 测试服
+)
+
+var FwhMqttParseEnv = fwh_mqtt.ParseEnv
 var FwhMqttInit = fwh_mqtt.InitFwhMqttClient
 
 // FwhMqtt - Xlog

service/services/faultRecordServices.go

@@ -8,5 +8,9 @@ import (
 
 // 设置故障录波参数
 func SetFaultRecordParam(ctx context.Context, para dto.FaultRecordParamInfo) error {
-	return repositories.FwhMqttFaultRecordParamSetting(ctx, para)
+	env, err := repositories.FwhMqttParseEnv(para.Env)
+	if err != nil {
+		return err
+	}
+	return repositories.FwhMqttFaultRecordParamSetting(ctx, env, para)
 }