feat: omc原始代码

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

@@ -0,0 +1,91 @@
+package service
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"nms_nbi/src/framework/logger"
+	"nms_nbi/src/framework/utils/ssh"
+	"nms_nbi/src/framework/utils/telnet"
+	"nms_nbi/src/framework/vo/result"
+	"nms_nbi/src/modules/ws/model"
+	"nms_nbi/src/modules/ws/processor"
+)
+
+// 实例化服务层 WSReceiveImpl 结构体
+var NewWSReceiveImpl = &WSReceiveImpl{}
+
+// WSReceiveImpl WebSocket消息接收处理 服务层处理
+type WSReceiveImpl struct{}
+
+// AsyncReceive 接收业务异步处理
+func (s *WSReceiveImpl) AsyncReceive(client *model.WSClient, reqMsg model.WSRequest) {
+	// 必传requestId确认消息
+	if reqMsg.RequestID == "" {
+		msg := "message requestId is required"
+		logger.Infof("ws AsyncReceive UID %s err: %s", client.BindUid, msg)
+		msgByte, _ := json.Marshal(result.ErrMsg(msg))
+		client.MsgChan <- msgByte
+		return
+	}
+
+	var resByte []byte
+	var err error
+
+	switch reqMsg.Type {
+	case "close":
+		// 主动关闭
+		resultByte, _ := json.Marshal(result.OkMsg("user initiated closure"))
+		client.MsgChan <- resultByte
+		// 等待1s后关闭连接
+		time.Sleep(1 * time.Second)
+		client.StopChan <- struct{}{}
+	case "ssh":
+		// SSH会话消息接收直接写入会话
+		command := reqMsg.Data.(string)
+		sshClientSession := client.ChildConn.(*ssh.SSHClientSession)
+		_, err = sshClientSession.Write(command)
+	case "ssh_resize":
+		// SSH会话窗口重置
+		msgByte, _ := json.Marshal(reqMsg.Data)
+		var data struct {
+			Cols int `json:"cols"`
+			Rows int `json:"rows"`
+		}
+		err = json.Unmarshal(msgByte, &data)
+		if err == nil {
+			sshClientSession := client.ChildConn.(*ssh.SSHClientSession)
+			err = sshClientSession.Session.WindowChange(data.Rows, data.Cols)
+		}
+	case "telnet":
+		// Telnet会话消息接收直接写入会话
+		command := reqMsg.Data.(string)
+		telnetClientSession := client.ChildConn.(*telnet.TelnetClientSession)
+		_, err = telnetClientSession.Write(command)
+	case "ps":
+		resByte, err = processor.GetProcessData(reqMsg.RequestID, reqMsg.Data)
+	case "net":
+		resByte, err = processor.GetNetConnections(reqMsg.RequestID, reqMsg.Data)
+	case "cdr":
+		resByte, err = processor.GetCDRConnect(reqMsg.RequestID, reqMsg.Data)
+	case "ue":
+		resByte, err = processor.GetUEConnect(reqMsg.RequestID, reqMsg.Data)
+	case "upf_tf":
+		resByte, err = processor.GetUPFTotalFlow(reqMsg.RequestID, reqMsg.Data)
+	case "ne_state":
+		resByte, err = processor.GetNeState(reqMsg.RequestID, reqMsg.Data)
+	default:
+		err = fmt.Errorf("message type not supported")
+	}
+
+	if err != nil {
+		logger.Warnf("ws AsyncReceive UID %s err: %s", client.BindUid, err.Error())
+		msgByte, _ := json.Marshal(result.ErrMsg(err.Error()))
+		client.MsgChan <- msgByte
+		return
+	}
+	if len(resByte) > 0 {
+		client.MsgChan <- resByte
+	}
+}