Mirrors/cloud-game
1
0
Fork 0
mirror of https://github.com/giongto35/cloud-game.git synced 2026-07-18 00:55:40 +00:00
Code Issues Projects Releases Packages Wiki Activity

Add an option for initiating WebRTC offer on the client side

Browse source 
WebRTC negotiation will start in the browser with an offer. Added because a reversed negotiation (server's offer) doesn't work in Firefox-based browsers.
This commit is contained in:
sergystepanov 2026-06-04 17:48:22 +03:00
parent 97a0d0a4c5
commit 8c4b4bf96f
8 changed files with 252 additions and 147 deletions
Download patch file Download diff file Expand all files Collapse all files

7
pkg/api/worker.go
View file

@ -49,7 +49,12 @@ type (
Stateful
Candidate string `json:"candidate"` // Base64-encoded ICE candidate
}
InitWebrtcStreamRequest Stateful
InitWebrtcStreamRequest struct {
// Stateful
Id string `json:"id"`
Initiator bool `json:"initiator"`
Sdp string `json:"sdp,omitempty"`
}
InitWebrtcStreamResponse string
AppVideoInfo struct {

9
pkg/coordinator/userhandlers.go
View file

@ -9,18 +9,11 @@ import (
)
func (u *User) HandleInitWebrtcStream(rq api.InitUserWebrtcStreamRequest) {
if rq.Initiator {
u.log.Warn().Msg("active initiator is not supported")
return
}
if u.w == nil {
u.log.Warn().Msg("no worker assigned")
return
}
uid := u.Id().String()
resp, err := u.w.InitWebrtcStream(uid)
resp, err := u.w.InitWebrtcStream(u.Id().String(), rq.Initiator, rq.Sdp)
if err != nil || resp == nil || *resp == api.EMPTY {
u.log.Error().Err(err).Msg("malformed WebRTC init response")
return

4
pkg/coordinator/workerapi.go
View file

@ -2,9 +2,9 @@ package coordinator
import "github.com/giongto35/cloud-game/v3/pkg/api"
func (w *Worker) InitWebrtcStream(id string) (*api.InitWebrtcStreamResponse, error) {
func (w *Worker) InitWebrtcStream(id string, initiator bool, sdp string) (*api.InitWebrtcStreamResponse, error) {
return api.UnwrapChecked[api.InitWebrtcStreamResponse](
w.Send(api.InitWebrtcStream, api.InitWebrtcStreamRequest{Id: id}))
w.Send(api.InitWebrtcStream, api.InitWebrtcStreamRequest{Id: id, Initiator: initiator, Sdp: sdp}))
}
func (w *Worker) WebrtcAnswer(id string, sdp string) {

59
pkg/network/webrtc/webrtc.go
View file

@ -28,7 +28,7 @@ type Decoder func(data string, obj any) error
func New(log *logger.Logger, api *ApiFactory) *Peer { return &Peer{api: api, log: log} }
func (p *Peer) NewCall(vCodec, aCodec string, onICECandidate func(ice any)) (sdp any, err error) {
func (p *Peer) NewCall(vCodec, aCodec string, onICECandidate func(ice any)) (err error) {
if p.conn != nil && p.conn.ConnectionState() == webrtc.PeerConnectionStateConnected {
return
}
@ -36,15 +36,18 @@ func (p *Peer) NewCall(vCodec, aCodec string, onICECandidate func(ice any)) (sdp
if p.conn, err = p.api.NewPeer(); err != nil {
return
}
p.conn.OnConnectionStateChange(func(pcs webrtc.PeerConnectionState) {
p.log.Debug().Msgf("WebRTC state change: %v", pcs)
})
p.conn.OnICECandidate(p.handleICECandidate(onICECandidate))
// plug in the [video] track (out)
video, err := newTrack("video", "video", vCodec)
if err != nil {
return "", err
return err
}
vs, err := p.conn.AddTrack(video)
if err != nil {
return "", err
return err
}
// Read incoming RTCP packets
go func() {
@ -62,11 +65,11 @@ func (p *Peer) NewCall(vCodec, aCodec string, onICECandidate func(ice any)) (sdp
// plug in the [audio] track (out)
audio, err := newTrack("audio", "audio", aCodec)
if err != nil {
return "", err
return err
}
as, err := p.conn.AddTrack(audio)
if err != nil {
return "", err
return err
}
// Read incoming RTCP packets
go func() {
@ -81,17 +84,54 @@ func (p *Peer) NewCall(vCodec, aCodec string, onICECandidate func(ice any)) (sdp
p.log.Debug().Msgf("Added [%s] track", audio.Codec().MimeType)
p.a = audio
err = p.AddChannel("data", func(data []byte) {
p.conn.OnICEConnectionStateChange(p.handleICEState(func() { p.log.Info().Msg("Connected") }))
p.conn.OnDataChannel(func(dc *webrtc.DataChannel) {
p.log.Debug().Msgf(">>>>> Added [%s] track", dc.Label())
if dc.Label() == "data" {
err := p.AddDataChannel()
if err != nil {
p.log.Error().Msgf("Failed to add data channel: %v", err)
}
}
})
return nil
}
func (p *Peer) AddDataChannel() error {
err := p.AddChannel("data", func(data []byte) {
if len(data) == 0 || p.OnMessage == nil {
return
}
p.OnMessage(data)
})
if err != nil {
return err
}
return nil
}
func (p *Peer) Answer() (sdp any, err error) {
answer, err := p.conn.CreateAnswer(&webrtc.AnswerOptions{
OfferAnswerOptions: webrtc.OfferAnswerOptions{ICETricklingSupported: true},
})
if err != nil {
return "", err
}
p.log.Debug().Msg("Created answer")
err = p.conn.SetLocalDescription(answer)
if err != nil {
return "", err
}
p.conn.OnICEConnectionStateChange(p.handleICEState(func() { p.log.Info().Msg("Connected") }))
return answer, nil
}
func (p *Peer) Offer() (sdp any, err error) {
offer, err := p.conn.CreateOffer(&webrtc.OfferOptions{
OfferAnswerOptions: webrtc.OfferAnswerOptions{ICETricklingSupported: true},
})
@ -253,7 +293,10 @@ func (p *Peer) Disconnect() {
// addDataChannel creates new WebRTC data channel.
// Default params -- ordered: true, negotiated: false.
func (p *Peer) addDataChannel(label string) (*webrtc.DataChannel, error) {
ch, err := p.conn.CreateDataChannel(label, nil)
ch, err := p.conn.CreateDataChannel(label, &webrtc.DataChannelInit{
Ordered: new(bool),
MaxRetransmits: new(uint16),
})
if err != nil {
return nil, err
}

34
pkg/worker/coordinatorhandlers.go
View file

@ -30,7 +30,7 @@ func buildConnQuery(id com.Uid, conf config.Worker, address string) (string, err
func (c *coordinator) HandleInitWebrtcStream(rq api.InitWebrtcStreamRequest, w *Worker, factory *webrtc.ApiFactory) api.Out {
peer := webrtc.New(c.log, factory)
localSDP, err := peer.NewCall(w.conf.Encoder.Video.Codec, "opus", func(data any) {
err := peer.NewCall(w.conf.Encoder.Video.Codec, "opus", func(data any) {
candidate, err := toBase64Json(data)
if err != nil {
c.log.Error().Err(err).Msgf("ICE candidate encode fail for [%v]", data)
@ -42,6 +42,36 @@ func (c *coordinator) HandleInitWebrtcStream(rq api.InitWebrtcStreamRequest, w *
c.log.Error().Err(err).Msg("cannot create new webrtc session")
return api.EmptyPacket
}
var localSDP any
if rq.Initiator {
err := peer.SetRemoteSDP(rq.Sdp, fromBase64Json)
if err != nil {
c.log.Error().Err(err).Msgf("cannot set remote SDP of peer [%v]", rq.Id)
return api.EmptyPacket
}
lsdp, err := peer.Answer()
if err != nil {
c.log.Error().Err(err).Msgf("cannot create answer for peer [%v]", rq.Id)
return api.EmptyPacket
}
localSDP = lsdp
} else {
err = peer.AddDataChannel()
if err != nil {
c.log.Error().Err(err).Msgf("cannot add data channel for peer [%v]", rq.Id)
return api.EmptyPacket
}
lsdp, err := peer.Offer()
if err != nil {
c.log.Error().Err(err).Msgf("cannot create offer for peer [%v]", rq.Id)
return api.EmptyPacket
}
localSDP = lsdp
}
sdp, err := toBase64Json(localSDP)
if err != nil {
c.log.Error().Err(err).Msgf("SDP encode fail fro [%v]", localSDP)
@ -221,7 +251,6 @@ func (c *coordinator) HandleGameStart(rq api.StartGameRequest, w *Worker) api.Ou
func (c *coordinator) HandleTerminateSession(rq api.TerminateSessionRequest, w *Worker) {
if user := w.router.FindUser(rq.Id); user != nil {
w.router.Remove(user)
c.log.Debug().Msgf(">>> users: %v", w.router.Users())
user.Disconnect()
}
}
@ -230,7 +259,6 @@ func (c *coordinator) HandleTerminateSession(rq api.TerminateSessionRequest, w *
func (c *coordinator) HandleQuitGame(rq api.GameQuitRequest, w *Worker) {
if user := w.router.FindUser(rq.Id); user != nil {
w.router.Remove(user)
c.log.Debug().Msgf(">>> users: %v", w.router.Users())
}
}

94
web/js/app.js
View file

@ -29,13 +29,6 @@ import {
RECORDING_TOGGLED,
REFRESH_INPUT,
SETTINGS_CHANGED,
WEBRTC_CONNECTION_CLOSED,
WEBRTC_CONNECTION_READY,
WEBRTC_ICE_CANDIDATE_FOUND,
WEBRTC_ICE_CANDIDATE_RECEIVED,
WEBRTC_NEW_CONNECTION,
WEBRTC_SDP_LOCAL,
WEBRTC_SDP_REMOTE,
WORKER_LIST_FETCHED,
pub,
sub,
@ -148,8 +141,10 @@ const showMenuScreen = () => {
setState(app.state.menu);
};
const isConnected = () => webrtc.isConnected();
const startGame = () => {
if (!webrtc.isConnected()) {
if (!isConnected()) {
message.show("Game cannot load. Please refresh");
return;
}
@ -183,16 +178,13 @@ const onMessage = (m) => {
log.debug(`[msg] ${api.endpointName[t] || t}`);
switch (t) {
case api.endpoint.INIT:
pub(WEBRTC_NEW_CONNECTION, payload);
handleWebrtcStart({ data: payload, initiator: true });
break;
case api.endpoint.OFFER:
pub(WEBRTC_SDP_REMOTE, api.fromBase64(payload));
webrtc.setRemoteDescription(api.fromBase64(payload));
break;
case api.endpoint.ICE_CANDIDATE:
pub(
WEBRTC_ICE_CANDIDATE_RECEIVED,
payload ? api.fromBase64(payload) : "",
);
webrtc.addCandidate(payload ? api.fromBase64(payload) : "");
break;
case api.endpoint.GAME_START:
if (payload.av) pub(APP_VIDEO_CHANGED, payload.av);
@ -486,6 +478,56 @@ document.onfullscreenchange = () =>
// subscriptions
sub(MESSAGE, onMessage);
// webrtc
function handleWebrtcStart({ data, initiator }) {
workerManager.whoami(data.wid);
let makingOffer = false;
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));
};
const datachannel = (ch) => {
log.debug("> datachannel", ch.label);
if (ch.label === "data") {
// we'll handle ws and webrtc server messages in one place
ch.onmessage = (x) => onMessage(api.fromBytes(x.data));
}
return ch;
};
webrtc.start({
initiator,
iceServers: data.ice,
media: stream.video.el,
onNegotiationNeeded: negotiate,
onDataChannel: datachannel,
onConnect: onConnectionReady,
onDisconnect: () => input.retropad.toggle(false),
onIceCandidate: api.server.sendIceCandidate,
onSdpAnswer: api.server.sendSdp,
});
if (initiator) {
negotiate();
webrtc.createDataChannel({ onChannel: datachannel });
} else {
api.server.initWebrtcStream();
}
gameList.set(data.games);
}
sub(GAME_ROOM_AVAILABLE, stream.play, 2);
sub(GAME_SAVED, () => message.show("Saved"));
sub(GAME_PLAYER_IDX, (data) => {
@ -496,30 +538,6 @@ sub(GAME_PLAYER_IDX_SET, (idx) => {
});
sub(GAME_ERROR_NO_FREE_SLOTS, () => message.show("No free slots :(", 2500));
// WebRTC connection handling
sub(WEBRTC_NEW_CONNECTION, (data) => {
workerManager.whoami(data.wid);
webrtc.start({ iceServers: data.ice, media: stream.video.el });
webrtc.modDataChannel = (ch) => {
ch.binaryType = "arraybuffer";
if (ch.label === "data") {
ch.onmessage = (x) => onMessage(api.fromBytes(x.data));
}
return ch;
};
api.server.initWebrtcStream();
gameList.set(data.games);
});
sub(WEBRTC_SDP_REMOTE, webrtc.setRemoteDescription);
sub(WEBRTC_SDP_LOCAL, api.server.sendSdp);
sub(WEBRTC_ICE_CANDIDATE_FOUND, api.server.sendIceCandidate);
sub(WEBRTC_ICE_CANDIDATE_RECEIVED, webrtc.addCandidate);
sub(WEBRTC_CONNECTION_READY, onConnectionReady);
sub(WEBRTC_CONNECTION_CLOSED, () => {
input.retropad.toggle(false);
webrtc.stop();
});
sub(LATENCY_CHECK_REQUESTED, onLatencyCheck);
sub(GAMEPAD_CONNECTED, () => message.show("Gamepad connected"));
sub(GAMEPAD_DISCONNECTED, () => message.show("Gamepad disconnected"));

8
web/js/event.js
View file

@ -60,14 +60,6 @@ export const GAME_PLAYER_IDX = "gamePlayerIndex";
export const GAME_PLAYER_IDX_SET = "gamePlayerIndexSet";
export const GAME_ERROR_NO_FREE_SLOTS = "gameNoFreeSlots";
export const WEBRTC_CONNECTION_CLOSED = "webrtcConnectionClosed";
export const WEBRTC_CONNECTION_READY = "webrtcConnectionReady";
export const WEBRTC_ICE_CANDIDATE_FOUND = "webrtcIceCandidateFound";
export const WEBRTC_ICE_CANDIDATE_RECEIVED = "webrtcIceCandidateReceived";
export const WEBRTC_NEW_CONNECTION = "webrtcNewConnection";
export const WEBRTC_SDP_LOCAL = "webrtcSdpLocal";
export const WEBRTC_SDP_REMOTE = "webrtcSdpRemote";
export const MESSAGE = "message";
export const GAMEPAD_CONNECTED = "gamepadConnected";

184
web/js/network/webrtc.js
View file

@ -1,26 +1,19 @@
import {
pub,
WEBRTC_CONNECTION_READY,
WEBRTC_CONNECTION_CLOSED,
WEBRTC_ICE_CANDIDATE_FOUND,
WEBRTC_SDP_LOCAL,
} from "event";
import { log } from "log";
let /** @type {RTCPeerConnection} */ pc;
let /** @type {Map<string, RTCDataChannel>} */ channels = new Map();
let /** @type {MediaStream} */ stream;
let /** @type {RTCLocalIceCandidateInit[]} */ candidateBuf = [];
let /** @type {(channel: RTCDataChannel) => RTCDataChannel} */ modDataChannel;
let handleSdpAnswer;
let _initiator = false;
const ice = ((timeout = 3000) => {
let timeoutId;
const ice = (() => {
let handleIceCandidate;
const onIceCandidate = (/** @type {RTCPeerConnectionIceEvent} */ ev) => {
if (!ev.candidate) return;
log.debug(`[rtc] [ice] local: ${ev.candidate.candidate}`);
pub(WEBRTC_ICE_CANDIDATE_FOUND, ev.candidate);
log.debug(`[rtc] [ice] local`, ev.candidate);
if (handleIceCandidate) handleIceCandidate(ev.candidate);
};
const onIceCandidateError = (
@ -40,17 +33,6 @@ const ice = ((timeout = 3000) => {
const onIceGatheringStateChange = (event) => {
const /** @type {RTCPeerConnection} */ t = event.target;
log.debug(`[rtc] [ice] state: ${t.iceGatheringState}`);
switch (t.iceGatheringState) {
case "gathering":
timeoutId = setTimeout(() => {
log.warn(`[rtc] [ice] stopped due to timeout ${timeout}ms`);
}, timeout);
break;
case "complete":
clearTimeout(timeoutId);
break;
}
};
const onIceConnectionStateChange = () => {
@ -68,20 +50,26 @@ const ice = ((timeout = 3000) => {
onIceCandidateError,
onIceGatheringStateChange,
onIceConnectionStateChange,
set handleIceCandidate(cb) {
handleIceCandidate = cb;
},
};
})();
const isConnected = () => pc?.connectionState === "connected";
const hasRemoteDescription = () => pc?.remoteDescription !== null;
const addRemoteCandidate = (data) => {
if (!data) return;
pc.addIceCandidate(new RTCIceCandidate(data)).catch((e) => {
const candidate = new RTCIceCandidate(data);
pc.addIceCandidate(candidate).catch((e) => {
log.error("[rtc] [ice] remote candidate add failed", e.name);
});
log.debug(`[rtc] [ice] added remote: ${data.candidate}`);
log.debug(`[rtc] [ice] added remote`, candidate);
};
const flushRemoteCandidates = () => {
// this will work only when the remote description is set
if (!pc.remoteDescription || candidateBuf.length === 0) return;
if (!hasRemoteDescription() || candidateBuf.length === 0) return;
log.debug(`[rtc] [ice] remote candidate buf (${candidateBuf.length})`);
let data = undefined;
@ -90,21 +78,65 @@ const flushRemoteCandidates = () => {
}
};
const isConnected = () => pc?.connectionState === "connected";
// hacks
// Chrome bug https://bugs.chromium.org/p/chromium/issues/detail?id=818180 workaround
// force stereo params for Opus tracks (a=fmtp:111 ...)
const enableOpusStereo = (sdp) =>
sdp.replace(/(a=fmtp:111 .*)/g, "$1;stereo=1");
const pushChannel = (chan) => {
channels.set(chan.label, chan);
log.debug(`[rtc] [data-ch] push: ${chan.label}`);
};
const stop = () => {
if (stream) {
while (stream.getTracks().length > 0) {
const t = stream.getTracks()[0];
t.stop();
stream.removeTrack(t);
}
stream = null;
}
if (pc) {
ice.handleIceCandidate = null;
handleSdpAnswer = null;
pc.oniceconnectionstatechange = null;
pc.onicegatheringstatechange = null;
pc.onicecandidate = null;
pc.onicecandidateerror = null;
pc.onconnectionstatechange = null;
pc.ondatachannel = null;
pc.ontrack = null;
pc.close();
pc = null;
}
for (const [, channel] of channels) {
channel.close();
}
channels.clear();
candidateBuf = [];
log.debug("[rtc] WebRTC has been closed");
};
/**
* WebRTC connection module.
*/
export const webrtc = {
start: ({ iceServers = [], media, initiator = false } = {}) => {
start: ({
iceServers = [],
media,
initiator = false,
onNegotiationNeeded,
onDataChannel,
onConnect,
onDisconnect,
onIceCandidate,
onSdpAnswer,
} = {}) => {
log.debug("[rtc] got remote ICE servers", iceServers);
pc = new RTCPeerConnection({ iceCandidatePoolSize: 16, iceServers });
pc = new RTCPeerConnection({ iceCandidatePoolSize: 1, iceServers });
stream = new MediaStream();
_initiator = initiator;
@ -118,14 +150,17 @@ export const webrtc = {
);
}
if (onIceCandidate) ice.handleIceCandidate = onIceCandidate;
if (onSdpAnswer) handleSdpAnswer = onSdpAnswer;
pc.addTransceiver("video", { direction: "recvonly" });
pc.addTransceiver("audio", { direction: "recvonly" });
pc.ondatachannel = (ev) => {
let chan = modDataChannel ? modDataChannel(ev.channel) : ev.channel;
channels.set(chan.label, chan);
log.debug(`[rtc] [data-ch] push: ${chan.label}`);
let chan = onDataChannel ? onDataChannel(ev.channel) : ev.channel;
pushChannel(chan);
};
pc.oniceconnectionstatechange = ice.onIceConnectionStateChange;
pc.onicegatheringstatechange = ice.onIceGatheringStateChange;
pc.onicecandidate = ice.onIceCandidate;
@ -135,31 +170,38 @@ export const webrtc = {
switch (pc.connectionState) {
case "connected":
pub(WEBRTC_CONNECTION_READY);
if (onConnect) onConnect();
break;
case "failed":
case "closed":
pub(WEBRTC_CONNECTION_CLOSED);
if (onDisconnect) onDisconnect();
stop();
break;
}
};
pc.onnegotiationneeded = () => {
log.debug("[rtc] negotiation needed");
// todo implement
// pc.createOffer()
// .then((description) =>
// pc.setLocalDescription(description).catch(log.error),
// )
// .catch(log.error);
};
pc.ontrack = (event) => {
stream.addTrack(event.track);
if (onNegotiationNeeded) onNegotiationNeeded();
};
pc.ontrack = (event) => stream.addTrack(event.track);
},
offerSdp: async () => {
if (!pc || !_initiator) return;
try {
const offer = await pc.createOffer();
offer.sdp = enableOpusStereo(offer.sdp);
await pc.setLocalDescription(offer);
log.debug("[rtc] [sdp] local offer", offer);
return offer;
} catch (e) {
log.error(`[rtc] [sdp] local offer error: ${e}`);
}
},
setRemoteDescription: async (
/** @type {RTCSessionDescriptionInit} */ sdp,
) => {
log.debug("[rtc] [sdp] remote offer", sdp);
log.debug("[rtc] [sdp] remote SDP", sdp);
try {
const offer = new RTCSessionDescription(sdp);
@ -172,22 +214,23 @@ export const webrtc = {
flushRemoteCandidates();
if (_initiator) return;
try {
const answer = await pc.createAnswer();
answer.sdp = enableOpusStereo(answer.sdp);
await pc.setLocalDescription(answer);
log.debug("[rtc] [sdp] local answer", answer);
pub(WEBRTC_SDP_LOCAL, answer);
handleSdpAnswer(answer);
} catch (e) {
log.error(`[rtc] [sdp] local answer error: ${e}`);
}
},
pushChannel,
addCandidate: (
/** @type {RTCLocalIceCandidateInit | string} */ candidate,
) => {
const allowed = pc.remoteDescription !== null;
if (allowed) {
if (hasRemoteDescription()) {
addRemoteCandidate(candidate);
} else {
candidateBuf.push(candidate);
@ -206,35 +249,18 @@ export const webrtc = {
if (!isConnected()) return Promise.resolve();
return await pc.getStats();
},
stop: () => {
if (stream) {
while (stream.getTracks().length > 0) {
const t = stream.getTracks()[0];
t.stop();
stream.removeTrack(t);
}
stream = null;
createDataChannel: ({ onChannel }) => {
try {
let ch = pc.createDataChannel("data", {
ordered: false,
maxRetransmits: 0,
});
ch = onChannel ? onChannel(ch) : ch;
if (!ch) throw new Error("null channel");
channels.set(ch.label, ch);
} catch (e) {
log.error("[rtc] failed to create data channel", e);
}
if (pc) {
pc.oniceconnectionstatechange = null;
pc.onicegatheringstatechange = null;
pc.onicecandidate = null;
pc.onicecandidateerror = null;
pc.onconnectionstatechange = null;
pc.ondatachannel = null;
pc.ontrack = null;
pc.close();
pc = null;
}
for (const [, channel] of channels) {
channel.close();
}
channels.clear();
candidateBuf = [];
log.debug("[rtc] WebRTC has been closed");
},
set modDataChannel(fn) {
modDataChannel = fn;
},
stop,
};