diff --git a/server/service/stream/direct/h264.go b/server/service/stream/direct/h264.go index ea65d73..b2813dc 100644 --- a/server/service/stream/direct/h264.go +++ b/server/service/stream/direct/h264.go @@ -12,6 +12,7 @@ import ( var ( streamer = newStreamer() upgrader = websocket.Upgrader{ + WriteBufferSize: 256 * 1024, CheckOrigin: func(r *http.Request) bool { return true }, @@ -36,7 +37,7 @@ func Connect(c *gin.Context) { defer streamer.removeClient(ws) for { - if _, _, err := ws.ReadMessage(); err != nil { + if _, _, err := ws.NextReader(); err != nil { log.Debugf("failed to read message (client disconnected): %s", err) return } diff --git a/server/service/stream/direct/streamer.go b/server/service/stream/direct/streamer.go index 48d5663..f7986df 100644 --- a/server/service/stream/direct/streamer.go +++ b/server/service/stream/direct/streamer.go @@ -14,20 +14,25 @@ import ( ) type Streamer struct { - mutex sync.RWMutex - clients map[*websocket.Conn]bool - running int32 + mutex sync.Mutex + clients map[*websocket.Conn]bool + clientSnapshot atomic.Pointer[[]*websocket.Conn] + running int32 } func newStreamer() *Streamer { - return &Streamer{ + s := &Streamer{ clients: make(map[*websocket.Conn]bool), } + s.updateClientSnapshotLocked() + + return s } func (s *Streamer) addClient(ws *websocket.Conn) { s.mutex.Lock() s.clients[ws] = true + s.updateClientSnapshotLocked() s.mutex.Unlock() if atomic.CompareAndSwapInt32(&s.running, 0, 1) { @@ -39,31 +44,47 @@ func (s *Streamer) addClient(ws *websocket.Conn) { func (s *Streamer) removeClient(ws *websocket.Conn) { s.mutex.Lock() delete(s.clients, ws) + count := s.updateClientSnapshotLocked() s.mutex.Unlock() - log.Debugf("h264 websocket disconnected, remaining clients: %d", len(s.clients)) + log.Debugf("h264 websocket disconnected, remaining clients: %d", count) } -func (s *Streamer) getClientCount() int { - s.mutex.RLock() - defer s.mutex.RUnlock() +func (s *Streamer) updateClientSnapshotLocked() int { + clients := make([]*websocket.Conn, 0, len(s.clients)) + for client := range s.clients { + clients = append(clients, client) + } + s.clientSnapshot.Store(&clients) - return len(s.clients) + return len(clients) +} + +func (s *Streamer) getClients() []*websocket.Conn { + clients := s.clientSnapshot.Load() + if clients == nil { + return nil + } + + return *clients } func (s *Streamer) run() { defer atomic.StoreInt32(&s.running, 0) - duration := time.Second / time.Duration(120) - ticker := time.NewTicker(duration) + screen := common.GetScreen() + common.CheckScreen() + fps := screen.FPS + + ticker := time.NewTicker(time.Second / time.Duration(fps)) defer ticker.Stop() - screen := common.GetScreen() vision := common.GetKvmVision() startTime := time.Now() for range ticker.C { - if s.getClientCount() == 0 { + clients := s.getClients() + if len(clients) == 0 { log.Debug("h264 stream stopped due to no clients") return } @@ -81,28 +102,34 @@ func (s *Streamer) run() { timestamp := time.Since(startTime).Microseconds() - if err := s.send(isKeyFrame, timestamp, data); err != nil { + if err := s.send(clients, isKeyFrame, timestamp, data); err != nil { continue } + if screen.FPS != fps && screen.FPS != 0 { + fps = screen.FPS + ticker.Reset(time.Second / time.Duration(fps)) + } + stream.GetFrameRateCounter().Update() } } -func (s *Streamer) send(isKeyFrame byte, timestamp int64, data []byte) error { +func (s *Streamer) send(clients []*websocket.Conn, isKeyFrame byte, timestamp int64, data []byte) error { buf := BufferPool.Get().(*bytes.Buffer) defer BufferPool.Put(buf) buf.Reset() + buf.Grow(1 + 8 + len(data)) if err := buf.WriteByte(isKeyFrame); err != nil { log.Errorf("failed to write keyframe flag: %s", err) return err } - tsBytes := make([]byte, 8) - binary.LittleEndian.PutUint64(tsBytes, uint64(timestamp)) - if _, err := buf.Write(tsBytes); err != nil { + var tsBytes [8]byte + binary.LittleEndian.PutUint64(tsBytes[:], uint64(timestamp)) + if _, err := buf.Write(tsBytes[:]); err != nil { log.Errorf("failed to write timestamp: %s", err) return err } @@ -112,7 +139,7 @@ func (s *Streamer) send(isKeyFrame byte, timestamp int64, data []byte) error { return err } - for client := range s.clients { + for _, client := range clients { if err := client.WriteMessage(websocket.BinaryMessage, buf.Bytes()); err != nil { log.Errorf("failed to write message to client %s: %s.", client.RemoteAddr(), err) diff --git a/server/service/stream/frame_rate.go b/server/service/stream/frame_rate.go index f5de521..8c717b2 100644 --- a/server/service/stream/frame_rate.go +++ b/server/service/stream/frame_rate.go @@ -16,9 +16,8 @@ var ( ) type FrameRateCounter struct { - frameCount int32 - fps int32 - mutex sync.Mutex + frameCount atomic.Int32 + fps atomic.Int32 } func GetFrameRateCounter() *FrameRateCounter { @@ -30,16 +29,10 @@ func GetFrameRateCounter() *FrameRateCounter { defer ticker.Stop() for range ticker.C { - counter.mutex.Lock() + currentCount := counter.frameCount.Swap(0) + counter.fps.Store(currentCount / 3) - currentCount := atomic.LoadInt32(&counter.frameCount) - - counter.fps = currentCount / 3 - atomic.StoreInt32(&counter.frameCount, 0) - - counter.mutex.Unlock() - - data := fmt.Sprintf("%d", counter.fps) + data := fmt.Sprintf("%d", counter.fps.Load()) err := os.WriteFile("/kvmapp/kvm/now_fps", []byte(data), 0o666) if err != nil { log.Errorf("failed to write fps: %s", err) @@ -52,12 +45,9 @@ func GetFrameRateCounter() *FrameRateCounter { } func (f *FrameRateCounter) Update() { - atomic.AddInt32(&f.frameCount, 1) + f.frameCount.Add(1) } func (f *FrameRateCounter) GetFPS() int32 { - f.mutex.Lock() - defer f.mutex.Unlock() - - return f.fps + return f.fps.Load() } diff --git a/server/service/stream/mjpeg/streamer.go b/server/service/stream/mjpeg/streamer.go index 1360071..0371803 100644 --- a/server/service/stream/mjpeg/streamer.go +++ b/server/service/stream/mjpeg/streamer.go @@ -13,24 +13,31 @@ import ( log "github.com/sirupsen/logrus" ) +var crlf = []byte("\r\n") + type Streamer struct { - mutex sync.RWMutex - clients map[*gin.Context]bool - running int32 - frameMutex sync.RWMutex - latestFrame LatestFrame - cacheRefs int32 + mutex sync.Mutex + clients map[*gin.Context]bool + clientSnapshot atomic.Pointer[[]*gin.Context] + running int32 + frameMutex sync.RWMutex + latestFrame LatestFrame + cacheRefs int32 } func NewStreamer() *Streamer { - return &Streamer{ + s := &Streamer{ clients: make(map[*gin.Context]bool), } + s.updateClientSnapshotLocked() + + return s } func (s *Streamer) AddClient(c *gin.Context) { s.mutex.Lock() s.clients[c] = true + s.updateClientSnapshotLocked() s.mutex.Unlock() if atomic.CompareAndSwapInt32(&s.running, 0, 1) { @@ -42,28 +49,29 @@ func (s *Streamer) AddClient(c *gin.Context) { func (s *Streamer) RemoveClient(c *gin.Context) { s.mutex.Lock() delete(s.clients, c) + count := s.updateClientSnapshotLocked() s.mutex.Unlock() - log.Debugf("mjpeg connection removed, remaining clients: %d", len(s.clients)) + log.Debugf("mjpeg connection removed, remaining clients: %d", count) } -func (s *Streamer) getClients() []*gin.Context { - s.mutex.RLock() - defer s.mutex.RUnlock() - +func (s *Streamer) updateClientSnapshotLocked() int { clients := make([]*gin.Context, 0, len(s.clients)) for c := range s.clients { clients = append(clients, c) } + s.clientSnapshot.Store(&clients) - return clients + return len(clients) } -func (s *Streamer) getClientCount() int { - s.mutex.RLock() - defer s.mutex.RUnlock() +func (s *Streamer) getClients() []*gin.Context { + clients := s.clientSnapshot.Load() + if clients == nil { + return nil + } - return len(s.clients) + return *clients } func (s *Streamer) run() { @@ -79,7 +87,8 @@ func (s *Streamer) run() { defer ticker.Stop() for range ticker.C { - if s.getClientCount() == 0 { + clients := s.getClients() + if len(clients) == 0 { log.Debug("mjpeg stream stopped due to no clients") return } @@ -94,7 +103,6 @@ func (s *Streamer) run() { s.setLatestFrame(data, screen.Width, screen.Height) } - clients := s.getClients() for _, client := range clients { if err := writeFrame(client, data); err != nil { log.Errorf("failed to write mjpeg frame for client %s: %s", client.Request.RemoteAddr, err) @@ -195,7 +203,7 @@ func writeFrame(c *gin.Context, data []byte) (err error) { return err } - if _, err = c.Writer.Write([]byte("\r\n")); err != nil { + if _, err = c.Writer.Write(crlf); err != nil { return err } diff --git a/server/service/stream/webrtc/client.go b/server/service/stream/webrtc/client.go index 1ace3cd..18f7c70 100644 --- a/server/service/stream/webrtc/client.go +++ b/server/service/stream/webrtc/client.go @@ -21,6 +21,20 @@ func NewClient(ws *websocket.Conn, videoConn *webrtc.PeerConnection) *Client { } } +func (c *Client) Close() { + if c.video != nil { + if err := c.video.Close(); err != nil { + log.Debugf("failed to close video peer connection: %s", err) + } + } + + if c.ws != nil { + if err := c.ws.Close(); err != nil { + log.Debugf("failed to close websocket: %s", err) + } + } +} + func (c *Client) WriteMessage(event string, data string) error { c.mutex.Lock() defer c.mutex.Unlock() diff --git a/server/service/stream/webrtc/h264.go b/server/service/stream/webrtc/h264.go index f2217c7..6be33c9 100644 --- a/server/service/stream/webrtc/h264.go +++ b/server/service/stream/webrtc/h264.go @@ -2,6 +2,7 @@ package webrtc import ( "NanoKVM-Server/config" + "encoding/json" "net/http" "sync" "time" @@ -15,6 +16,7 @@ import ( var ( upgrader = websocket.Upgrader{ + WriteBufferSize: 256 * 1024, CheckOrigin: func(r *http.Request) bool { return true }, @@ -79,6 +81,10 @@ func Connect(c *gin.Context) { // handle signaling signalingHandler := NewSignalingHandler(client) signalingHandler.RegisterCallbacks() + if err := sendICEServers(client, iceServers); err != nil { + log.Errorf("failed to send ICE servers: %s", err) + return + } // read and wait for { @@ -116,6 +122,30 @@ func createICEServers() []webrtc.ICEServer { return iceServers } +type clientICEServer struct { + URLs []string `json:"urls"` + Username string `json:"username,omitempty"` + Credential interface{} `json:"credential,omitempty"` +} + +func sendICEServers(client *Client, iceServers []webrtc.ICEServer) error { + clientServers := make([]clientICEServer, 0, len(iceServers)) + for _, server := range iceServers { + clientServers = append(clientServers, clientICEServer{ + URLs: server.URLs, + Username: server.Username, + Credential: server.Credential, + }) + } + + data, err := json.Marshal(clientServers) + if err != nil { + return err + } + + return client.WriteMessage("ice-servers", string(data)) +} + func createMediaEngine() (*webrtc.MediaEngine, error) { mediaEngine := &webrtc.MediaEngine{} diff --git a/server/service/stream/webrtc/manager.go b/server/service/stream/webrtc/manager.go index b80823b..cbce43f 100644 --- a/server/service/stream/webrtc/manager.go +++ b/server/service/stream/webrtc/manager.go @@ -3,7 +3,6 @@ package webrtc import ( "NanoKVM-Server/common" "NanoKVM-Server/service/stream" - "sync" "sync/atomic" "time" @@ -13,11 +12,13 @@ import ( ) func NewWebRTCManager() *WebRTCManager { - return &WebRTCManager{ + m := &WebRTCManager{ clients: make(map[*websocket.Conn]*Client), videoSending: 0, - mutex: sync.RWMutex{}, } + m.updateClientSnapshotLocked() + + return m } func (m *WebRTCManager) AddClient(ws *websocket.Conn, client *Client) { @@ -25,24 +26,42 @@ func (m *WebRTCManager) AddClient(ws *websocket.Conn, client *Client) { m.mutex.Lock() m.clients[ws] = client + count := m.updateClientSnapshotLocked() m.mutex.Unlock() - log.Debugf("added client %s, total clients: %d", ws.RemoteAddr(), len(m.clients)) + log.Debugf("added client %s, total clients: %d", ws.RemoteAddr(), count) } func (m *WebRTCManager) RemoveClient(ws *websocket.Conn) { m.mutex.Lock() delete(m.clients, ws) + count := m.updateClientSnapshotLocked() m.mutex.Unlock() - log.Debugf("removed client %s, total clients: %d", ws.RemoteAddr(), len(m.clients)) + log.Debugf("removed client %s, total clients: %d", ws.RemoteAddr(), count) } func (m *WebRTCManager) GetClientCount() int { - m.mutex.RLock() - defer m.mutex.RUnlock() + return len(m.getClients()) +} - return len(m.clients) +func (m *WebRTCManager) updateClientSnapshotLocked() int { + clients := make([]*Client, 0, len(m.clients)) + for _, client := range m.clients { + clients = append(clients, client) + } + m.clientSnapshot.Store(&clients) + + return len(clients) +} + +func (m *WebRTCManager) getClients() []*Client { + clients := m.clientSnapshot.Load() + if clients == nil { + return nil + } + + return *clients } func (m *WebRTCManager) StartVideoStream() { @@ -66,7 +85,8 @@ func (m *WebRTCManager) sendVideoStream() { defer ticker.Stop() for range ticker.C { - if m.GetClientCount() == 0 { + clients := m.getClients() + if len(clients) == 0 { log.Debugf("stop sending h264 stream") return } @@ -82,8 +102,12 @@ func (m *WebRTCManager) sendVideoStream() { Duration: duration, } - for _, client := range m.clients { - client.track.writeVideo(sample) + for _, client := range clients { + if err := client.track.writeVideoSample(sample); err != nil { + log.Errorf("failed to write h264 video to client: %s", err) + m.RemoveClient(client.WsConn()) + client.Close() + } } if screen.FPS != fps && screen.FPS != 0 { diff --git a/server/service/stream/webrtc/track.go b/server/service/stream/webrtc/track.go index 3c3b279..4155b04 100644 --- a/server/service/stream/webrtc/track.go +++ b/server/service/stream/webrtc/track.go @@ -11,7 +11,7 @@ func (t *Track) updateExtension() { t.playoutDelayExtensionID = 5 } - if t.playoutDelayExtensionData == nil || len(t.playoutDelayExtensionData) == 0 { + if len(t.playoutDelayExtensionData) == 0 { playoutDelay := &rtp.PlayoutDelayExtension{ MinDelay: 0, MaxDelay: 0, @@ -44,10 +44,3 @@ func (t *Track) writeVideoSample(sample media.Sample) error { return nil } - -func (t *Track) writeVideo(sample media.Sample) { - err := t.writeVideoSample(sample) - if err != nil { - log.Errorf("failed to write h264 video: %s", err) - } -} diff --git a/server/service/stream/webrtc/types.go b/server/service/stream/webrtc/types.go index e109254..64bdfed 100644 --- a/server/service/stream/webrtc/types.go +++ b/server/service/stream/webrtc/types.go @@ -2,6 +2,7 @@ package webrtc import ( "sync" + "sync/atomic" "github.com/gorilla/websocket" "github.com/pion/rtp" @@ -9,9 +10,10 @@ import ( ) type WebRTCManager struct { - clients map[*websocket.Conn]*Client - videoSending int32 - mutex sync.RWMutex + clients map[*websocket.Conn]*Client + clientSnapshot atomic.Pointer[[]*Client] + videoSending int32 + mutex sync.Mutex } type Client struct { @@ -21,6 +23,10 @@ type Client struct { mutex sync.Mutex } +func (c *Client) WsConn() *websocket.Conn { + return c.ws +} + type SignalingHandler struct { client *Client } diff --git a/web/src/pages/desktop/screen/direct.worker.ts b/web/src/pages/desktop/screen/direct.worker.ts index 2ac7342..b7ede20 100644 --- a/web/src/pages/desktop/screen/direct.worker.ts +++ b/web/src/pages/desktop/screen/direct.worker.ts @@ -3,9 +3,17 @@ import Queue from 'yocto-queue'; let canvas: OffscreenCanvas | null = null; let ctx: OffscreenCanvasRenderingContext2D | null = null; let rendering: boolean = false; +let flushScheduled: boolean = false; let decoder: VideoDecoder | null = null; +const maxQueuedFrames = 3; const frameQueue = new Queue(); +const frameChannel = new MessageChannel(); + +frameChannel.port1.onmessage = () => { + flushScheduled = false; + processFrameQueue(); +}; self.onmessage = (event: MessageEvent) => { const { type, data, canvas: offscreenCanvas } = event.data; @@ -61,13 +69,13 @@ function initializeDecoder() { const init = { output: (frame: VideoFrame) => { frameQueue.enqueue(frame); - if (frameQueue.size >= 10) { + while (frameQueue.size > maxQueuedFrames) { frameQueue.dequeue()?.close(); } if (!rendering) { rendering = true; - processFrameQueue(); + scheduleFrameQueue(); } }, error: () => { @@ -110,12 +118,21 @@ function processFrameQueue() { } if (frameQueue.size > 0) { - setTimeout(processFrameQueue, 0); + scheduleFrameQueue(); } else { rendering = false; } } +function scheduleFrameQueue() { + if (flushScheduled) { + return; + } + + flushScheduled = true; + frameChannel.port2.postMessage(null); +} + function renderFrame(frame: VideoFrame) { if (!canvas || !ctx) { frame.close(); @@ -142,6 +159,7 @@ function resetDecoder() { decoder = null; rendering = false; + flushScheduled = false; Array.from(frameQueue.drain()).forEach((frame) => frame.close()); } diff --git a/web/src/pages/desktop/screen/h264-webrtc.tsx b/web/src/pages/desktop/screen/h264-webrtc.tsx index 3411e99..b959ca4 100644 --- a/web/src/pages/desktop/screen/h264-webrtc.tsx +++ b/web/src/pages/desktop/screen/h264-webrtc.tsx @@ -8,6 +8,19 @@ import { getBaseUrl } from '@/lib/service.ts'; import { mouseStyleAtom } from '@/jotai/mouse.ts'; import { resolutionAtom, videoScaleAtom } from '@/jotai/screen.ts'; +type SignalingMessage = { + event?: string; + data?: string; +}; + +const parseSignalingData = (data?: string): T | null => { + if (!data) { + return null; + } + + return JSON.parse(data) as T; +}; + export const H264Webrtc = () => { const resolution = useAtomValue(resolutionAtom); const mouseStyle = useAtomValue(mouseStyleAtom); @@ -21,120 +34,11 @@ export const H264Webrtc = () => { useEffect(() => { const url = `${getBaseUrl('ws')}/api/stream/h264`; const ws = new W3cWebSocket(url); + const videoElement = videoRef.current; - const iceServers = [{ urls: ['stun:stun.l.google.com:19302'] }]; - const video = new RTCPeerConnection({ iceServers }); - - // --- Init Video --- - video.onnegotiationneeded = async () => { - if (videoOfferSent.current || video.signalingState !== 'stable') { - console.log('Skipping video negotiation - Waiting for answer or state unstable'); - return; - } - - try { - videoOfferSent.current = true; - const offer = await video.createOffer({ - offerToReceiveVideo: true, - offerToReceiveAudio: false - }); - - await video.setLocalDescription(offer); - - sendMsg('video-offer', JSON.stringify(video.localDescription)); - } catch (error) { - videoOfferSent.current = false; - console.error('Video negotiation failed:', error); - } - }; - - video.onconnectionstatechange = () => { - if (video.iceConnectionState === 'connected') { - setIsLoading(false); - } - }; - - video.ontrack = (event) => { - if (videoRef.current && event.track.kind === 'video') { - videoRef.current.srcObject = new MediaStream([event.track]); - } - }; - - ws.onopen = () => { - videoOfferSent.current = false; - - video.onicecandidate = (event) => { - if (event.candidate) { - sendMsg('video-candidate', JSON.stringify(event.candidate)); - } - }; - - video.addTransceiver('video', { direction: 'recvonly' }); - }; - - ws.onmessage = (event) => { - try { - const msg = JSON.parse(event.data as string); - if (!msg?.data) return; - - const data = JSON.parse(msg.data); - if (!data) return; - - switch (msg.event) { - case 'video-answer': - handleVideoAnswer(data); - break; - case 'video-candidate': - handleVideoCandidate(data); - break; - case 'heartbeat': - break; - default: - console.log('Unhandled event:', msg.event); - } - } catch (err) { - console.error('Message processing error:', err); - } - }; - - const handleVideoAnswer = (data: any) => { - if (video.signalingState !== 'have-local-offer') { - videoOfferSent.current = false; - console.warn(`Video signaling state incorrect for answer: ${video.signalingState}`); - return; - } - - video - .setRemoteDescription(new RTCSessionDescription(data)) - .then(() => { - videoOfferSent.current = false; - videoIceCandidates.current.forEach((candidate) => { - video - .addIceCandidate(candidate) - .catch((e) => console.error('Video candidate failed to add:', e.message)); - }); - videoIceCandidates.current = []; - }) - .catch((error) => { - console.error('Video answer set failed:', error); - videoOfferSent.current = false; - }); - }; - - const handleVideoCandidate = (data: any) => { - if (!data.candidate) { - return; - } - - const candidate = new RTCIceCandidate(data); - if (video.remoteDescription) { - video - .addIceCandidate(candidate) - .catch((e) => console.error('Video candidate failed to add:', e.message)); - } else { - videoIceCandidates.current.push(candidate); - } - }; + let video: RTCPeerConnection | null = null; + let heartbeatTimer: ReturnType | null = null; + let disposed = false; const sendMsg = (event: string, data: string) => { if (ws.readyState !== WebSocket.OPEN) { @@ -148,25 +52,173 @@ export const H264Webrtc = () => { } }; - const heartbeatTimer = setInterval(() => { - sendMsg('heartbeat', ''); - }, 60 * 1000); + const startVideo = (iceServers: RTCIceServer[]) => { + if (video || disposed) { + return; + } - setTimeout(() => { + const peer = new RTCPeerConnection({ iceServers }); + video = peer; + videoOfferSent.current = false; + videoIceCandidates.current = []; + + // --- Init Video --- + peer.onnegotiationneeded = async () => { + if (videoOfferSent.current || peer.signalingState !== 'stable') { + console.log('Skipping video negotiation - Waiting for answer or state unstable'); + return; + } + + try { + videoOfferSent.current = true; + const offer = await peer.createOffer({ + offerToReceiveVideo: true, + offerToReceiveAudio: false + }); + + await peer.setLocalDescription(offer); + + sendMsg('video-offer', JSON.stringify(peer.localDescription)); + } catch (error) { + videoOfferSent.current = false; + console.error('Video negotiation failed:', error); + } + }; + + peer.onconnectionstatechange = () => { + if (peer.iceConnectionState === 'connected' || peer.connectionState === 'connected') { + setIsLoading(false); + } + }; + + peer.ontrack = (event) => { + if (videoElement && event.track.kind === 'video') { + videoElement.srcObject = new MediaStream([event.track]); + } + }; + + peer.onicecandidate = (event) => { + if (event.candidate) { + sendMsg('video-candidate', JSON.stringify(event.candidate)); + } + }; + + peer.addTransceiver('video', { direction: 'recvonly' }); + }; + + const handleVideoAnswer = (data: RTCSessionDescriptionInit) => { + const peer = video; + if (!peer) { + return; + } + + if (peer.signalingState !== 'have-local-offer') { + videoOfferSent.current = false; + console.warn(`Video signaling state incorrect for answer: ${peer.signalingState}`); + return; + } + + peer + .setRemoteDescription(new RTCSessionDescription(data)) + .then(() => { + videoOfferSent.current = false; + videoIceCandidates.current.forEach((candidate) => { + peer + .addIceCandidate(candidate) + .catch((e) => console.error('Video candidate failed to add:', e.message)); + }); + videoIceCandidates.current = []; + }) + .catch((error) => { + console.error('Video answer set failed:', error); + videoOfferSent.current = false; + }); + }; + + const handleVideoCandidate = (data: RTCIceCandidateInit) => { + const peer = video; + if (!peer || !data.candidate) { + return; + } + + const candidate = new RTCIceCandidate(data); + if (peer.remoteDescription) { + peer + .addIceCandidate(candidate) + .catch((e) => console.error('Video candidate failed to add:', e.message)); + } else { + videoIceCandidates.current.push(candidate); + } + }; + + ws.onopen = () => { + if (disposed) { + ws.close(); + return; + } + + heartbeatTimer = setInterval(() => { + sendMsg('heartbeat', ''); + }, 60 * 1000); + }; + + ws.onmessage = (event) => { + try { + const msg = JSON.parse(event.data as string) as SignalingMessage; + + switch (msg.event) { + case 'ice-servers': { + const iceServers = parseSignalingData(msg.data); + startVideo(Array.isArray(iceServers) ? iceServers : []); + break; + } + case 'video-answer': { + const data = parseSignalingData(msg.data); + if (data) { + handleVideoAnswer(data); + } + break; + } + case 'video-candidate': { + const data = parseSignalingData(msg.data); + if (data) { + handleVideoCandidate(data); + } + break; + } + case 'heartbeat': + break; + default: + console.log('Unhandled event:', msg.event); + } + } catch (err) { + console.error('Message processing error:', err); + } + }; + + const loadingTimer = setTimeout(() => { setIsLoading(false); }, 5 * 1000); return () => { - if (ws.readyState === WebSocket.OPEN) { + disposed = true; + + if (ws.readyState !== WebSocket.CLOSING && ws.readyState !== WebSocket.CLOSED) { ws.close(); } - video.close(); + video?.close(); + video = null; + if (videoElement) { + videoElement.srcObject = null; + } videoOfferSent.current = false; + videoIceCandidates.current = []; if (heartbeatTimer) { clearInterval(heartbeatTimer); } + clearTimeout(loadingTimer); }; }, []);