Skip to content

Commit 95e7bc1

Browse files
h2zih2zi
committed
fix: lint
1 parent c5eeb7d commit 95e7bc1

3 files changed

Lines changed: 132 additions & 132 deletions

File tree

‎internal/route/ws.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,9 +18,9 @@ func NewWs(ws *service.WsService) *Ws {
1818

1919
func (route *Ws) Register(r *chi.Mux) {
2020
r.Route("/api/ws", func(r chi.Router) {
21-
r.Get("/ssh", route.ws.Session)
2221
r.Get("/exec", route.ws.Exec)
2322
r.Get("/pty", route.ws.PTY)
23+
r.Get("/ssh", route.ws.Session)
2424
r.Get("/container/{id}", route.ws.ContainerTerminal)
2525
r.Get("/container/image/pull", route.ws.ContainerImagePull)
2626
})

‎internal/service/ws.go‎

Lines changed: 120 additions & 120 deletions
Original file line numberDiff line numberDiff line change
@@ -38,88 +38,171 @@ func NewWsService(t *gotext.Locale, conf *config.Config, log *slog.Logger, ssh b
3838
}
3939
}
4040

41-
func (s *WsService) Session(w http.ResponseWriter, r *http.Request) {
42-
req, err := Bind[request.ID](r)
41+
func (s *WsService) Exec(w http.ResponseWriter, r *http.Request) {
42+
ws, err := s.upgrade(w, r)
4343
if err != nil {
44-
Error(w, http.StatusUnprocessableEntity, "%v", err)
44+
s.log.Warn("[Websocket] upgrade exec ws error", slog.Any("err", err))
4545
return
4646
}
47-
info, err := s.sshRepo.Get(req.ID)
47+
defer func(ws *websocket.Conn) { _ = ws.CloseNow() }(ws)
48+
49+
// 第一条消息是命令
50+
ctx, cancel := context.WithCancel(r.Context())
51+
defer cancel()
52+
53+
_, cmd, err := ws.Read(ctx)
4854
if err != nil {
49-
Error(w, http.StatusInternalServerError, "%v", err)
55+
_ = ws.Close(websocket.StatusNormalClosure, s.t.Get("failed to read command: %v", err))
5056
return
5157
}
5258

53-
ws, err := s.upgrade(w, r)
59+
out, err := shell.ExecfWithPipe(ctx, string(cmd))
5460
if err != nil {
55-
s.log.Warn("[Websocket] upgrade session ws error", slog.Any("err", err))
61+
_ = ws.Close(websocket.StatusNormalClosure, s.t.Get("failed to run command: %v", err))
5662
return
5763
}
58-
defer func(ws *websocket.Conn) { _ = ws.CloseNow() }(ws)
5964

60-
client, err := ssh.NewSSHClient(info.Config)
65+
go func() {
66+
scanner := bufio.NewScanner(out)
67+
for scanner.Scan() {
68+
line := scanner.Text()
69+
_ = ws.Write(ctx, websocket.MessageText, []byte(line))
70+
}
71+
if err = scanner.Err(); err != nil {
72+
_ = ws.Close(websocket.StatusNormalClosure, s.t.Get("failed to read command output: %v", err))
73+
}
74+
}()
75+
76+
s.readLoop(ctx, ws)
77+
}
78+
79+
// PTY 通用 PTY 命令执行
80+
// 前端发送第一条消息为要执行的命令，后端通过 PTY 执行并实时返回输出
81+
func (s *WsService) PTY(w http.ResponseWriter, r *http.Request) {
82+
ws, err := s.upgrade(w, r)
6183
if err != nil {
62-
_ = ws.Close(websocket.StatusNormalClosure, err.Error())
84+
s.log.Warn("[Websocket] upgrade pty ws error", slog.Any("err", err))
6385
return
6486
}
65-
defer func(client *stdssh.Client) { _ = client.Close() }(client)
87+
defer func(ws *websocket.Conn) { _ = ws.CloseNow() }(ws)
6688

6789
ctx, cancel := context.WithCancel(r.Context())
6890
defer cancel()
6991

70-
turn, err := ssh.NewTurn(ctx, ws, client)
92+
// 读取第一条消息获取要执行的命令
93+
_, message, err := ws.Read(ctx)
94+
if err != nil {
95+
_ = ws.Close(websocket.StatusNormalClosure, s.t.Get("failed to read command: %v", err))
96+
return
97+
}
98+
99+
command := string(message)
100+
if command == "" {
101+
_ = ws.Close(websocket.StatusNormalClosure, s.t.Get("command is empty"))
102+
return
103+
}
104+
105+
// 使用 PTY 执行命令
106+
ptyResult, err := shell.ExecWithPTY(ctx, command)
71107
if err != nil {
72-
_ = ws.Close(websocket.StatusNormalClosure, err.Error())
108+
_ = ws.Write(ctx, websocket.MessageBinary, []byte("\r\n"+s.t.Get("Failed to start command: %v", err)+"\r\n"))
109+
_ = ws.Close(websocket.StatusNormalClosure, "")
73110
return
74111
}
112+
defer func() { _ = ptyResult.Close() }()
75113

114+
// 读取 WebSocket 输入并转发到 PTY
76115
go func() {
77-
defer turn.Close() // Handle 退出后关闭 SSH 连接，以结束 Wait 阶段
78-
_ = turn.Handle(ctx)
116+
for {
117+
_, data, err := ws.Read(ctx)
118+
if err != nil {
119+
// 通常是客户端关闭连接，取消运行
120+
cancel()
121+
return
122+
}
123+
// 将用户输入写入 PTY
124+
if len(data) > 0 {
125+
_, _ = ptyResult.Write(data)
126+
}
127+
}
79128
}()
80129

81-
turn.Wait()
130+
// 读取 PTY 输出并发送到 WebSocket
131+
go func() {
132+
buf := make([]byte, 4096)
133+
for {
134+
n, err := ptyResult.Read(buf)
135+
if n > 0 {
136+
if writeErr := ws.Write(ctx, websocket.MessageBinary, buf[:n]); writeErr != nil {
137+
s.log.Warn("[Websocket] write pty output error", slog.Any("err", writeErr))
138+
cancel()
139+
return
140+
}
141+
}
142+
if err != nil {
143+
if shell.IsPTYError(err) != nil {
144+
s.log.Debug("[Websocket] pty read error", slog.Any("err", err))
145+
}
146+
return
147+
}
148+
}
149+
}()
150+
151+
// 等待命令完成
152+
if err = ptyResult.Wait(); err != nil {
153+
// 如果是因为被杀死，不输出错误
154+
if ctx.Err() == nil {
155+
_ = ws.Write(ctx, websocket.MessageBinary, []byte("\r\n"+s.t.Get("Command failed: %v", err)+"\r\n"))
156+
}
157+
}
158+
159+
_ = ws.Close(websocket.StatusNormalClosure, "")
82160
}
83161

84-
func (s *WsService) Exec(w http.ResponseWriter, r *http.Request) {
162+
func (s *WsService) Session(w http.ResponseWriter, r *http.Request) {
163+
req, err := Bind[request.ID](r)
164+
if err != nil {
165+
Error(w, http.StatusUnprocessableEntity, "%v", err)
166+
return
167+
}
168+
info, err := s.sshRepo.Get(req.ID)
169+
if err != nil {
170+
Error(w, http.StatusInternalServerError, "%v", err)
171+
return
172+
}
173+
85174
ws, err := s.upgrade(w, r)
86175
if err != nil {
87-
s.log.Warn("[Websocket] upgrade exec ws error", slog.Any("err", err))
176+
s.log.Warn("[Websocket] upgrade session ws error", slog.Any("err", err))
88177
return
89178
}
90179
defer func(ws *websocket.Conn) { _ = ws.CloseNow() }(ws)
91180

92-
// 第一条消息是命令
93-
ctx, cancel := context.WithCancel(r.Context())
94-
defer cancel()
95-
96-
_, cmd, err := ws.Read(ctx)
181+
sshClient, err := ssh.NewSSHClient(info.Config)
97182
if err != nil {
98-
_ = ws.Close(websocket.StatusNormalClosure, s.t.Get("failed to read command: %v", err))
183+
_ = ws.Close(websocket.StatusNormalClosure, err.Error())
99184
return
100185
}
186+
defer func(sshClient *stdssh.Client) { _ = sshClient.Close() }(sshClient)
101187

102-
out, err := shell.ExecfWithPipe(ctx, string(cmd))
188+
ctx, cancel := context.WithCancel(r.Context())
189+
defer cancel()
190+
191+
turn, err := ssh.NewTurn(ctx, ws, sshClient)
103192
if err != nil {
104-
_ = ws.Close(websocket.StatusNormalClosure, s.t.Get("failed to run command: %v", err))
193+
_ = ws.Close(websocket.StatusNormalClosure, err.Error())
105194
return
106195
}
107196

108197
go func() {
109-
scanner := bufio.NewScanner(out)
110-
for scanner.Scan() {
111-
line := scanner.Text()
112-
_ = ws.Write(ctx, websocket.MessageText, []byte(line))
113-
}
114-
if err = scanner.Err(); err != nil {
115-
_ = ws.Close(websocket.StatusNormalClosure, s.t.Get("failed to read command output: %v", err))
116-
}
198+
defer turn.Close() // Handle 退出后关闭 SSH 连接，以结束 Wait 阶段
199+
_ = turn.Handle(ctx)
117200
}()
118201

119-
s.readLoop(ctx, ws)
202+
turn.Wait()
120203
}
121204

122-
// ContainerTerminal 容器终端 WebSocket 处理
205+
// ContainerTerminal 容器终端
123206
func (s *WsService) ContainerTerminal(w http.ResponseWriter, r *http.Request) {
124207
req, err := Bind[request.ContainerID](r)
125208
if err != nil {
@@ -155,7 +238,7 @@ func (s *WsService) ContainerTerminal(w http.ResponseWriter, r *http.Request) {
155238
turn.Wait()
156239
}
157240

158-
// ContainerImagePull 镜像拉取 WebSocket 处理
241+
// ContainerImagePull 镜像拉取
159242
func (s *WsService) ContainerImagePull(w http.ResponseWriter, r *http.Request) {
160243
ws, err := s.upgrade(w, r)
161244
if err != nil {
@@ -247,89 +330,6 @@ func (s *WsService) ContainerImagePull(w http.ResponseWriter, r *http.Request) {
247330
_ = ws.Close(websocket.StatusNormalClosure, "")
248331
}
249332

250-
// PTY 通用 PTY 命令执行 WebSocket 处理
251-
// 前端发送第一条消息为要执行的命令，后端通过 PTY 执行并实时返回输出
252-
func (s *WsService) PTY(w http.ResponseWriter, r *http.Request) {
253-
ws, err := s.upgrade(w, r)
254-
if err != nil {
255-
s.log.Warn("[Websocket] upgrade pty ws error", slog.Any("err", err))
256-
return
257-
}
258-
defer func(ws *websocket.Conn) { _ = ws.CloseNow() }(ws)
259-
260-
ctx, cancel := context.WithCancel(r.Context())
261-
defer cancel()
262-
263-
// 读取第一条消息获取要执行的命令
264-
_, message, err := ws.Read(ctx)
265-
if err != nil {
266-
_ = ws.Close(websocket.StatusNormalClosure, s.t.Get("failed to read command: %v", err))
267-
return
268-
}
269-
270-
command := string(message)
271-
if command == "" {
272-
_ = ws.Close(websocket.StatusNormalClosure, s.t.Get("command is empty"))
273-
return
274-
}
275-
276-
// 使用 PTY 执行命令
277-
ptyResult, err := shell.ExecWithPTY(ctx, command)
278-
if err != nil {
279-
_ = ws.Write(ctx, websocket.MessageBinary, []byte("\r\n"+s.t.Get("Failed to start command: %v", err)+"\r\n"))
280-
_ = ws.Close(websocket.StatusNormalClosure, "")
281-
return
282-
}
283-
defer func() { _ = ptyResult.Close() }()
284-
285-
// 读取 WebSocket 输入并转发到 PTY
286-
go func() {
287-
for {
288-
_, data, err := ws.Read(ctx)
289-
if err != nil {
290-
// 通常是客户端关闭连接，取消运行
291-
cancel()
292-
return
293-
}
294-
// 将用户输入写入 PTY
295-
if len(data) > 0 {
296-
_, _ = ptyResult.Write(data)
297-
}
298-
}
299-
}()
300-
301-
// 读取 PTY 输出并发送到 WebSocket
302-
go func() {
303-
buf := make([]byte, 4096)
304-
for {
305-
n, err := ptyResult.Read(buf)
306-
if n > 0 {
307-
if writeErr := ws.Write(ctx, websocket.MessageBinary, buf[:n]); writeErr != nil {
308-
s.log.Warn("[Websocket] write pty output error", slog.Any("err", writeErr))
309-
cancel()
310-
return
311-
}
312-
}
313-
if err != nil {
314-
if shell.IsPTYError(err) != nil {
315-
s.log.Debug("[Websocket] pty read error", slog.Any("err", err))
316-
}
317-
return
318-
}
319-
}
320-
}()
321-
322-
// 等待命令完成
323-
if err = ptyResult.Wait(); err != nil {
324-
// 如果是因为被杀死，不输出错误
325-
if ctx.Err() == nil {
326-
_ = ws.Write(ctx, websocket.MessageBinary, []byte("\r\n"+s.t.Get("Command failed: %v", err)+"\r\n"))
327-
}
328-
}
329-
330-
_ = ws.Close(websocket.StatusNormalClosure, "")
331-
}
332-
333333
func (s *WsService) upgrade(w http.ResponseWriter, r *http.Request) (*websocket.Conn, error) {
334334
opts := &websocket.AcceptOptions{
335335
CompressionMode: websocket.CompressionContextTakeover,

‎web/src/api/ws/index.ts‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,17 @@ export default {
1313
ws.onerror = (e) => reject(e)
1414
})
1515
},
16+
// PTY 命令执行
17+
pty: (command: string): Promise<WebSocket> => {
18+
return new Promise((resolve, reject) => {
19+
const ws = new WebSocket(`${base}/pty`)
20+
ws.onopen = () => {
21+
ws.send(command)
22+
resolve(ws)
23+
}
24+
ws.onerror = (e) => reject(e)
25+
})
26+
},
1627
// 连接SSH
1728
ssh: (id: number): Promise<WebSocket> => {
1829
return new Promise((resolve, reject) => {
@@ -41,16 +52,5 @@ export default {
4152
}
4253
ws.onerror = (e) => reject(e)
4354
})
44-
},
45-
// PTY 命令执行（通用接口）
46-
pty: (command: string): Promise<WebSocket> => {
47-
return new Promise((resolve, reject) => {
48-
const ws = new WebSocket(`${base}/pty`)
49-
ws.onopen = () => {
50-
ws.send(command)
51-
resolve(ws)
52-
}
53-
ws.onerror = (e) => reject(e)
54-
})
5555
}
5656
}

0 commit comments

Comments
 (0)