From 3388a0fbceac516b105eefe2c0ee13ddd69a0a6a Mon Sep 17 00:00:00 2001 From: sergystepanov Date: Fri, 5 Jun 2026 14:46:55 +0300 Subject: [PATCH] Create data channels when pushed to the server --- pkg/network/webrtc/webrtc.go | 33 +++++++++++----- web/js/app.js | 22 ++++++----- web/js/network/webrtc.js | 76 ++++++++++++++++++++---------------- 3 files changed, 77 insertions(+), 54 deletions(-) diff --git a/pkg/network/webrtc/webrtc.go b/pkg/network/webrtc/webrtc.go index 1686b967..30b5698d 100644 --- a/pkg/network/webrtc/webrtc.go +++ b/pkg/network/webrtc/webrtc.go @@ -38,6 +38,9 @@ func (p *Peer) NewConnection(vCodec, aCodec string, onICECandidate func(ice any) } p.conn.OnConnectionStateChange(func(pcs webrtc.PeerConnectionState) { p.log.Debug().Msgf("WebRTC state change: %v", pcs) + if pcs == webrtc.PeerConnectionStateConnected { + p.log.Info().Msg("Connected") + } }) p.conn.OnICECandidate(p.handleICECandidate(onICECandidate)) @@ -51,14 +54,24 @@ func (p *Peer) NewConnection(vCodec, aCodec string, onICECandidate func(ice any) return err } - p.conn.OnICEConnectionStateChange(p.handleICEState(func() { p.log.Info().Msg("Connected") })) + p.conn.OnICEConnectionStateChange(p.handleICEState(func() {})) - p.conn.OnDataChannel(func(dc *webrtc.DataChannel) { - p.log.Debug().Msgf("Added [%s] datachannel", dc.Label()) - if dc.Label() == "data" { - if err := p.AddDataChannel(); err != nil { - p.log.Error().Msgf("Failed to add data channel: %v", err) - } + p.conn.OnDataChannel(func(ch *webrtc.DataChannel) { + p.log.Debug().Msgf("Added [%s] datachannel", ch.Label()) + + if ch.Label() == "data" { + p.d = ch + ch.OnMessage(func(m webrtc.DataChannelMessage) { + if len(m.Data) == 0 || p.OnMessage == nil { + return + } + p.OnMessage(m.Data) + }) + ch.OnOpen(func() { + p.log.Debug().Uint16("id", *ch.ID()).Msgf("Data channel [%v] opened", ch.Label()) + }) + ch.OnError(p.logx) + ch.OnClose(func() { p.log.Debug().Msgf("Data channel [%v] has been closed", ch.Label()) }) } }) @@ -212,9 +225,9 @@ func (p *Peer) handleICEState(onConnect func()) func(webrtc.ICEConnectionState) p.log.Error().Msgf("WebRTC connection fail! connection: %v, ice: %v, gathering: %v, signalling: %v", p.conn.ConnectionState(), p.conn.ICEConnectionState(), p.conn.ICEGatheringState(), p.conn.SignalingState()) - p.Disconnect() - case webrtc.ICEConnectionStateClosed, - webrtc.ICEConnectionStateDisconnected: + // make ICE restart + case webrtc.ICEConnectionStateDisconnected: + case webrtc.ICEConnectionStateClosed: p.Disconnect() default: p.log.Debug().Msg("ICE state is not handled!") diff --git a/web/js/app.js b/web/js/app.js index 1fac5f18..69552981 100755 --- a/web/js/app.js +++ b/web/js/app.js @@ -490,15 +490,13 @@ function handleWebrtcStart({ data, initiator }) { const negotiate = () => { if (makingOffer) return; - makingOffer = true; - webrtc - .offerSdp() - .then((offer) => { - if (!offer) return; - log.debug("> offer", offer); - api.server.initWebrtcStream({ initiator, sdpOffer: offer }); - }) - .finally(() => (makingOffer = false)); + // makingOffer = true; + // webrtc + // .offerSdp() + // .then((offer) => { + // if (!offer) return; + // }) + // .finally(() => (makingOffer = false)); }; const datachannel = (ch) => { @@ -523,8 +521,12 @@ function handleWebrtcStart({ data, initiator }) { }); if (initiator) { - negotiate(); + // negotiate(); webrtc.createDataChannel({ onChannel: datachannel }); + webrtc.offerSdp().then((offer) => { + if (!offer) return; + api.server.initWebrtcStream({ initiator, sdpOffer: offer }); + }); } else { api.server.initWebrtcStream(); } diff --git a/web/js/network/webrtc.js b/web/js/network/webrtc.js index b1fcde1a..8b462e0d 100755 --- a/web/js/network/webrtc.js +++ b/web/js/network/webrtc.js @@ -3,13 +3,15 @@ import { log } from "log"; let /** @type {RTCPeerConnection} */ pc; let /** @type {Map} */ channels = new Map(); let /** @type {MediaStream} */ stream; -let /** @type {RTCLocalIceCandidateInit[]} */ candidateBuf = []; let handleSdpAnswer; let _initiator = false; const ice = (() => { + let /** @type {RTCIceCandidateInit[]} */ buf = []; let handleIceCandidate; + const END_OF_CANDIDATES = null; + const onIceCandidate = (/** @type {RTCPeerConnectionIceEvent} */ ev) => { if (!ev.candidate) return; log.debug(`[rtc] [ice] local`, ev.candidate); @@ -50,6 +52,12 @@ const ice = (() => { onIceCandidateError, onIceGatheringStateChange, onIceConnectionStateChange, + buf, + new: (/** @type RTCIceCandidateInit */ candidate) => { + return candidate + ? new RTCIceCandidate(candidate) + : END_OF_CANDIDATES; + }, set handleIceCandidate(cb) { handleIceCandidate = cb; }, @@ -59,22 +67,33 @@ const ice = (() => { const isConnected = () => pc?.connectionState === "connected"; const hasRemoteDescription = () => pc?.remoteDescription !== null; -const addRemoteCandidate = (data) => { - if (!data) return; - const candidate = new RTCIceCandidate(data); +const addIceCandidate = (data) => { + const candidate = ice.new(data); pc.addIceCandidate(candidate).catch((e) => { - log.error("[rtc] [ice] remote candidate add failed", e.name); + log.error("[rtc] [ice] remote add", e.name); }); - log.debug(`[rtc] [ice] added remote`, candidate); + log.debug("[rtc] [ice] add", candidate); }; -const flushRemoteCandidates = () => { - if (!hasRemoteDescription() || candidateBuf.length === 0) return; +const flushIceCandidates = () => { + if (ice.buf.length === 0) return; - log.debug(`[rtc] [ice] remote candidate buf (${candidateBuf.length})`); + log.debug(`[rtc] [ice] buf (${ice.buf.length}) flush`); let data = undefined; - while (typeof (data = candidateBuf.shift()) !== "undefined") { - addRemoteCandidate(data); + while (typeof (data = ice.buf.shift()) !== "undefined") { + addIceCandidate(data); + } +}; + +const _setRemoteDescription = async (sdp) => { + try { + await pc.setRemoteDescription(new RTCSessionDescription(sdp)); + log.debug(`[rtc] [sdp] Trickle ICE: ${pc.canTrickleIceCandidates}`); + flushIceCandidates(); + return true; + } catch (e) { + log.error(`[rtc] [sdp] set remote error: ${e}`); + return false; } }; @@ -100,6 +119,7 @@ const stop = () => { } if (pc) { ice.handleIceCandidate = null; + ice.buf = []; handleSdpAnswer = null; pc.oniceconnectionstatechange = null; pc.onicegatheringstatechange = null; @@ -116,7 +136,6 @@ const stop = () => { channel.close(); } channels.clear(); - candidateBuf = []; log.debug("[rtc] WebRTC has been closed"); }; @@ -135,8 +154,9 @@ export const webrtc = { onIceCandidate, onSdpAnswer, } = {}) => { - log.debug("[rtc] got remote ICE servers", iceServers); - pc = new RTCPeerConnection({ iceCandidatePoolSize: 1, iceServers }); + iceServers = iceServers || []; + log.debug("[rtc] ICE servers", iceServers); + pc = new RTCPeerConnection({ iceServers }); stream = new MediaStream(); _initiator = initiator; @@ -203,19 +223,10 @@ export const webrtc = { ) => { log.debug("[rtc] [sdp] remote SDP", sdp); - if (!pc) return; - - try { - await pc.setRemoteDescription(new RTCSessionDescription(sdp)); - } catch (e) { - log.error(`[rtc] [sdp] remote offer error: ${e}`); + if (!pc || !_setRemoteDescription(sdp)) { return; } - log.debug(`[rtc] [sdp] Trickle ICE: ${pc.canTrickleIceCandidates}`); - - flushRemoteCandidates(); - if (_initiator) return; try { @@ -229,20 +240,17 @@ export const webrtc = { } }, pushChannel, - addCandidate: ( - /** @type {RTCLocalIceCandidateInit | string} */ candidate, - ) => { + addCandidate: (/** @type {RTCIceCandidateInit | string} */ candidate) => { + log.debug(`[rtc] [ice] remote`, candidate); + if (!pc) return; - if (hasRemoteDescription()) { - addRemoteCandidate(candidate); - } else { - candidateBuf.push(candidate); + if (!hasRemoteDescription()) { + ice.buf.push(candidate); + return; } - if (candidate === "") { - flushRemoteCandidates(); - } + addIceCandidate(candidate); }, send: (chan, data) => { const ch = channels.get(chan);