100 lines
2.9 KiB
Go
100 lines
2.9 KiB
Go
package service
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"time"
|
|
|
|
"be.ems/src/framework/logger"
|
|
"be.ems/src/framework/utils/ssh"
|
|
"be.ems/src/framework/utils/telnet"
|
|
"be.ems/src/framework/vo/result"
|
|
"be.ems/src/modules/ws/model"
|
|
"be.ems/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 "ims_cdr":
|
|
resByte, err = processor.GetCDRConnectByIMS(reqMsg.RequestID, reqMsg.Data)
|
|
case "smf_cdr":
|
|
resByte, err = processor.GetCDRConnectBySMF(reqMsg.RequestID, reqMsg.Data)
|
|
case "amf_ue":
|
|
resByte, err = processor.GetUEConnectByAMF(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
|
|
if err == io.EOF {
|
|
// 等待1s后关闭连接
|
|
time.Sleep(1 * time.Second)
|
|
client.StopChan <- struct{}{}
|
|
}
|
|
return
|
|
}
|
|
if len(resByte) > 0 {
|
|
client.MsgChan <- resByte
|
|
}
|
|
}
|