fix stream
All checks were successful
Deploy on Master Change / deploy (push) Successful in 1m48s

This commit is contained in:
2026-07-14 08:14:10 +07:00
parent a55b0a4567
commit 1ed246eec8
4 changed files with 57 additions and 28 deletions

View File

@@ -96,7 +96,7 @@ export const ExamGridProctor: React.FC<ExamGridProctorProps> = ({
});
currentVisibleIds.forEach((id) => {
if (!prev.has(id)) {
ws.send(JSON.stringify({ event: 'teacher:subscribe', data: { studentId: id } }));
ws.send(JSON.stringify({ event: 'teacher:subscribe', data: { studentId: id, mode: 'grid' } }));
prev.add(id);
}
});

View File

@@ -70,7 +70,7 @@ export const ProctorStreamPanels: React.FC<ProctorStreamPanelsProps> = ({
hasOpened.current = true;
setStreaming(true);
setErrorMessage(null);
ws.send(JSON.stringify({ event: 'teacher:subscribe', data: { studentId } }));
ws.send(JSON.stringify({ event: 'teacher:subscribe', data: { studentId, mode: 'focus' } }));
clearPing();
pingTimer = setInterval(() => {
if (ws.readyState === WebSocket.OPEN) {

View File

@@ -60,6 +60,8 @@ type WsHub struct {
subscribers map[int64][]string // studentId -> list of teacher connection addresses
grace map[int64]offlineGrace
graceTimers map[int64]*time.Timer
subscriberModes map[string]string // "teacherAddr_studentId" -> "grid"|"focus"
lastRelayed map[string]time.Time // "teacherAddr_studentId_event" -> time
}
var Hub = &WsHub{
@@ -69,6 +71,8 @@ var Hub = &WsHub{
subscribers: make(map[int64][]string),
grace: make(map[int64]offlineGrace),
graceTimers: make(map[int64]*time.Timer),
subscriberModes: make(map[string]string),
lastRelayed: make(map[string]time.Time),
}
func (h *WsHub) IsStudentOnline(studentRkID int64) bool {
@@ -317,6 +321,12 @@ func (h *WsHub) Unregister(c *SocketClient) {
newList = append(newList, addr)
}
}
// Clean up mode map for this student and teacher
key := c.Addr + "_" + strconv.FormatInt(sID, 10)
delete(h.subscriberModes, key)
delete(h.lastRelayed, key+"_screenshot_stream_frame")
delete(h.lastRelayed, key+"_webcam_stream_frame")
if len(newList) == 0 {
delete(h.subscribers, sID)
if student, exists := h.students[sID]; exists {
@@ -330,7 +340,7 @@ func (h *WsHub) Unregister(c *SocketClient) {
}
}
func (h *WsHub) Subscribe(teacherAddr string, studentID int64) {
func (h *WsHub) Subscribe(teacherAddr string, studentID int64, mode string) {
h.mu.Lock()
defer h.mu.Unlock()
@@ -344,9 +354,12 @@ func (h *WsHub) Subscribe(teacherAddr string, studentID int64) {
}
if !alreadySubscribed {
h.subscribers[studentID] = append(teachersList, teacherAddr)
log.Printf("[WS] Teacher %s subscribed to student %d stream", teacherAddr, studentID)
log.Printf("[WS] Teacher %s subscribed to student %d stream in %s mode", teacherAddr, studentID, mode)
}
key := teacherAddr + "_" + strconv.FormatInt(studentID, 10)
h.subscriberModes[key] = mode
if student, exists := h.students[studentID]; exists {
_ = student.WriteJSON(SocketMsg{Event: "start_screenshot_stream"})
_ = student.WriteJSON(SocketMsg{Event: "start_webcam_stream"})
@@ -357,6 +370,11 @@ func (h *WsHub) Unsubscribe(teacherAddr string, studentID int64) {
h.mu.Lock()
defer h.mu.Unlock()
key := teacherAddr + "_" + strconv.FormatInt(studentID, 10)
delete(h.subscriberModes, key)
delete(h.lastRelayed, key+"_screenshot_stream_frame")
delete(h.lastRelayed, key+"_webcam_stream_frame")
teachersList, exists := h.subscribers[studentID]
if !exists {
return
@@ -382,15 +400,13 @@ func (h *WsHub) Unsubscribe(teacherAddr string, studentID int64) {
}
func (h *WsHub) RelayFrameRaw(studentID int64, event string, rawImageBuffer json.RawMessage) {
h.mu.RLock()
h.mu.Lock()
defer h.mu.Unlock()
teachersList, exists := h.subscribers[studentID]
if !exists || len(teachersList) == 0 {
h.mu.RUnlock()
return
}
addrs := make([]string, len(teachersList))
copy(addrs, teachersList)
h.mu.RUnlock()
relayEvent := "teacher:screenshot-stream-frame"
if event == "webcam_stream_frame" {
@@ -416,27 +432,33 @@ func (h *WsHub) RelayFrameRaw(studentID int64, event string, rawImageBuffer json
return
}
var dead []string
h.mu.RLock()
for _, addr := range addrs {
if t, found := h.teachers[addr]; found {
if err := t.WriteRaw(msgBytes); err != nil {
log.Printf("[WS] Relay to teacher %s failed: %v", addr, err)
dead = append(dead, addr)
}
now := time.Now()
for _, addr := range teachersList {
t, found := h.teachers[addr]
if !found || t == nil {
continue
}
}
h.mu.RUnlock()
for _, addr := range dead {
if t, ok := func() (*SocketClient, bool) {
h.mu.RLock()
defer h.mu.RUnlock()
t, ok := h.teachers[addr]
return t, ok
}(); ok && t != nil {
_ = t.Conn.Close()
subKey := addr + "_" + strconv.FormatInt(studentID, 10)
mode := h.subscriberModes[subKey]
if mode == "grid" {
// In grid mode, rate limit to max 1 frame per 3 seconds per stream type
relayKey := subKey + "_" + event
lastTime, ok := h.lastRelayed[relayKey]
if ok && now.Sub(lastTime) < 3*time.Second {
continue
}
h.lastRelayed[relayKey] = now
}
// Send asynchronously to avoid head-of-line blocking on slower clients
go func(client *SocketClient, data []byte) {
if err := client.WriteRaw(data); err != nil {
log.Printf("[WS] Relay to teacher %s failed: %v", client.Addr, err)
_ = client.Conn.Close()
}
}(t, msgBytes)
}
}
@@ -546,8 +568,15 @@ func WebSocketHandler(db *gorm.DB) func(*websocket.Conn) {
case string:
sID, _ = strconv.ParseInt(v, 10, 64)
}
// Nhận chế độ subscription (mặc định là "focus")
mode := "focus"
if mVal, ok := msg.Data["mode"].(string); ok && mVal != "" {
mode = mVal
}
if sID > 0 {
Hub.Subscribe(client.Addr, sID)
Hub.Subscribe(client.Addr, sID, mode)
}
}
}

Binary file not shown.