package controller import ( "encoding/json" "fmt" "strings" "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" "be.ems/src/modules/tool/service" wsService "be.ems/src/modules/ws/service" "github.com/gin-gonic/gin" ) // 实例化控制层 IPerfController 结构体 var NewIPerf = &IPerfController{ iperfService: service.NewIPerf, wsService: wsService.NewWS, } // iperf 网络性能测试工具 https://iperf.fr/iperf-download.php // // PATH /tool/iperf type IPerfController struct { iperfService *service.IPerf // IPerf3 网络性能测试工具服务 wsService *wsService.WS // WebSocket 服务 } // iperf 版本信息 // // GET /v func (s *IPerfController) Version(c *gin.Context) { language := ctx.AcceptLanguage(c) var query struct { NeType string `form:"neType" binding:"required"` // 网元类型 NeID string `form:"neId" binding:"required"` // 网元ID } if err := c.ShouldBindQuery(&query); err != nil { c.JSON(400, result.CodeMsg(400, i18n.TKey(language, "app.common.err400"))) return } output, err := s.iperfService.Version(query.NeType, query.NeID) if err != nil { c.JSON(200, result.ErrMsg(i18n.TKey(language, err.Error()))) return } data := strings.Split(output, "\n") c.JSON(200, result.OkData(data)) } // iperf 软件安装 // // POST /i func (s *IPerfController) Install(c *gin.Context) { language := ctx.AcceptLanguage(c) var body struct { NeType string `json:"neType" binding:"required"` // 网元类型 NeID string `json:"neId" binding:"required"` // 网元ID } if err := c.ShouldBindBodyWithJSON(&body); err != nil { c.JSON(400, result.CodeMsg(400, i18n.TKey(language, "app.common.err400"))) return } if err := s.iperfService.Install(body.NeType, body.NeID); err != nil { c.JSON(200, result.ErrMsg(i18n.TKey(language, err.Error()))) return } c.JSON(200, result.Ok(nil)) } // iperf 软件运行 // // GET /run func (s *IPerfController) Run(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 } // 登录用户信息 loginUser, err := ctx.LoginUser(c) if err != nil { c.JSON(401, result.CodeMsg(401, i18n.TKey(language, err.Error()))) return } // 网元主机的SSH客户端 sshClient, err := neService.NewNeInfo.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.iperfService.Run) // 等待1秒,排空首次消息 time.Sleep(1 * time.Second) _ = clientSession.Read() // 实时读取Run消息直接输出 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("iperf3_%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 } } }