PhuocNTB
6ee9773f7f
All checks were successful
Deploy on Master Change / deploy (push) Successful in 1m45s
783 lines
21 KiB
Go
783 lines
21 KiB
Go
package websocket
|
||
|
||
import (
|
||
"bufio"
|
||
"encoding/base64"
|
||
"encoding/json"
|
||
"fmt"
|
||
"log"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
internalDb "server/internal/db"
|
||
|
||
"github.com/gofiber/fiber/v2"
|
||
"github.com/gofiber/websocket/v2"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
const (
|
||
// Heartbeat kiểu game: phát hiện mất kết nối trong ~30–45s.
|
||
wsPingInterval = 15 * time.Second
|
||
wsPongWait = 45 * time.Second
|
||
// Grace khi SV reconnect — tránh nhấp nháy online/offline.
|
||
studentOfflineGrace = 8 * time.Second
|
||
)
|
||
|
||
type SocketMsg struct {
|
||
Event string `json:"event"`
|
||
Data map[string]any `json:"data"`
|
||
}
|
||
|
||
type SocketClient struct {
|
||
Conn *websocket.Conn
|
||
StudentID int64
|
||
ClassID int64
|
||
StaffID uint
|
||
Role string // "student" | "teacher"
|
||
Addr string
|
||
writeMu sync.Mutex
|
||
}
|
||
|
||
func (c *SocketClient) WriteJSON(v any) error {
|
||
c.writeMu.Lock()
|
||
defer c.writeMu.Unlock()
|
||
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 offlineGrace struct {
|
||
until time.Time
|
||
classID int64
|
||
}
|
||
|
||
type WsHub struct {
|
||
mu sync.RWMutex
|
||
students map[int64]*SocketClient
|
||
teachers map[string]*SocketClient
|
||
teachersByStaff map[uint][]string // staffId -> teacher connection addresses
|
||
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
|
||
httpScreenSubscribers map[int64][]chan []byte // studentId -> list of channels for screen MJPEG
|
||
httpWebcamSubscribers map[int64][]chan []byte // studentId -> list of channels for webcam MJPEG
|
||
}
|
||
|
||
var Hub = &WsHub{
|
||
students: make(map[int64]*SocketClient),
|
||
teachers: make(map[string]*SocketClient),
|
||
teachersByStaff: make(map[uint][]string),
|
||
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),
|
||
httpScreenSubscribers: make(map[int64][]chan []byte),
|
||
httpWebcamSubscribers: make(map[int64][]chan []byte),
|
||
}
|
||
|
||
func (h *WsHub) IsStudentOnline(studentRkID int64) bool {
|
||
h.mu.RLock()
|
||
defer h.mu.RUnlock()
|
||
if _, ok := h.students[studentRkID]; ok {
|
||
return true
|
||
}
|
||
g, ok := h.grace[studentRkID]
|
||
return ok && time.Now().Before(g.until)
|
||
}
|
||
|
||
func (h *WsHub) PushExamToStudent(studentRkID int64, data map[string]any) {
|
||
h.mu.RLock()
|
||
client, ok := h.students[studentRkID]
|
||
h.mu.RUnlock()
|
||
if !ok || client == nil {
|
||
return
|
||
}
|
||
_ = client.WriteJSON(SocketMsg{Event: "exam:paper-sent", Data: data})
|
||
}
|
||
|
||
func (h *WsHub) PushChatToStudent(studentRkID int64, data map[string]any) {
|
||
h.mu.RLock()
|
||
client, ok := h.students[studentRkID]
|
||
h.mu.RUnlock()
|
||
if !ok || client == nil {
|
||
return
|
||
}
|
||
_ = client.WriteJSON(SocketMsg{Event: "chat:message", Data: data})
|
||
}
|
||
|
||
func (h *WsHub) PushChatToStaff(staffID uint, data map[string]any) {
|
||
if data == nil {
|
||
data = map[string]any{}
|
||
}
|
||
data["targetStaffId"] = staffID
|
||
msg := SocketMsg{Event: "chat:message", Data: data}
|
||
h.mu.RLock()
|
||
defer h.mu.RUnlock()
|
||
for _, t := range h.teachers {
|
||
if t != nil && t.Role == "teacher" {
|
||
_ = t.WriteJSON(msg)
|
||
}
|
||
}
|
||
}
|
||
|
||
func (h *WsHub) BroadcastStudentViolation(data map[string]any) {
|
||
msg := SocketMsg{Event: "teacher:student-violation", Data: data}
|
||
msgBytes, err := json.Marshal(msg)
|
||
if err != nil {
|
||
return
|
||
}
|
||
h.mu.RLock()
|
||
defer h.mu.RUnlock()
|
||
for _, t := range h.teachers {
|
||
if t != nil {
|
||
_ = t.WriteRaw(msgBytes)
|
||
}
|
||
}
|
||
}
|
||
|
||
func (h *WsHub) broadcastPresence(studentID int64, online bool, classID int64) {
|
||
msg := SocketMsg{
|
||
Event: "presence:update",
|
||
Data: map[string]any{
|
||
"studentId": studentID,
|
||
"online": online,
|
||
"classId": classID,
|
||
},
|
||
}
|
||
msgBytes, err := json.Marshal(msg)
|
||
if err != nil {
|
||
return
|
||
}
|
||
for _, t := range h.teachers {
|
||
if t != nil {
|
||
_ = t.WriteRaw(msgBytes)
|
||
}
|
||
}
|
||
}
|
||
|
||
// ForceStudentOffline — tắt ngay (không chờ grace), dùng khi SV cố ý đóng app / vi phạm.
|
||
func (h *WsHub) ForceStudentOffline(studentID int64) {
|
||
h.mu.Lock()
|
||
defer h.mu.Unlock()
|
||
|
||
classID := int64(0)
|
||
if client, ok := h.students[studentID]; ok && client != nil {
|
||
classID = client.ClassID
|
||
delete(h.students, studentID)
|
||
}
|
||
if g, ok := h.grace[studentID]; ok {
|
||
if classID == 0 {
|
||
classID = g.classID
|
||
}
|
||
}
|
||
h.cancelGraceLocked(studentID)
|
||
|
||
log.Printf("[WS] Student %d force offline (intentional quit/violation)", studentID)
|
||
h.broadcastPresence(studentID, false, classID)
|
||
if teachers, exists := h.subscribers[studentID]; exists {
|
||
for _, tAddr := range teachers {
|
||
if t, found := h.teachers[tAddr]; found {
|
||
_ = t.WriteJSON(SocketMsg{
|
||
Event: "teacher:stream-stopped",
|
||
Data: map[string]any{"studentId": studentID},
|
||
})
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func (h *WsHub) cancelGraceLocked(studentID int64) {
|
||
if t, ok := h.graceTimers[studentID]; ok {
|
||
t.Stop()
|
||
delete(h.graceTimers, studentID)
|
||
}
|
||
delete(h.grace, studentID)
|
||
}
|
||
|
||
func (h *WsHub) finalizeStudentOffline(studentID int64, classID int64) {
|
||
h.mu.Lock()
|
||
defer h.mu.Unlock()
|
||
|
||
if _, online := h.students[studentID]; online {
|
||
return
|
||
}
|
||
if _, ok := h.grace[studentID]; !ok {
|
||
return
|
||
}
|
||
delete(h.grace, studentID)
|
||
delete(h.graceTimers, studentID)
|
||
|
||
log.Printf("[WS] Student %d offline after grace", studentID)
|
||
h.broadcastPresence(studentID, false, classID)
|
||
|
||
if teachers, exists := h.subscribers[studentID]; exists {
|
||
for _, tAddr := range teachers {
|
||
if t, found := h.teachers[tAddr]; found {
|
||
_ = t.WriteJSON(SocketMsg{
|
||
Event: "teacher:stream-stopped",
|
||
Data: map[string]any{"studentId": studentID},
|
||
})
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func (h *WsHub) GetOnlineStudentIDs(classID int64) []int64 {
|
||
h.mu.RLock()
|
||
defer h.mu.RUnlock()
|
||
|
||
seen := map[int64]bool{}
|
||
var ids []int64
|
||
now := time.Now()
|
||
for _, client := range h.students {
|
||
if client.ClassID == classID {
|
||
seen[client.StudentID] = true
|
||
ids = append(ids, client.StudentID)
|
||
}
|
||
}
|
||
for sid, g := range h.grace {
|
||
if now.Before(g.until) && g.classID == classID && !seen[sid] {
|
||
ids = append(ids, sid)
|
||
}
|
||
}
|
||
if ids == nil {
|
||
return []int64{}
|
||
}
|
||
return ids
|
||
}
|
||
|
||
func (h *WsHub) Register(c *SocketClient) {
|
||
h.mu.Lock()
|
||
defer h.mu.Unlock()
|
||
|
||
c.Addr = c.Conn.RemoteAddr().String()
|
||
if c.Role == "student" {
|
||
wasInGrace := false
|
||
if _, ok := h.grace[c.StudentID]; ok {
|
||
wasInGrace = true
|
||
}
|
||
h.cancelGraceLocked(c.StudentID)
|
||
h.students[c.StudentID] = c
|
||
log.Printf("[WS] Student %d registered (Address: %s, Class: %d, reconnectGrace=%v)", c.StudentID, c.Addr, c.ClassID, wasInGrace)
|
||
h.broadcastPresence(c.StudentID, true, c.ClassID)
|
||
if subs, exists := h.subscribers[c.StudentID]; exists && len(subs) > 0 {
|
||
_ = c.WriteJSON(SocketMsg{Event: "start_screenshot_stream"})
|
||
_ = c.WriteJSON(SocketMsg{Event: "start_webcam_stream"})
|
||
log.Printf("[WS] Student %d has active subscribers. Sent start stream commands.", c.StudentID)
|
||
}
|
||
} else if c.Role == "teacher" {
|
||
h.teachers[c.Addr] = c
|
||
if c.StaffID > 0 {
|
||
h.teachersByStaff[c.StaffID] = append(h.teachersByStaff[c.StaffID], c.Addr)
|
||
}
|
||
log.Printf("[WS] Teacher registered (Address: %s, Staff: %d)", c.Addr, c.StaffID)
|
||
}
|
||
}
|
||
|
||
func (h *WsHub) Unregister(c *SocketClient) {
|
||
h.mu.Lock()
|
||
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)
|
||
classID := c.ClassID
|
||
studentID := c.StudentID
|
||
|
||
h.cancelGraceLocked(studentID)
|
||
deadline := time.Now().Add(studentOfflineGrace)
|
||
h.grace[studentID] = offlineGrace{until: deadline, classID: classID}
|
||
h.graceTimers[studentID] = time.AfterFunc(studentOfflineGrace, func() {
|
||
h.finalizeStudentOffline(studentID, classID)
|
||
})
|
||
log.Printf("[WS] Student %d disconnected — grace %v before offline", studentID, studentOfflineGrace)
|
||
// Chưa broadcast offline / stream-stopped — chờ grace (reconnect nhanh như game).
|
||
} else if c.Role == "teacher" {
|
||
delete(h.teachers, c.Addr)
|
||
if c.StaffID > 0 {
|
||
list := h.teachersByStaff[c.StaffID]
|
||
next := list[:0]
|
||
for _, addr := range list {
|
||
if addr != c.Addr {
|
||
next = append(next, addr)
|
||
}
|
||
}
|
||
if len(next) == 0 {
|
||
delete(h.teachersByStaff, c.StaffID)
|
||
} else {
|
||
h.teachersByStaff[c.StaffID] = next
|
||
}
|
||
}
|
||
log.Printf("[WS] Teacher %s disconnected", c.Addr)
|
||
|
||
for sID, teachersList := range h.subscribers {
|
||
newList := []string{}
|
||
for _, addr := range teachersList {
|
||
if addr != c.Addr {
|
||
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 {
|
||
_ = student.WriteJSON(SocketMsg{Event: "stop_screenshot_stream"})
|
||
_ = student.WriteJSON(SocketMsg{Event: "stop_webcam_stream"})
|
||
}
|
||
} else {
|
||
h.subscribers[sID] = newList
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func (h *WsHub) Subscribe(teacherAddr string, studentID int64, mode string) {
|
||
h.mu.Lock()
|
||
defer h.mu.Unlock()
|
||
|
||
teachersList := h.subscribers[studentID]
|
||
alreadySubscribed := false
|
||
for _, addr := range teachersList {
|
||
if addr == teacherAddr {
|
||
alreadySubscribed = true
|
||
break
|
||
}
|
||
}
|
||
if !alreadySubscribed {
|
||
h.subscribers[studentID] = append(teachersList, teacherAddr)
|
||
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"})
|
||
}
|
||
}
|
||
|
||
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
|
||
}
|
||
|
||
newList := []string{}
|
||
for _, addr := range teachersList {
|
||
if addr != teacherAddr {
|
||
newList = append(newList, addr)
|
||
}
|
||
}
|
||
|
||
if len(newList) == 0 {
|
||
delete(h.subscribers, studentID)
|
||
log.Printf("[WS] Student %d has no more proctor subscribers. Stopping streams.", studentID)
|
||
if student, exists := h.students[studentID]; exists {
|
||
_ = student.WriteJSON(SocketMsg{Event: "stop_screenshot_stream"})
|
||
_ = student.WriteJSON(SocketMsg{Event: "stop_webcam_stream"})
|
||
}
|
||
} else {
|
||
h.subscribers[studentID] = newList
|
||
}
|
||
}
|
||
|
||
func (h *WsHub) RelayFrameRaw(studentID int64, event string, rawImageBuffer json.RawMessage) {
|
||
h.mu.Lock()
|
||
defer h.mu.Unlock()
|
||
|
||
// Broadcast to HTTP subscribers if any
|
||
if event == "screenshot_stream_frame" {
|
||
subs, exists := h.httpScreenSubscribers[studentID]
|
||
if exists && len(subs) > 0 {
|
||
var b64Str string
|
||
if err := json.Unmarshal(rawImageBuffer, &b64Str); err == nil && len(b64Str) > 0 {
|
||
if idx := strings.Index(b64Str, ","); idx != -1 {
|
||
b64Str = b64Str[idx+1:]
|
||
}
|
||
if rawBytes, err := base64.StdEncoding.DecodeString(b64Str); err == nil {
|
||
for _, ch := range subs {
|
||
select {
|
||
case ch <- rawBytes:
|
||
default:
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
} else if event == "webcam_stream_frame" {
|
||
subs, exists := h.httpWebcamSubscribers[studentID]
|
||
if exists && len(subs) > 0 {
|
||
var b64Str string
|
||
if err := json.Unmarshal(rawImageBuffer, &b64Str); err == nil && len(b64Str) > 0 {
|
||
if idx := strings.Index(b64Str, ","); idx != -1 {
|
||
b64Str = b64Str[idx+1:]
|
||
}
|
||
if rawBytes, err := base64.StdEncoding.DecodeString(b64Str); err == nil {
|
||
for _, ch := range subs {
|
||
select {
|
||
case ch <- rawBytes:
|
||
default:
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
teachersList, exists := h.subscribers[studentID]
|
||
if !exists || len(teachersList) == 0 {
|
||
return
|
||
}
|
||
|
||
relayEvent := "teacher:screenshot-stream-frame"
|
||
if event == "webcam_stream_frame" {
|
||
relayEvent = "teacher:webcam-stream-frame"
|
||
}
|
||
|
||
type RelayFrameRawMsg struct {
|
||
Event string `json:"event"`
|
||
Data struct {
|
||
StudentID int64 `json:"studentId"`
|
||
ImageBuffer json.RawMessage `json:"imageBuffer"`
|
||
} `json:"data"`
|
||
}
|
||
|
||
msg := RelayFrameRawMsg{
|
||
Event: relayEvent,
|
||
}
|
||
msg.Data.StudentID = studentID
|
||
msg.Data.ImageBuffer = rawImageBuffer
|
||
|
||
msgBytes, err := json.Marshal(msg)
|
||
if err != nil {
|
||
return
|
||
}
|
||
|
||
now := time.Now()
|
||
for _, addr := range teachersList {
|
||
t, found := h.teachers[addr]
|
||
if !found || t == nil {
|
||
continue
|
||
}
|
||
|
||
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)
|
||
}
|
||
}
|
||
|
||
func WebSocketHandler(db *gorm.DB) func(*websocket.Conn) {
|
||
return func(c *websocket.Conn) {
|
||
role := c.Query("role", "student")
|
||
studentIDStr := c.Query("studentId", "0")
|
||
classIDStr := c.Query("classId", "0")
|
||
staffIDStr := c.Query("staffId", "0")
|
||
|
||
studentID, _ := strconv.ParseInt(studentIDStr, 10, 64)
|
||
classID, _ := strconv.ParseInt(classIDStr, 10, 64)
|
||
staffID64, _ := strconv.ParseUint(staffIDStr, 10, 64)
|
||
|
||
if role == "student" && db != nil {
|
||
resolvedClassID := internalDb.FindActiveClassForStudent(db, studentID)
|
||
if resolvedClassID > 0 {
|
||
classID = resolvedClassID
|
||
}
|
||
}
|
||
|
||
client := &SocketClient{
|
||
Conn: c,
|
||
StudentID: studentID,
|
||
ClassID: classID,
|
||
StaffID: uint(staffID64),
|
||
Role: role,
|
||
}
|
||
|
||
Hub.Register(client)
|
||
defer func() {
|
||
Hub.Unregister(client)
|
||
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.WriteControl(websocket.PingMessage, []byte("ping"), time.Now().Add(5*time.Second))
|
||
client.writeMu.Unlock()
|
||
if err != nil {
|
||
return
|
||
}
|
||
}
|
||
}
|
||
}()
|
||
|
||
for {
|
||
_, msgBytes, err := c.ReadMessage()
|
||
if err != nil {
|
||
break
|
||
}
|
||
_ = c.SetReadDeadline(time.Now().Add(wsPongWait))
|
||
|
||
type EventOnlyMsg struct {
|
||
Event string `json:"event"`
|
||
}
|
||
var eventMsg EventOnlyMsg
|
||
if err := json.Unmarshal(msgBytes, &eventMsg); err != nil {
|
||
continue
|
||
}
|
||
|
||
if eventMsg.Event == "screenshot_stream_frame" || eventMsg.Event == "webcam_stream_frame" {
|
||
var frameMsg struct {
|
||
Data struct {
|
||
ImageBuffer json.RawMessage `json:"imageBuffer"`
|
||
} `json:"data"`
|
||
}
|
||
if err := json.Unmarshal(msgBytes, &frameMsg); err == nil && len(frameMsg.Data.ImageBuffer) > 0 {
|
||
// Copy slice to avoid memory race / corruption since WebSocket read buffer gets recycled
|
||
bufCopy := make([]byte, len(frameMsg.Data.ImageBuffer))
|
||
copy(bufCopy, frameMsg.Data.ImageBuffer)
|
||
Hub.RelayFrameRaw(client.StudentID, eventMsg.Event, json.RawMessage(bufCopy))
|
||
}
|
||
continue
|
||
}
|
||
|
||
var msg SocketMsg
|
||
if err := json.Unmarshal(msgBytes, &msg); err != nil {
|
||
continue
|
||
}
|
||
|
||
switch msg.Event {
|
||
case "client:ping":
|
||
_ = client.WriteJSON(SocketMsg{Event: "client:pong", Data: map[string]any{"t": time.Now().UnixMilli()}})
|
||
|
||
case "teacher:subscribe":
|
||
if client.Role == "teacher" {
|
||
if sIDVal, ok := msg.Data["studentId"]; ok {
|
||
var sID int64
|
||
switch v := sIDVal.(type) {
|
||
case float64:
|
||
sID = int64(v)
|
||
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, mode)
|
||
}
|
||
}
|
||
}
|
||
|
||
case "teacher:unsubscribe":
|
||
if client.Role == "teacher" {
|
||
if sIDVal, ok := msg.Data["studentId"]; ok {
|
||
var sID int64
|
||
switch v := sIDVal.(type) {
|
||
case float64:
|
||
sID = int64(v)
|
||
case string:
|
||
sID, _ = strconv.ParseInt(v, 10, 64)
|
||
}
|
||
if sID > 0 {
|
||
Hub.Unsubscribe(client.Addr, sID)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func (h *WsHub) RegisterHttpSubscriber(studentID int64, kind string) chan []byte {
|
||
h.mu.Lock()
|
||
defer h.mu.Unlock()
|
||
|
||
ch := make(chan []byte, 16)
|
||
if kind == "screen" {
|
||
h.httpScreenSubscribers[studentID] = append(h.httpScreenSubscribers[studentID], ch)
|
||
} else if kind == "webcam" {
|
||
h.httpWebcamSubscribers[studentID] = append(h.httpWebcamSubscribers[studentID], ch)
|
||
}
|
||
|
||
// Always trigger streams start if there's any HTTP subscriber
|
||
if student, exists := h.students[studentID]; exists {
|
||
_ = student.WriteJSON(SocketMsg{Event: "start_screenshot_stream"})
|
||
_ = student.WriteJSON(SocketMsg{Event: "start_webcam_stream"})
|
||
}
|
||
|
||
return ch
|
||
}
|
||
|
||
func (h *WsHub) UnregisterHttpSubscriber(studentID int64, kind string, ch chan []byte) {
|
||
h.mu.Lock()
|
||
defer h.mu.Unlock()
|
||
|
||
if kind == "screen" {
|
||
subs := h.httpScreenSubscribers[studentID]
|
||
var next []chan []byte
|
||
for _, c := range subs {
|
||
if c != ch {
|
||
next = append(next, c)
|
||
}
|
||
}
|
||
if len(next) == 0 {
|
||
delete(h.httpScreenSubscribers, studentID)
|
||
} else {
|
||
h.httpScreenSubscribers[studentID] = next
|
||
}
|
||
} else if kind == "webcam" {
|
||
subs := h.httpWebcamSubscribers[studentID]
|
||
var next []chan []byte
|
||
for _, c := range subs {
|
||
if c != ch {
|
||
next = append(next, c)
|
||
}
|
||
}
|
||
if len(next) == 0 {
|
||
delete(h.httpWebcamSubscribers, studentID)
|
||
} else {
|
||
h.httpWebcamSubscribers[studentID] = next
|
||
}
|
||
}
|
||
|
||
// If no subscribers left (WS or HTTP), stop student's stream
|
||
wsSubs := h.subscribers[studentID]
|
||
httpScSubs := h.httpScreenSubscribers[studentID]
|
||
httpCamSubs := h.httpWebcamSubscribers[studentID]
|
||
|
||
if len(wsSubs) == 0 && len(httpScSubs) == 0 && len(httpCamSubs) == 0 {
|
||
if student, exists := h.students[studentID]; exists {
|
||
_ = student.WriteJSON(SocketMsg{Event: "stop_screenshot_stream"})
|
||
_ = student.WriteJSON(SocketMsg{Event: "stop_webcam_stream"})
|
||
}
|
||
}
|
||
|
||
close(ch)
|
||
}
|
||
|
||
func GetStudentScreenStreamHandler(c *fiber.Ctx) error {
|
||
studentIDVal := c.Params("studentId")
|
||
studentID, err := strconv.ParseInt(studentIDVal, 10, 64)
|
||
if err != nil {
|
||
return c.Status(fiber.StatusBadRequest).SendString("Invalid student ID")
|
||
}
|
||
|
||
ch := Hub.RegisterHttpSubscriber(studentID, "screen")
|
||
c.Set("Content-Type", "multipart/x-mixed-replace; boundary=frame")
|
||
c.Set("Cache-Control", "no-cache")
|
||
c.Set("Connection", "keep-alive")
|
||
c.Set("Pragma", "no-cache")
|
||
|
||
c.Status(fiber.StatusOK)
|
||
c.Context().SetBodyStreamWriter(func(w *bufio.Writer) {
|
||
defer Hub.UnregisterHttpSubscriber(studentID, "screen", ch)
|
||
|
||
for frame := range ch {
|
||
_, _ = fmt.Fprintf(w, "--frame\r\n")
|
||
_, _ = fmt.Fprintf(w, "Content-Type: image/jpeg\r\n")
|
||
_, _ = fmt.Fprintf(w, "Content-Length: %d\r\n\r\n", len(frame))
|
||
_, _ = w.Write(frame)
|
||
_, _ = fmt.Fprintf(w, "\r\n")
|
||
if err := w.Flush(); err != nil {
|
||
return
|
||
}
|
||
}
|
||
})
|
||
|
||
return nil
|
||
}
|
||
|
||
func GetStudentWebcamStreamHandler(c *fiber.Ctx) error {
|
||
studentIDVal := c.Params("studentId")
|
||
studentID, err := strconv.ParseInt(studentIDVal, 10, 64)
|
||
if err != nil {
|
||
return c.Status(fiber.StatusBadRequest).SendString("Invalid student ID")
|
||
}
|
||
|
||
ch := Hub.RegisterHttpSubscriber(studentID, "webcam")
|
||
c.Set("Content-Type", "multipart/x-mixed-replace; boundary=frame")
|
||
c.Set("Cache-Control", "no-cache")
|
||
c.Set("Connection", "keep-alive")
|
||
c.Set("Pragma", "no-cache")
|
||
|
||
c.Status(fiber.StatusOK)
|
||
c.Context().SetBodyStreamWriter(func(w *bufio.Writer) {
|
||
defer Hub.UnregisterHttpSubscriber(studentID, "webcam", ch)
|
||
|
||
for frame := range ch {
|
||
_, _ = fmt.Fprintf(w, "--frame\r\n")
|
||
_, _ = fmt.Fprintf(w, "Content-Type: image/jpeg\r\n")
|
||
_, _ = fmt.Fprintf(w, "Content-Length: %d\r\n\r\n", len(frame))
|
||
_, _ = w.Write(frame)
|
||
_, _ = fmt.Fprintf(w, "\r\n")
|
||
if err := w.Flush(); err != nil {
|
||
return
|
||
}
|
||
}
|
||
})
|
||
|
||
return nil
|
||
}
|