feat: ws支持upf_tf类型信息获取UPF-总流量数

src/modules/ws/processor/upf_total_flow.go

@@ -0,0 +1,44 @@
+package processor
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"ems.agt/src/framework/logger"
+	"ems.agt/src/framework/vo/result"
+	neDataService "ems.agt/src/modules/network_data/service"
+	neService "ems.agt/src/modules/network_element/service"
+)
+
+// GetUPFTotalFlow 获取UPF-总流量数 N3上行 N6下行 单位比特(bit)
+func GetUPFTotalFlow(requestID string, data any) ([]byte, error) {
+	msgByte, _ := json.Marshal(data)
+	var querys struct {
+		NeType string `json:"neType" form:"neType" binding:"required"`
+		NeID   string `json:"neId" form:"neId" binding:"required"`
+		Day    int    `json:"day" binding:"required"`
+	}
+	err := json.Unmarshal(msgByte, &querys)
+	if err != nil {
+		logger.Warnf("ws processor GetUPFTotalFlow err: %s", err.Error())
+		return nil, fmt.Errorf("query data structure error")
+	}
+
+	if querys.NeType == "" || querys.NeID == "" || querys.Day < 0 {
+		return nil, fmt.Errorf("query neType any neId empty or day less 0 ")
+	}
+
+	// 查询网元获取IP
+	neInfo := neService.NewNeInfoImpl.SelectNeInfoByNeTypeAndNeID(querys.NeType, querys.NeID)
+	if neInfo.NeId != querys.NeID || neInfo.IP == "" {
+		return nil, fmt.Errorf("no matching network element information found")
+	}
+
+	dataMap := neDataService.NewPerfKPIImpl.SelectUPFTotalFlow(neInfo.NeType, neInfo.RmUID, querys.Day)
+
+	resultByte, err := json.Marshal(result.Ok(map[string]any{
+		"requestId": requestID,
+		"data":      dataMap,
+	}))
+	return resultByte, err
+}

src/modules/ws/service/ws_receive.impl.go

@@ -18,33 +18,28 @@ func (s *WSReceiveImpl) Receive(client *model.WSClient, reqMsg model.WSRequest)
 	if reqMsg.RequestID == "" {
 		return fmt.Errorf("message requestId is required")
 	}
+
+	var resByte []byte
+	var err error
+
 	switch reqMsg.Type {
 	case "ps":
-		res, err := processor.GetProcessData(reqMsg.RequestID, reqMsg.Data)
+		resByte, err = processor.GetProcessData(reqMsg.RequestID, reqMsg.Data)
-		if err != nil {
-			return err
-		}
-		client.MsgChan <- res
 	case "net":
-		res, err := processor.GetNetConnections(reqMsg.RequestID, reqMsg.Data)
+		resByte, err = processor.GetNetConnections(reqMsg.RequestID, reqMsg.Data)
-		if err != nil {
-			return err
-		}
-		client.MsgChan <- res
 	case "cdr":
-		res, err := processor.GetCDRConnect(reqMsg.RequestID, reqMsg.Data)
+		resByte, err = processor.GetCDRConnect(reqMsg.RequestID, reqMsg.Data)
-		if err != nil {
-			return err
-		}
-		client.MsgChan <- res
 	case "ue":
-		res, err := processor.GetUEConnect(reqMsg.RequestID, reqMsg.Data)
+		resByte, err = processor.GetUEConnect(reqMsg.RequestID, reqMsg.Data)
-		if err != nil {
+	case "upf_tf":
-			return err
+		resByte, err = processor.GetUPFTotalFlow(reqMsg.RequestID, reqMsg.Data)
-		}
-		client.MsgChan <- res
 	default:
 		return fmt.Errorf("message type not supported")
 	}
+
+	if err != nil {
+		return err
+	}
+	client.MsgChan <- resByte
 	return nil
 }