diff --git a/src/modules/ws/processor/upf_total_flow.go b/src/modules/ws/processor/upf_total_flow.go new file mode 100644 index 00000000..5c1feb7d --- /dev/null +++ b/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 +} diff --git a/src/modules/ws/service/ws_receive.impl.go b/src/modules/ws/service/ws_receive.impl.go index 2fe5dbb1..754036b6 100644 --- a/src/modules/ws/service/ws_receive.impl.go +++ b/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) - if err != nil { - return err - } - client.MsgChan <- res + resByte, err = processor.GetProcessData(reqMsg.RequestID, reqMsg.Data) case "net": - res, err := processor.GetNetConnections(reqMsg.RequestID, reqMsg.Data) - if err != nil { - return err - } - client.MsgChan <- res + resByte, err = processor.GetNetConnections(reqMsg.RequestID, reqMsg.Data) case "cdr": - res, err := processor.GetCDRConnect(reqMsg.RequestID, reqMsg.Data) - if err != nil { - return err - } - client.MsgChan <- res + resByte, err = processor.GetCDRConnect(reqMsg.RequestID, reqMsg.Data) case "ue": - res, err := processor.GetUEConnect(reqMsg.RequestID, reqMsg.Data) - if err != nil { - return err - } - client.MsgChan <- res + resByte, err = processor.GetUEConnect(reqMsg.RequestID, reqMsg.Data) + case "upf_tf": + resByte, err = processor.GetUPFTotalFlow(reqMsg.RequestID, reqMsg.Data) default: return fmt.Errorf("message type not supported") } + + if err != nil { + return err + } + client.MsgChan <- resByte return nil }