From e3155b146be9db6da09a2171b3b77597b2d0d719 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Fri, 25 Jul 2025 14:49:04 +0530 Subject: [PATCH] Get to the point of establishing subscriber peer connection. (#3821) Proper message id checks. Some refactoring for common code in rtcv2service handlers. Instantiate signalling based on synchronous local candidates flag, but needs better defined code. --- go.mod | 2 +- go.sum | 4 +- pkg/rtc/participant.go | 57 ++++++++++---------- pkg/rtc/signalling/signalhandlerv2.go | 5 +- pkg/service/rtcv2service.go | 75 ++++++++++++--------------- pkg/service/utils.go | 1 + 6 files changed, 68 insertions(+), 76 deletions(-) diff --git a/go.mod b/go.mod index 070c2da52..f4f240d09 100644 --- a/go.mod +++ b/go.mod @@ -23,7 +23,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 github.com/livekit/mediatransportutil v0.0.0-20250519131108-fb90f5acfded - github.com/livekit/protocol v1.39.4-0.20250724172257-afef3ef967cc + github.com/livekit/protocol v1.39.4-0.20250725083335-7313a8195a4b github.com/livekit/psrpc v0.6.1-0.20250724161801-262a822e7cd7 github.com/mackerelio/go-osstat v0.2.5 github.com/magefile/mage v1.15.0 diff --git a/go.sum b/go.sum index 69e5e1af9..296576c2f 100644 --- a/go.sum +++ b/go.sum @@ -167,8 +167,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20250519131108-fb90f5acfded h1:ylZPdnlX1RW9Z15SD4mp87vT2D2shsk0hpLJwSPcq3g= github.com/livekit/mediatransportutil v0.0.0-20250519131108-fb90f5acfded/go.mod h1:mSNtYzSf6iY9xM3UX42VEI+STHvMgHmrYzEHPcdhB8A= -github.com/livekit/protocol v1.39.4-0.20250724172257-afef3ef967cc h1:kF9ju/WEqtzMhI3P9hCUcgoh1mFEuq7DuWPkji6iq/M= -github.com/livekit/protocol v1.39.4-0.20250724172257-afef3ef967cc/go.mod h1:YlgUxAegtU8jZ0tVXoIV/4fHeHqqLvS+6JnPKDbpFPU= +github.com/livekit/protocol v1.39.4-0.20250725083335-7313a8195a4b h1:uxvoeGd0vmGDIL0JyLLV9h2o97tpt3rR9s4ikuLVz/g= +github.com/livekit/protocol v1.39.4-0.20250725083335-7313a8195a4b/go.mod h1:YlgUxAegtU8jZ0tVXoIV/4fHeHqqLvS+6JnPKDbpFPU= github.com/livekit/psrpc v0.6.1-0.20250724161801-262a822e7cd7 h1:x50axcjXwfwnII7sMhJPyZ6f5LpPapZtsp75KJX8nIQ= github.com/livekit/psrpc v0.6.1-0.20250724161801-262a822e7cd7/go.mod h1:kmD+AZPkWu0MaXIMv57jhNlbiSZZ/Jx4bzlxBDVmJes= github.com/mackerelio/go-osstat v0.2.5 h1:+MqTbZUhoIt4m8qzkVoXUJg1EuifwlAJSk4Yl2GXh+o= diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index d1cb5820b..b1358f9a0 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -313,8 +313,8 @@ type ParticipantImpl struct { metricsCollector *metric.MetricsCollector metricsReporter *metric.MetricsReporter - signalhandler signalling.ParticipantSignalHandler signalling signalling.ParticipantSignalling + signalhandler signalling.ParticipantSignalHandler signaller signalling.ParticipantSignaller // loggers for publisher and subscriber @@ -353,30 +353,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { joiningMessageLastWrittenSeqs: make(map[livekit.ParticipantID]uint32), }, } - p.signalhandler = signalling.NewSignalHandler(signalling.SignalHandlerParams{ - Logger: params.Logger, - Participant: p, - }) - p.signalling = signalling.NewSignalling(signalling.SignallingParams{ - Logger: params.Logger, - }) - p.signaller = signalling.NewSignallerAsync(signalling.SignallerAsyncParams{ - Logger: params.Logger, - Participant: p, - }) - /* SIGNALLING-V2-TODO: instantiate these based on signalling version and signal transport - p.signalhandler = signalling.NewSignalHandlerv2(signalling.SignalHandlerv2Params{ - Logger: params.Logger, - Participant: p, - }) - p.signalling = signalling.NewSignallingv2(signalling.Signallingv2Params{ - Logger: params.Logger, - }) - p.signaller = signalling.NewSignallerv2Async(signalling.Signallerv2AsyncParams{ - Logger: params.Logger, - Participant: p, - }) - */ + p.setupSignalling() p.id.Store(params.SID) p.dataChannelStats = telemetry.NewBytesTrackStats( @@ -1841,7 +1818,35 @@ func (h PrimaryTransportHandler) OnFullyEstablished() { h.p.onPrimaryTransportFullyEstablished() } -// ---------------------------------------------------------- +func (p *ParticipantImpl) setupSignalling() { + // SIGNALLING-V2-TODO: do proper types to decide which signalling components to instantiate + if !p.params.SynchronousLocalCandidatesMode { + p.signalling = signalling.NewSignalling(signalling.SignallingParams{ + Logger: p.params.Logger, + }) + p.signalhandler = signalling.NewSignalHandler(signalling.SignalHandlerParams{ + Logger: p.params.Logger, + Participant: p, + }) + p.signaller = signalling.NewSignallerAsync(signalling.SignallerAsyncParams{ + Logger: p.params.Logger, + Participant: p, + }) + } else { + p.signalling = signalling.NewSignallingv2(signalling.Signallingv2Params{ + Logger: p.params.Logger, + }) + p.signalhandler = signalling.NewSignalHandlerv2(signalling.SignalHandlerv2Params{ + Logger: p.params.Logger, + Participant: p, + Signalling: p.signalling, + }) + p.signaller = signalling.NewSignallerv2Async(signalling.Signallerv2AsyncParams{ + Logger: p.params.Logger, + Participant: p, + }) + } +} func (p *ParticipantImpl) setupTransportManager() error { p.twcc = twcc.NewTransportWideCCResponder() diff --git a/pkg/rtc/signalling/signalhandlerv2.go b/pkg/rtc/signalling/signalhandlerv2.go index 43e8f6a51..3ca35951f 100644 --- a/pkg/rtc/signalling/signalhandlerv2.go +++ b/pkg/rtc/signalling/signalhandlerv2.go @@ -70,7 +70,6 @@ func (s *signalhandlerv2) HandleRequest(msg proto.Message) error { switch msg := req.GetMessage().(type) { case *livekit.Signalv2WireMessage_Envelope: for _, clientMessage := range msg.Envelope.ClientMessages { - /* SIGNALLING-V2-TODO: uncommment once remote side sends proper messageId sequencer := clientMessage.GetSequencer() if sequencer == nil || sequencer.MessageId == 0 { s.params.Logger.Warnw( @@ -90,6 +89,7 @@ func (s *signalhandlerv2) HandleRequest(msg proto.Message) error { continue } + // SIGNALLING-V2-TODO: ask for replay if there are gaps if lprmi != 0 && sequencer.MessageId != lprmi+1 { s.params.Logger.Infow( "gap in message stream", @@ -97,7 +97,6 @@ func (s *signalhandlerv2) HandleRequest(msg proto.Message) error { "current", clientMessage.Sequencer.MessageId, ) } - */ switch payload := clientMessage.GetMessage().(type) { case *livekit.Signalv2ClientMessage_PublisherSdp: @@ -115,11 +114,9 @@ func (s *signalhandlerv2) HandleRequest(msg proto.Message) error { s.params.Participant.AddICECandidate(candidateInit, payload.Trickle.Target) } - /* SIGNALLING-V2-TODO: uncomment once sequencer is implemented on both sides s.lastProcessedRemoteMessageId.Store(sequencer.MessageId) s.params.Signalling.AckMessageId(sequencer.LastProcessedRemoteMessageId) s.params.Signalling.SetLastProcessedRemoteMessageId(sequencer.MessageId) - */ } case *livekit.Signalv2WireMessage_Fragment: diff --git a/pkg/service/rtcv2service.go b/pkg/service/rtcv2service.go index 99bad3774..95a703543 100644 --- a/pkg/service/rtcv2service.go +++ b/pkg/service/rtcv2service.go @@ -73,7 +73,7 @@ func NewRTCv2Service( func (s *RTCv2Service) SetupRoutes(mux *http.ServeMux) { mux.HandleFunc("POST "+cRTCv2Path, s.handlePost) - mux.HandleFunc("GET "+cRTCv2Path, s.validate) + mux.HandleFunc("GET "+cRTCv2ValidatePath, s.validate) mux.HandleFunc("PATCH "+cRTCv2ParticipantIDPath, s.handleParticipantPatch) mux.HandleFunc("DELETE "+cRTCv2ParticipantIDPath, s.handleParticipantDelete) } @@ -212,31 +212,9 @@ func (s *RTCv2Service) handleParticipantPatch(w http.ResponseWriter, r *http.Req return } - claims := GetGrants(r.Context()) - if claims == nil || claims.Video == nil { - HandleErrorJson(w, r, http.StatusUnauthorized, rtc.ErrPermissionDenied) - return - } - - roomName, err := EnsureJoinPermission(r.Context()) + roomName, participantIdentity, pID, code, err := getParams(r) if err != nil { - HandleErrorJson(w, r, http.StatusUnauthorized, err) - return - } - if roomName == "" { - HandleErrorJson(w, r, http.StatusUnauthorized, ErrNoRoomName) - return - } - - participantIdentity := livekit.ParticipantIdentity(claims.Identity) - if participantIdentity == "" { - HandleErrorJson(w, r, http.StatusUnauthorized, ErrIdentityEmpty) - return - } - - pID := livekit.ParticipantID(r.PathValue("participant_id")) - if pID == "" { - HandleErrorJson(w, r, http.StatusBadRequest, ErrParticipantSidEmpty) + HandleErrorJson(w, r, code, err) return } @@ -301,25 +279,9 @@ func (s *RTCv2Service) handleParticipantDelete(w http.ResponseWriter, r *http.Re return } - roomName, err := EnsureJoinPermission(r.Context()) + roomName, participantIdentity, pID, code, err := getParams(r) if err != nil { - HandleErrorJson(w, r, http.StatusUnauthorized, err) - return - } - if roomName == "" { - HandleErrorJson(w, r, http.StatusUnauthorized, ErrNoRoomName) - return - } - - participantIdentity := livekit.ParticipantIdentity(claims.Identity) - if participantIdentity == "" { - HandleErrorJson(w, r, http.StatusUnauthorized, ErrIdentityEmpty) - return - } - - pID := livekit.ParticipantID(r.PathValue("participant_id")) - if pID == "" { - HandleErrorJson(w, r, http.StatusBadRequest, ErrParticipantSidEmpty) + HandleErrorJson(w, r, code, err) return } @@ -363,3 +325,30 @@ func getWireMessage(r *http.Request) (*livekit.Signalv2WireMessage, error) { return wireMessage, nil } + +func getParams(r *http.Request) (livekit.RoomName, livekit.ParticipantIdentity, livekit.ParticipantID, int, error) { + claims := GetGrants(r.Context()) + if claims == nil || claims.Video == nil { + return "", "", "", http.StatusUnauthorized, rtc.ErrPermissionDenied + } + + roomName, err := EnsureJoinPermission(r.Context()) + if err != nil { + return "", "", "", http.StatusUnauthorized, err + } + if roomName == "" { + return "", "", "", http.StatusUnauthorized, ErrNoRoomName + } + + participantIdentity := livekit.ParticipantIdentity(claims.Identity) + if participantIdentity == "" { + return "", "", "", http.StatusUnauthorized, ErrIdentityEmpty + } + + pID := livekit.ParticipantID(r.PathValue("participant_id")) + if pID == "" { + return "", "", "", http.StatusBadRequest, ErrParticipantSidEmpty + } + + return roomName, participantIdentity, pID, http.StatusOK, nil +} diff --git a/pkg/service/utils.go b/pkg/service/utils.go index 2d7614145..f82cf6627 100644 --- a/pkg/service/utils.go +++ b/pkg/service/utils.go @@ -59,6 +59,7 @@ func HandleErrorJson(w http.ResponseWriter, r *http.Request, status int, err err }{ Error: err.Error(), }) + w.Header().Add("Content-type", "application/json") } func boolValue(s string) bool {