fix: ws接收消息处理改为函数回调易于扩展

src/modules/ws/controller/ws_view.go

@@ -0,0 +1,97 @@
+package controller
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"be.ems/src/framework/i18n"
+	"be.ems/src/framework/logger"
+	"be.ems/src/framework/utils/ctx"
+	"be.ems/src/framework/vo/result"
+	neService "be.ems/src/modules/network_element/service"
+
+	"github.com/gin-gonic/gin"
+)
+
+// ShellView 终端交互式文件内容查看
+//
+// GET /view
+func (s *WSController) ShellView(c *gin.Context) {
+	language := ctx.AcceptLanguage(c)
+	var query struct {
+		NeType string `form:"neType" binding:"required"` // 网元类型
+		NeId   string `form:"neId" binding:"required"`   // 网元标识id
+		Cols   int    `form:"cols"`                      // 终端单行字符数
+		Rows   int    `form:"rows"`                      // 终端显示行数
+	}
+	if err := c.ShouldBindQuery(&query); err != nil {
+		c.JSON(400, result.CodeMsg(400, i18n.TKey(language, "app.common.err400")))
+		return
+	}
+	if query.Cols < 120 || query.Cols > 400 {
+		query.Cols = 120
+	}
+	if query.Rows < 40 || query.Rows > 1200 {
+		query.Rows = 40
+	}
+
+	// 登录用户信息
+	loginUser, err := ctx.LoginUser(c)
+	if err != nil {
+		c.JSON(401, result.CodeMsg(401, i18n.TKey(language, err.Error())))
+		return
+	}
+
+	// 网元主机的SSH客户端
+	sshClient, err := neService.NewNeInfoImpl.NeRunSSHClient(query.NeType, query.NeId)
+	if err != nil {
+		c.JSON(200, result.ErrMsg(err.Error()))
+		return
+	}
+	defer sshClient.Close()
+	// ssh连接会话
+	clientSession, err := sshClient.NewClientSession(query.Cols, query.Rows)
+	if err != nil {
+		c.JSON(200, result.ErrMsg("neinfo ssh client session new err"))
+		return
+	}
+	defer clientSession.Close()
+
+	// 将 HTTP 连接升级为 WebSocket 连接
+	wsConn := s.wsService.UpgraderWs(c.Writer, c.Request)
+	if wsConn == nil {
+		return
+	}
+	defer wsConn.Close()
+
+	wsClient := s.wsService.ClientCreate(loginUser.UserID, nil, wsConn, clientSession)
+	go s.wsService.ClientWriteListen(wsClient)
+	go s.wsService.ClientReadListen(wsClient, s.wsReceiveService.ShellView)
+
+	// 等待1秒，排空首次消息
+	time.Sleep(1 * time.Second)
+	_ = clientSession.Read()
+
+	// 实时读取SSH消息直接输出
+	msTicker := time.NewTicker(100 * time.Millisecond)
+	defer msTicker.Stop()
+	for {
+		select {
+		case ms := <-msTicker.C:
+			outputByte := clientSession.Read()
+			if len(outputByte) > 0 {
+				outputStr := string(outputByte)
+				msgByte, _ := json.Marshal(result.Ok(map[string]any{
+					"requestId": fmt.Sprintf("view_%d", ms.UnixMilli()),
+					"data":      outputStr,
+				}))
+				wsClient.MsgChan <- msgByte
+			}
+		case <-wsClient.StopChan: // 等待停止信号
+			s.wsService.ClientClose(wsClient.ID)
+			logger.Infof("ws Stop Client UID %s", wsClient.BindUid)
+			return
+		}
+	}
+}