From 719704f19e0de4ae4f62fe555be19521b3122003 Mon Sep 17 00:00:00 2001 From: PhuocNTB Date: Mon, 13 Jul 2026 08:12:04 +0700 Subject: [PATCH] tam --- client/app.go | 30 ++++- management/src/components/ExamGridProctor.tsx | 124 +++++++++++------- .../src/components/ProctorStreamPanels.tsx | 91 +++++++------ server/internal/websocket/websocket.go | 63 ++++++++- 4 files changed, 215 insertions(+), 93 deletions(-) diff --git a/client/app.go b/client/app.go index 4548b46..be526a4 100644 --- a/client/app.go +++ b/client/app.go @@ -578,6 +578,7 @@ func (a *App) Logout() { blocker.Instance.Stop() a.disconnectWS() a.stopScreenshotStream() + a.stopWebcamStream() guard.SuppressFor(5 * time.Second) runtime.WindowExecJS(a.ctx, "window.location.href = 'https://portal.rikkei.edu.vn/dangnhap'") @@ -1157,7 +1158,15 @@ func (a *App) connectWS() { log.Println("[WS] Connected to proctor websocket hub.") go func() { - defer a.disconnectWS() + defer func() { + a.disconnectWS(conn) + go func() { + time.Sleep(500 * time.Millisecond) + if a.isMonitoringActive() && a.CheckLoginStatus() { + a.connectWS() + } + }() + }() for { var msg struct { Event string `json:"event"` @@ -1214,15 +1223,28 @@ func (a *App) connectWS() { }() } -func (a *App) disconnectWS() { - a.mu.Lock() - defer a.mu.Unlock() +func (a *App) disconnectWS(expectedConn ...*websocket.Conn) { + var shouldStopStreams bool + a.mu.Lock() + if len(expectedConn) > 0 && expectedConn[0] != nil { + if a.wsConn != expectedConn[0] { + a.mu.Unlock() + return + } + } if a.wsConn != nil { _ = a.wsConn.Close() a.wsConn = nil } a.wsConnected = false + shouldStopStreams = a.isStreamingSc || a.isStreamingCam + a.mu.Unlock() + + if shouldStopStreams { + a.stopScreenshotStream() + a.stopWebcamStream() + } } func (a *App) startScreenshotStream() { diff --git a/management/src/components/ExamGridProctor.tsx b/management/src/components/ExamGridProctor.tsx index 0c1c301..5085207 100644 --- a/management/src/components/ExamGridProctor.tsx +++ b/management/src/components/ExamGridProctor.tsx @@ -22,73 +22,107 @@ export const ExamGridProctor: React.FC = ({ const [isFullscreen, setIsFullscreen] = useState(false); const wsRef = useRef(null); + const subscribedRef = useRef>(new Set()); const gridContainerRef = useRef(null); const studentIdsString = useMemo( () => students.map((s) => s.studentRkId).join(','), [students] ); + const onlineKey = useMemo(() => onlineIds.join(','), [onlineIds]); + + const syncSubscriptions = (ws: WebSocket) => { + if (ws.readyState !== WebSocket.OPEN) return; + const target = new Set(onlineIds); + const prev = subscribedRef.current; + + prev.forEach((id) => { + if (!target.has(id)) { + ws.send(JSON.stringify({ event: 'teacher:unsubscribe', data: { studentId: id } })); + prev.delete(id); + } + }); + onlineIds.forEach((id) => { + if (!prev.has(id)) { + ws.send(JSON.stringify({ event: 'teacher:subscribe', data: { studentId: id } })); + prev.add(id); + } + }); + }; useEffect(() => { if (students.length === 0) return; const wsUrl = getWsUrl('/ws?role=teacher'); - const ws = new WebSocket(wsUrl); - wsRef.current = ws; + let closed = false; + let retryTimer: ReturnType | null = null; - ws.onopen = () => { - students.forEach((s) => { - ws.send( - JSON.stringify({ - event: 'teacher:subscribe', - data: { studentId: s.studentRkId }, - }) - ); - }); - }; + const connect = () => { + if (closed) return; - ws.onmessage = (event) => { - try { - const msg = JSON.parse(event.data); - if (msg.event === 'teacher:screenshot-stream-frame') { - const { studentId, imageBuffer } = msg.data; - setScreenFrames((prev) => ({ ...prev, [studentId]: imageBuffer })); - } else if (msg.event === 'teacher:webcam-stream-frame') { - const { studentId, imageBuffer } = msg.data; - setWebcamFrames((prev) => ({ ...prev, [studentId]: imageBuffer })); - } else if (msg.event === 'teacher:stream-stopped') { - const { studentId } = msg.data; - setScreenFrames((prev) => { - const next = { ...prev }; - delete next[studentId]; - return next; - }); - setWebcamFrames((prev) => { - const next = { ...prev }; - delete next[studentId]; - return next; - }); + const ws = new WebSocket(wsUrl); + wsRef.current = ws; + subscribedRef.current = new Set(); + + ws.onopen = () => syncSubscriptions(ws); + + ws.onmessage = (event) => { + try { + const msg = JSON.parse(event.data); + if (msg.event === 'teacher:screenshot-stream-frame') { + const { studentId, imageBuffer } = msg.data; + setScreenFrames((prev) => ({ ...prev, [studentId]: imageBuffer })); + } else if (msg.event === 'teacher:webcam-stream-frame') { + const { studentId, imageBuffer } = msg.data; + setWebcamFrames((prev) => ({ ...prev, [studentId]: imageBuffer })); + } else if (msg.event === 'teacher:stream-stopped') { + const { studentId } = msg.data; + setScreenFrames((prev) => { + const next = { ...prev }; + delete next[studentId]; + return next; + }); + setWebcamFrames((prev) => { + const next = { ...prev }; + delete next[studentId]; + return next; + }); + } + } catch (err) { + console.error('Error parsing WS message in grid view:', err); } - } catch (err) { - console.error('Error parsing WS message in grid view:', err); - } + }; + + ws.onclose = () => { + subscribedRef.current = new Set(); + if (!closed) { + retryTimer = setTimeout(connect, 3000); + } + }; }; + connect(); + return () => { - if (ws.readyState === WebSocket.OPEN) { - students.forEach((s) => { - ws.send( - JSON.stringify({ - event: 'teacher:unsubscribe', - data: { studentId: s.studentRkId }, - }) - ); + closed = true; + if (retryTimer) clearTimeout(retryTimer); + const ws = wsRef.current; + if (ws?.readyState === WebSocket.OPEN) { + subscribedRef.current.forEach((id) => { + ws.send(JSON.stringify({ event: 'teacher:unsubscribe', data: { studentId: id } })); }); } - ws.close(); + ws?.close(); + wsRef.current = null; + subscribedRef.current = new Set(); }; }, [studentIdsString]); + useEffect(() => { + const ws = wsRef.current; + if (ws) syncSubscriptions(ws); + }, [onlineKey]); + // Clean up frames when students are removed or go offline useEffect(() => { setScreenFrames((prev) => { diff --git a/management/src/components/ProctorStreamPanels.tsx b/management/src/components/ProctorStreamPanels.tsx index 571be2b..527f979 100644 --- a/management/src/components/ProctorStreamPanels.tsx +++ b/management/src/components/ProctorStreamPanels.tsx @@ -32,6 +32,7 @@ export const ProctorStreamPanels: React.FC = ({ useEffect(() => { const wsUrl = getWsUrl('/ws?role=teacher'); + let retryTimer: ReturnType | null = null; intentionalClose.current = false; hasOpened.current = false; @@ -41,9 +42,6 @@ export const ProctorStreamPanels: React.FC = ({ setScreenFrame(null); setWebcamFrame(null); - const ws = new WebSocket(wsUrl); - wsRef.current = ws; - const markStreaming = () => { if (!hasFrames.current) { hasFrames.current = true; @@ -52,48 +50,63 @@ export const ProctorStreamPanels: React.FC = ({ } }; - ws.onopen = () => { - hasOpened.current = true; - setStreaming(true); - ws.send(JSON.stringify({ event: 'teacher:subscribe', data: { studentId } })); - }; - - ws.onmessage = (event) => { - try { - const msg = JSON.parse(event.data); - if (msg.event === 'teacher:screenshot-stream-frame' && msg.data.studentId === studentId) { - setScreenFrame(msg.data.imageBuffer); - markStreaming(); - } else if (msg.event === 'teacher:webcam-stream-frame' && msg.data.studentId === studentId) { - setWebcamFrame(msg.data.imageBuffer); - markStreaming(); - } else if (msg.event === 'teacher:stream-stopped' && msg.data.studentId === studentId) { - setScreenFrame(null); - setWebcamFrame(null); - setStreaming(false); - setErrorMessage('Sinh viên đã dừng stream'); - } - } catch (err) { - console.error('Error parsing WS frame:', err); - } - }; - - ws.onerror = () => {}; - - ws.onclose = () => { + const connect = () => { if (intentionalClose.current) return; - setStreaming(false); - if (!hasOpened.current && !hasFrames.current) { - setErrorMessage('Không thể kết nối máy chủ giám sát'); - } + + const ws = new WebSocket(wsUrl); + wsRef.current = ws; + + ws.onopen = () => { + hasOpened.current = true; + setStreaming(true); + setErrorMessage(null); + ws.send(JSON.stringify({ event: 'teacher:subscribe', data: { studentId } })); + }; + + ws.onmessage = (event) => { + try { + const msg = JSON.parse(event.data); + if (msg.event === 'teacher:screenshot-stream-frame' && msg.data.studentId === studentId) { + setScreenFrame(msg.data.imageBuffer); + markStreaming(); + } else if (msg.event === 'teacher:webcam-stream-frame' && msg.data.studentId === studentId) { + setWebcamFrame(msg.data.imageBuffer); + markStreaming(); + } else if (msg.event === 'teacher:stream-stopped' && msg.data.studentId === studentId) { + setScreenFrame(null); + setWebcamFrame(null); + setStreaming(false); + setErrorMessage('Sinh viên đã dừng stream'); + } + } catch (err) { + console.error('Error parsing WS frame:', err); + } + }; + + ws.onerror = () => {}; + + ws.onclose = () => { + if (intentionalClose.current) return; + setStreaming(false); + if (!hasOpened.current && !hasFrames.current) { + setErrorMessage('Không thể kết nối máy chủ giám sát'); + } else { + setErrorMessage('Mất kết nối giám sát — đang thử lại...'); + } + retryTimer = setTimeout(connect, 3000); + }; }; + connect(); + return () => { intentionalClose.current = true; - if (ws.readyState === WebSocket.OPEN) { - ws.send(JSON.stringify({ event: 'teacher:unsubscribe', data: { studentId } })); + if (retryTimer) clearTimeout(retryTimer); + if (wsRef.current?.readyState === WebSocket.OPEN) { + wsRef.current.send(JSON.stringify({ event: 'teacher:unsubscribe', data: { studentId } })); } - ws.close(); + wsRef.current?.close(); + wsRef.current = null; }; }, [studentId]); diff --git a/server/internal/websocket/websocket.go b/server/internal/websocket/websocket.go index 984e198..fd339ed 100644 --- a/server/internal/websocket/websocket.go +++ b/server/internal/websocket/websocket.go @@ -5,12 +5,18 @@ import ( "log" "strconv" "sync" + "time" "github.com/gofiber/websocket/v2" "gorm.io/gorm" internalDb "server/internal/db" ) +const ( + wsPingInterval = 30 * time.Second + wsPongWait = 60 * time.Second +) + type SocketMsg struct { Event string `json:"event"` Data map[string]any `json:"data"` @@ -32,6 +38,12 @@ func (c *SocketClient) WriteJSON(v any) error { return c.Conn.WriteJSON(v) } +func (c *SocketClient) WriteRaw(msg []byte) error { + c.writeMu.Lock() + defer c.writeMu.Unlock() + return c.Conn.WriteMessage(websocket.TextMessage, msg) +} + type WsHub struct { mu sync.RWMutex @@ -133,6 +145,11 @@ func (h *WsHub) Unregister(c *SocketClient) { defer h.mu.Unlock() if c.Role == "student" { + current, exists := h.students[c.StudentID] + if !exists || current != c { + log.Printf("[WS] Student %d stale disconnect ignored (replaced by newer connection)", c.StudentID) + return + } delete(h.students, c.StudentID) log.Printf("[WS] Student %d disconnected", c.StudentID) @@ -146,7 +163,6 @@ func (h *WsHub) Unregister(c *SocketClient) { }) } } - delete(h.subscribers, c.StudentID) } } else if c.Role == "teacher" { delete(h.teachers, c.Addr) @@ -248,12 +264,15 @@ func (h *WsHub) Unsubscribe(teacherAddr string, studentID int64) { // Chuyển tiếp frame ảnh từ Student đến các Teacher đã subscribe func (h *WsHub) RelayFrame(studentID int64, event string, data map[string]any) { h.mu.RLock() - defer h.mu.RUnlock() - teachersList, exists := h.subscribers[studentID] if !exists || len(teachersList) == 0 { + h.mu.RUnlock() return } + // Snapshot subscriber addresses while holding read lock + addrs := make([]string, len(teachersList)) + copy(addrs, teachersList) + h.mu.RUnlock() relayEvent := "teacher:screenshot-stream-frame" if event == "webcam_stream_frame" { @@ -267,10 +286,18 @@ func (h *WsHub) RelayFrame(studentID int64, event string, data map[string]any) { "imageBuffer": data["imageBuffer"], }, } + msgBytes, err := json.Marshal(msg) + if err != nil { + return + } - for _, addr := range teachersList { + h.mu.RLock() + defer h.mu.RUnlock() + for _, addr := range addrs { if t, found := h.teachers[addr]; found { - _ = t.WriteJSON(msg) + if err := t.WriteRaw(msgBytes); err != nil { + log.Printf("[WS] Relay to teacher %s failed: %v", addr, err) + } } } } @@ -308,11 +335,37 @@ func WebSocketHandler(db *gorm.DB) func(*websocket.Conn) { c.Close() }() + c.SetReadDeadline(time.Now().Add(wsPongWait)) + c.SetPongHandler(func(string) error { + return c.SetReadDeadline(time.Now().Add(wsPongWait)) + }) + + pingDone := make(chan struct{}) + defer close(pingDone) + go func() { + ticker := time.NewTicker(wsPingInterval) + defer ticker.Stop() + for { + select { + case <-pingDone: + return + case <-ticker.C: + client.writeMu.Lock() + err := c.WriteMessage(websocket.PingMessage, nil) + client.writeMu.Unlock() + if err != nil { + return + } + } + } + }() + for { _, msgBytes, err := c.ReadMessage() if err != nil { break } + _ = c.SetReadDeadline(time.Now().Add(wsPongWait)) var msg SocketMsg if err := json.Unmarshal(msgBytes, &msg); err != nil {