diff --git a/pkg/coordinator/userapi.go b/pkg/coordinator/userapi.go index 6da571b5..6a69ea40 100644 --- a/pkg/coordinator/userapi.go +++ b/pkg/coordinator/userapi.go @@ -26,14 +26,9 @@ func (u *User) InitSession(wid string, ice []config.IceServer, games []api.AppMe }) } -// SendWebrtcOffer sends SDP offer back to the user. -func (u *User) SendWebrtcOffer(sdp string) { - u.Notify(api.WebrtcSignal, api.WebrtcSignalUser{Sdp: &sdp}) -} - -// SendWebrtcIceCandidate sends remote ICE candidate back to the user. -func (u *User) SendWebrtcIceCandidate(candidate string) { - u.Notify(api.WebrtcSignal, api.WebrtcSignalUser{Ice: &candidate}) +// SendSignal sends webrtc signal back to the user. +func (u *User) SendSignal(sdp *string, ice *string) { + u.Notify(api.WebrtcSignal, api.WebrtcSignalUser{Sdp: sdp, Ice: ice}) } // StartGame signals the user that everything is ready to start a game. diff --git a/pkg/coordinator/userhandlers.go b/pkg/coordinator/userhandlers.go index 51e805ca..395978d4 100644 --- a/pkg/coordinator/userhandlers.go +++ b/pkg/coordinator/userhandlers.go @@ -18,7 +18,6 @@ func (u *User) HandleInitWebrtcStream(rq api.InitUserWebrtcStreamRequest) { u.log.Error().Err(err).Msg("malformed WebRTC init response") return } - u.SendWebrtcOffer(string(*resp)) } func (u *User) HandleWebrtcSignal(rq api.WebrtcSignalUser) { diff --git a/pkg/coordinator/worker.go b/pkg/coordinator/worker.go index 6770af3a..ad2ba608 100644 --- a/pkg/coordinator/worker.go +++ b/pkg/coordinator/worker.go @@ -87,10 +87,7 @@ func (w *Worker) HandleRequests(users HasUserRegistry) chan struct{} { err = api.Do(p, w.HandleCloseRoom) case api.WebrtcSignal: err = api.DoE(p, func(rq api.WebrtcSignalRequest) error { - if rq.Ice == nil { - return fmt.Errorf("ice candidate is missing") - } - return w.HandleIceCandidate(rq, users) + return w.HandleSignal(rq, users) }) case api.LibNewGameList: err = api.DoE(p, w.HandleLibGameList) diff --git a/pkg/coordinator/workerhandlers.go b/pkg/coordinator/workerhandlers.go index 84dd0ee1..1bd60f0c 100644 --- a/pkg/coordinator/workerhandlers.go +++ b/pkg/coordinator/workerhandlers.go @@ -1,6 +1,10 @@ package coordinator -import "github.com/giongto35/cloud-game/v3/pkg/api" +import ( + "fmt" + + "github.com/giongto35/cloud-game/v3/pkg/api" +) func (w *Worker) HandleRegisterRoom(rq api.RegisterRoomRequest) { w.RoomId = string(rq) } @@ -11,12 +15,17 @@ func (w *Worker) HandleCloseRoom(rq api.CloseRoomRequest) { } } -func (w *Worker) HandleIceCandidate(rq api.WebrtcSignalRequest, users HasUserRegistry) error { - if usr := users.Find(rq.Id); usr != nil { - usr.SendWebrtcIceCandidate(*rq.Ice) - } else { - w.log.Warn().Str("id", rq.Id).Msg("unknown session") +func (w *Worker) HandleSignal(rq api.WebrtcSignalRequest, users HasUserRegistry) error { + if rq.Ice == nil && rq.Sdp == nil { + return fmt.Errorf("ice candidate or SDP are missing") } + + user := users.Find(rq.Id) + if user == nil { + w.log.Warn().Str("id", rq.Id).Msg("unknown session") + return fmt.Errorf("unknown session") + } + user.SendSignal(rq.Sdp, rq.Ice) return nil } diff --git a/pkg/network/webrtc/ice.go b/pkg/network/webrtc/ice.go deleted file mode 100644 index 21f6b8ea..00000000 --- a/pkg/network/webrtc/ice.go +++ /dev/null @@ -1,34 +0,0 @@ -package webrtc - -import ( - "sync" - - "github.com/pion/webrtc/v4" -) - -type IceCandidateBuffer struct { - mu sync.Mutex - buf []webrtc.ICECandidateInit -} - -func (b *IceCandidateBuffer) push(c webrtc.ICECandidateInit) { - b.mu.Lock() - defer b.mu.Unlock() - b.buf = append(b.buf, c) -} - -// FlushAll atomically swaps the internal buffer with a new, empty one -// and returns the old buffer's contents for processing. -func (b *IceCandidateBuffer) FlushAll() []webrtc.ICECandidateInit { - b.mu.Lock() - defer b.mu.Unlock() - oldBuffer := b.buf - b.buf = nil // Or b.buf = make([]Candidate, 0, initialCapacity) - return oldBuffer -} - -func (b *IceCandidateBuffer) Clear() { - b.mu.Lock() - defer b.mu.Unlock() - b.buf = nil -} diff --git a/pkg/network/webrtc/webrtc.go b/pkg/network/webrtc/webrtc.go index a4f8a833..6f6a2488 100644 --- a/pkg/network/webrtc/webrtc.go +++ b/pkg/network/webrtc/webrtc.go @@ -1,6 +1,7 @@ package webrtc import ( + "encoding/json" "fmt" "strings" "sync" @@ -21,13 +22,17 @@ type Peer struct { v *webrtc.TrackLocalStaticSample d *webrtc.DataChannel - icb *IceCandidateBuffer + // flags + makingOffer bool + ignoreOffer bool + isSettingRemoteAnswerPending bool + polite bool + + signaller func(ice, sdp *string) } var samplePool sync.Pool -type Decoder func(data string, obj any) error - func New(log *logger.Logger, api *ApiFactory) *Peer { // hide directions (->) and clients (w, c, ...) custom := log.Extend( @@ -35,17 +40,109 @@ func New(log *logger.Logger, api *ApiFactory) *Peer { Str(logger.DirectionField, ""). Str(logger.ClientField, ""), ) - return &Peer{api: api, log: custom, icb: &IceCandidateBuffer{}} + return &Peer{api: api, log: custom} } -func (p *Peer) NewConnection(vCodec, aCodec string, onICECandidate func(ice any)) (err error) { - if p.conn != nil && p.conn.ConnectionState() == webrtc.PeerConnectionStateConnected { +func (p *Peer) NewConnection(vCodec, aCodec string, signal func(*string, *string)) (err error) { + p.log.Debug().Msg("rtc start") + + pc := p.conn + p.signaller = signal + + if pc != nil && pc.ConnectionState() == webrtc.PeerConnectionStateConnected { return } - p.log.Debug().Msg("rtc start") + if p.conn, err = p.api.NewPeer(); err != nil { return } + pc = p.conn + + pc.OnConnectionStateChange(func(pcs webrtc.PeerConnectionState) { + p.log.Debug(). + Str("state", pcs.String()). + Msg("rtc [connection] state change") + if pcs == webrtc.PeerConnectionStateConnected { + p.log.Info(). + Str(logger.DirectionField, "rtc"). + Str(logger.ClientField, ""). + Msg("(connected)") + } + }) + pc.OnICECandidate(func(ice *webrtc.ICECandidate) { + if ice == nil { + signal(new(string), nil) + p.log.Debug().Msg("rtc [ice] gathering (complete)") + return + } + candidate := ice.ToJSON() + p.log.Debug().Str("local", candidate.Candidate).Msg("rtc [ice] candidate") + + encoded, err := toJson(candidate) + if err != nil { + p.log.Error().Err(err).Msg("rtc [ice] candidate encoding failed") + return + } + + signal(&encoded, nil) + }) + pc.OnSignalingStateChange(func(ss webrtc.SignalingState) { + p.log.Debug().Str("state", ss.String()).Msg("rtc [signal] state change") + }) + pc.OnICEConnectionStateChange(p.handleICEState(func() {})) + pc.OnDataChannel(func(ch *webrtc.DataChannel) { + p.log.Debug().Msgf("rtc [chan] [%s] remote", ch.Label()) + }) + + p.makingOffer = false + + pc.OnNegotiationNeeded(func() { + p.log.Debug().Msg("rtc [negotiation] needed") + defer func() { p.makingOffer = false }() + + p.makingOffer = true + offer, err := p.Offer() + if err != nil { + p.log.Error().Err(err).Msg("rtc [negotiation] failed") + return + } + + // if p.conn.SignalingState() != webrtc.SignalingStateStable { + // p.log.Debug().Msg("rtc [negotiation] waiting for signaling state stable") + // return + // } + + sdp, err := toJson(offer) + if err != nil { + p.log.Error().Err(err).Msg("rtc [negotiation] failed") + return + } + + signal(nil, &sdp) + }) + + if p.v, err = NewTrack("video", "video", vCodec); err != nil { + return err + } + p.log.Debug().Msgf("rtc [media] added [%s] track", p.v.Codec().MimeType) + + if p.a, err = NewTrack("audio", "audio", aCodec); err != nil { + return err + } + p.log.Debug().Msgf("rtc [media] added [%s] track", p.a.Codec().MimeType) + + sendOnly := webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionSendonly} + + if _, err = p.conn.AddTransceiverFromTrack(p.v, sendOnly); err != nil { + panic(err) + } + if _, err = p.conn.AddTransceiverFromTrack(p.a, sendOnly); err != nil { + panic(err) + } + + for _, rtpSender := range p.conn.GetSenders() { + go processRTCP(rtpSender) + } id := uint16(0) negotiated := true @@ -68,95 +165,20 @@ func (p *Peer) NewConnection(vCodec, aCodec string, onICECandidate func(ice any) return err } - p.conn.OnConnectionStateChange(func(pcs webrtc.PeerConnectionState) { - p.log.Debug(). - Str("state", pcs.String()). - Msg("rtc [connection] state change") - if pcs == webrtc.PeerConnectionStateConnected { - p.log.Info(). - Str(logger.DirectionField, "rtc"). - Str(logger.ClientField, ""). - Msg("(connected)") - } - }) - p.conn.OnICECandidate(p.handleICECandidate(onICECandidate)) - - p.conn.OnSignalingStateChange(func(ss webrtc.SignalingState) { - p.log.Debug(). - Str("state", ss.String()). - Msg("rtc [signal] state change") - if ss == webrtc.SignalingStateStable { - p.flushPendingCandidates() - } - }) - - // plug in the [video] track (out) - if p.v, err = p.AddTrack("video", "video", vCodec); err != nil { - return err - } - - // plug in the [audio] track (out) - if p.a, err = p.AddTrack("audio", "audio", aCodec); err != nil { - return err - } - - p.conn.OnICEConnectionStateChange(p.handleICEState(func() {})) - - p.conn.OnDataChannel(func(ch *webrtc.DataChannel) { - p.log.Debug().Msgf("rtc [chan] [%s] remote", ch.Label()) - }) - - p.conn.OnNegotiationNeeded(func() { - p.log.Debug().Msg("rtc [negotiation] needed") - }) - return nil } -func (p *Peer) AddTrack(id, label, codec string) (*webrtc.TrackLocalStaticSample, error) { - track, err := newTrack(id, label, codec) - if err != nil { - return nil, err - } - as, err := p.conn.AddTrack(track) - if err != nil { - return nil, err - } - // Read incoming RTCP packets - go func() { - buf := make([]byte, 1500) - for { - if _, _, err := as.Read(buf); err != nil { - return - } - } - }() - p.log.Debug().Msgf("rtc [media] added [%s] track", track.Codec().MimeType) - return track, nil -} - -func (p *Peer) OfferAnswer(offer bool) (*webrtc.SessionDescription, error) { +func (p *Peer) Offer() (*webrtc.SessionDescription, error) { opts := webrtc.OfferAnswerOptions{ICETricklingSupported: true} - var sdp webrtc.SessionDescription - var err error - - if offer { - sdp, err = p.conn.CreateOffer(&webrtc.OfferOptions{OfferAnswerOptions: opts}) - } else { - sdp, err = p.conn.CreateAnswer(&webrtc.AnswerOptions{OfferAnswerOptions: opts}) - } + sdp, err := p.conn.CreateOffer(&webrtc.OfferOptions{OfferAnswerOptions: opts}) if err != nil { return nil, err } - if err = p.conn.SetLocalDescription(sdp); err != nil { return nil, err } - - p.log.Debug(). - Str("type", sdp.Type.String()). - Msg("rtc [sdp] set (local)") + p.log.Debug().Str("type", sdp.Type.String()).Msg("rtc [sdp] set (local)") return &sdp, nil } @@ -190,23 +212,47 @@ func (p *Peer) send(data []byte, duration int64, fn func(media.Sample) error) er return nil } -func (p *Peer) SetRemoteSDP(sdp string, decoder Decoder) error { +func (p *Peer) SetDescription(sdp string) error { var answer webrtc.SessionDescription - if err := decoder(sdp, &answer); err != nil { + if err := fromJson(sdp, &answer); err != nil { return err } + + readyForOffer := !p.makingOffer && + (p.conn.SignalingState() == webrtc.SignalingStateStable || p.isSettingRemoteAnswerPending) + + offerCollision := answer.Type == webrtc.SDPTypeOffer && !readyForOffer + + ignoreOffer := !p.polite && offerCollision + if ignoreOffer { + return nil + } + p.isSettingRemoteAnswerPending = answer.Type == webrtc.SDPTypeAnswer if err := p.conn.SetRemoteDescription(answer); err != nil { p.log.Error().Err(err).Msg("Set remote description from peer failed") return err } - p.flushPendingCandidates() - p.log.Debug(). - Str("type", answer.Type.String()). - Msg("rtc [sdp] set (remote)") + p.isSettingRemoteAnswerPending = false + if answer.Type == webrtc.SDPTypeOffer { + err := p.conn.SetLocalDescription(answer) + if err != nil { + p.log.Error().Err(err).Msg("Set remote description from peer failed") + return err + } + + sdp, err := toJson(answer) + if err != nil { + return err + } + + p.signaller(nil, &sdp) + } + p.log.Debug().Str("type", answer.Type.String()).Msg("rtc [sdp] set (remote)") + return nil } -func newTrack(id string, label string, codec string) (*webrtc.TrackLocalStaticSample, error) { +func NewTrack(id string, label string, codec string) (*webrtc.TrackLocalStaticSample, error) { codec = strings.ToLower(codec) var mime string switch id { @@ -231,20 +277,6 @@ func newTrack(id string, label string, codec string) (*webrtc.TrackLocalStaticSa return webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: mime}, id, label) } -func (p *Peer) handleICECandidate(callback func(any)) func(*webrtc.ICECandidate) { - return func(ice *webrtc.ICECandidate) { - // ICE gathering finish condition - if ice == nil { - callback(nil) - p.log.Debug().Msg("rtc [ice] gathering (complete)") - return - } - candidate := ice.ToJSON() - p.log.Debug().Str("local", candidate.Candidate).Msg("rtc [ice] candidate") - callback(&candidate) - } -} - func (p *Peer) handleICEState(onConnect func()) func(webrtc.ICEConnectionState) { return func(state webrtc.ICEConnectionState) { p.log.Debug().Str("state", state.String()).Msg("rtc [ice] connection state change") @@ -267,17 +299,18 @@ func (p *Peer) handleICEState(onConnect func()) func(webrtc.ICEConnectionState) } } -func (p *Peer) AddCandidate(candidate string, decoder Decoder) error { +func (p *Peer) AddCandidate(candidate string) error { var iceCandidate webrtc.ICECandidateInit - if err := decoder(candidate, &iceCandidate); err != nil { + if err := fromJson(candidate, &iceCandidate); err != nil { return err } - p.log.Debug(). - Str("remote", iceCandidate.Candidate). - Msg("rtc [ice] candidate") - buffered := p.conn.RemoteDescription() == nil || - p.conn.SignalingState() != webrtc.SignalingStateStable - return p.addCandidate(iceCandidate, buffered) + p.log.Debug().Str("remote", iceCandidate.Candidate).Msg("rtc [ice] candidate") + + err := p.conn.AddICECandidate(iceCandidate) + if p.ignoreOffer { + err = nil + } + return err } func (p *Peer) AddChannel(label string, conf *webrtc.DataChannelInit, onMessage func([]byte)) (*webrtc.DataChannel, error) { @@ -316,38 +349,34 @@ func (p *Peer) Disconnect() { // ignore this due to DTLS fatal: conn is closed _ = p.conn.Close() } - p.icb.Clear() p.log.Debug().Msg("rtc stopped") } -func (p *Peer) addCandidate(candidate webrtc.ICECandidateInit, wait bool) error { - if wait { - p.icb.push(candidate) - return nil - } - if err := p.conn.AddICECandidate(candidate); err != nil { - return err - } - return nil -} - -func (p *Peer) flushPendingCandidates() { - prev := p.icb.FlushAll() - if len(prev) == 0 { - return - } - - p.log.Debug().Msg(fmt.Sprintf("rtc [ice] buf (%d) flush", len(prev))) - - for _, candidate := range prev { - if err := p.addCandidate(candidate, false); err != nil { - p.log.Error(). - Str("remote", candidate.Candidate). - Err(err). - Msg("rtc [ice] add") +// Read incoming RTCP packets +// Before these packets are returned they are processed by interceptors. For things +// like NACK this needs to be called. +func processRTCP(rtpSender *webrtc.RTPSender) { + rtcpBuf := make([]byte, 1500) + for { + if _, _, rtcpErr := rtpSender.Read(rtcpBuf); rtcpErr != nil { + return } } - clear(prev) +} + +func fromJson(data string, obj any) error { + return json.Unmarshal([]byte(data), obj) +} + +func toJson(data any) (string, error) { + if data == nil { + return "", nil + } + b, err := json.Marshal(data) + if err != nil { + return "", err + } + return string(b), nil } func (p *Peer) logx(err error) { p.log.Error().Err(err) } diff --git a/pkg/worker/coordinator.go b/pkg/worker/coordinator.go index 2840a0e4..43c2ed57 100644 --- a/pkg/worker/coordinator.go +++ b/pkg/worker/coordinator.go @@ -107,10 +107,12 @@ func (c *coordinator) RegisterRoom(id string) { c.Notify(api.RegisterRoom, id) } // CloseRoom sends a signal to coordinator which will remove that room from its list. func (c *coordinator) CloseRoom(id string) { c.Notify(api.CloseRoom, id) } -func (c *coordinator) IceCandidate(candidate string, sessionId string) { + +func (c *coordinator) Signal(ice *string, sdp *string, id string) { c.Notify(api.WebrtcSignal, api.WebrtcSignalRequest{ - Stateful: api.Stateful{Id: sessionId}, - Ice: &candidate, + Stateful: api.Stateful{Id: id}, + Ice: ice, + Sdp: sdp, }) } diff --git a/pkg/worker/coordinatorhandlers.go b/pkg/worker/coordinatorhandlers.go index 560b6134..bbb870da 100644 --- a/pkg/worker/coordinatorhandlers.go +++ b/pkg/worker/coordinatorhandlers.go @@ -1,6 +1,8 @@ package worker import ( + "encoding/json" + "github.com/giongto35/cloud-game/v3/pkg/api" "github.com/giongto35/cloud-game/v3/pkg/com" "github.com/giongto35/cloud-game/v3/pkg/config" @@ -9,7 +11,6 @@ import ( "github.com/giongto35/cloud-game/v3/pkg/worker/caged" "github.com/giongto35/cloud-game/v3/pkg/worker/media" "github.com/giongto35/cloud-game/v3/pkg/worker/room" - "github.com/goccy/go-json" ) // buildConnQuery builds initial connection data query to a coordinator. @@ -28,43 +29,19 @@ 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) - err := peer.NewConnection(w.conf.Encoder.Video.Codec, "opus", func(data any) { - candidate, err := toJson(data) - if err != nil { - c.log.Error().Err(err).Msgf("ICE candidate encode fail for [%v]", data) - return - } - c.IceCandidate(candidate, rq.Id) + err := peer.NewConnection(w.conf.Encoder.Video.Codec, "opus", func(ice, sdp *string) { + c.Signal(ice, sdp, rq.Id) }) if err != nil { c.log.Error().Err(err).Msg("cannot create new webrtc session") return api.EmptyPacket } - if rq.Initiator { - if err := peer.SetRemoteSDP(rq.Sdp, fromJson); err != nil { - c.log.Error().Err(err).Msgf("cannot set remote SDP of peer [%v]", rq.Id) - return api.EmptyPacket - } - } - - lsdp, err := peer.OfferAnswer(!rq.Initiator) - if err != nil { - c.log.Error().Err(err).Msgf("cannot create SDP for peer [%v]", rq.Id) - return api.EmptyPacket - } - - sdp, err := toJson(*lsdp) - if err != nil { - c.log.Error().Err(err).Msgf("SDP encode fail for [%v]", *lsdp) - return api.EmptyPacket - } - user := room.NewGameSession(rq.Id, peer) // use user uid from the coordinator c.log.Info().Msgf("Peer connection: %s", user.Id()) w.router.AddUser(user) - return api.Out{Payload: sdp} + return api.OkPacket } func (c *coordinator) HandleWebrtcSignal(rq api.WebrtcSignalRequest, w *Worker) { @@ -75,13 +52,13 @@ func (c *coordinator) HandleWebrtcSignal(rq api.WebrtcSignalRequest, w *Worker) } if rq.Sdp != nil { - if err := room.WithWebRTC(user.Session).SetRemoteSDP(*rq.Sdp, fromJson); err != nil { + if err := room.WithWebRTC(user.Session).SetDescription(*rq.Sdp); err != nil { c.log.Error().Err(err).Msgf("cannot set remote SDP of client [%v]", rq.Id) } } if rq.Ice != nil { - if err := room.WithWebRTC(user.Session).AddCandidate(*rq.Ice, fromJson); err != nil { + if err := room.WithWebRTC(user.Session).AddCandidate(*rq.Ice); err != nil { c.log.Error().Err(err).Msgf("cannot add ICE candidate of the client [%v]", rq.Id) } } @@ -302,10 +279,6 @@ func (c *coordinator) HandleRecordGame(rq api.RecordGameRequest, w *Worker) api. return api.OkPacket } -func fromJson(data string, obj any) error { - return json.Unmarshal([]byte(data), obj) -} - func toJson(data any) (string, error) { if data == nil { return "", nil