dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-27 22:34:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

More v2 signalling changes (#3814)

Browse Source 
* More v2 signalling changes

- Use wire message in relay signal
- Generate offerId for SDP for sync mode also
- more checks for SDP offer/answer state to see if there are cases of
  missed/duplicate messages

* clean up
This commit is contained in:
Raja Subramanian
2025-07-24 10:14:01 +05:30
committed by GitHub
parent b20db94dc9
commit a75295fcd5
14 changed files with 277 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+4 -4
View File
@@ -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.20250721114233-52633eee694f
github.com/livekit/protocol v1.39.4-0.20250724043230-34c7a0eb632f
github.com/livekit/psrpc v0.6.1-0.20250511053145-465289d72c3c
github.com/mackerelio/go-osstat v0.2.5
github.com/magefile/mage v1.15.0
@@ -135,8 +135,8 @@ require (
golang.org/x/sys v0.34.0 // indirect
golang.org/x/text v0.27.0 // indirect
golang.org/x/tools v0.35.0 // indirect
google.golang.org/genproto/googleapis/api v0.0.0-20250715232539-7130f93afb79 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20250715232539-7130f93afb79 // indirect
google.golang.org/grpc v1.74.0 // indirect
google.golang.org/genproto/googleapis/api v0.0.0-20250721164621-a45f3dfb1074 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20250721164621-a45f3dfb1074 // indirect
google.golang.org/grpc v1.74.2 // indirect
gopkg.in/yaml.v2 v2.4.0 // indirect
)
go.sum
+8 -8
View File
@@ -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.20250721114233-52633eee694f h1:Cwe38+/ld3r5dnNmIZSALSoZPWNEMeYPZIi/qjpplLo=
github.com/livekit/protocol v1.39.4-0.20250721114233-52633eee694f/go.mod h1:YlgUxAegtU8jZ0tVXoIV/4fHeHqqLvS+6JnPKDbpFPU=
github.com/livekit/protocol v1.39.4-0.20250724043230-34c7a0eb632f h1:kdn/KF/2IRCQCXZWvm9RxKN4u79olHRZKo2JDxr/kPw=
github.com/livekit/protocol v1.39.4-0.20250724043230-34c7a0eb632f/go.mod h1:YlgUxAegtU8jZ0tVXoIV/4fHeHqqLvS+6JnPKDbpFPU=
github.com/livekit/psrpc v0.6.1-0.20250511053145-465289d72c3c h1:WwEr0YBejYbKzk8LSaO9h8h0G9MnE7shyDu8yXQWmEc=
github.com/livekit/psrpc v0.6.1-0.20250511053145-465289d72c3c/go.mod h1:kmD+AZPkWu0MaXIMv57jhNlbiSZZ/Jx4bzlxBDVmJes=
github.com/mackerelio/go-osstat v0.2.5 h1:+MqTbZUhoIt4m8qzkVoXUJg1EuifwlAJSk4Yl2GXh+o=
@@ -471,12 +471,12 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/genproto/googleapis/api v0.0.0-20250715232539-7130f93afb79 h1:iOye66xuaAK0WnkPuhQPUFy8eJcmwUXqGGP3om6IxX8=
google.golang.org/genproto/googleapis/api v0.0.0-20250715232539-7130f93afb79/go.mod h1:HKJDgKsFUnv5VAGeQjz8kxcgDP0HoE0iZNp0OdZNlhE=
google.golang.org/genproto/googleapis/rpc v0.0.0-20250715232539-7130f93afb79 h1:1ZwqphdOdWYXsUHgMpU/101nCtf/kSp9hOrcvFsnl10=
google.golang.org/genproto/googleapis/rpc v0.0.0-20250715232539-7130f93afb79/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A=
google.golang.org/grpc v1.74.0 h1:sxRSkyLxlceWQiqDofxDot3d4u7DyoHPc7SBXMj8gGY=
google.golang.org/grpc v1.74.0/go.mod h1:NZUaK8dAMUfzhK6uxZ+9511LtOrk73UGWOFoNvz7z+s=
google.golang.org/genproto/googleapis/api v0.0.0-20250721164621-a45f3dfb1074 h1:mVXdvnmR3S3BQOqHECm9NGMjYiRtEvDYcqAqedTXY6s=
google.golang.org/genproto/googleapis/api v0.0.0-20250721164621-a45f3dfb1074/go.mod h1:vYFwMYFbmA8vl6Z/krj/h7+U/AqpHknwJX4Uqgfyc7I=
google.golang.org/genproto/googleapis/rpc v0.0.0-20250721164621-a45f3dfb1074 h1:qJW29YvkiJmXOYMu5Tf8lyrTp3dOS+K4z6IixtLaCf8=
google.golang.org/genproto/googleapis/rpc v0.0.0-20250721164621-a45f3dfb1074/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A=
google.golang.org/grpc v1.74.2 h1:WoosgB65DlWVC9FqI82dGsZhWFNBSLjQ84bjROOpMu4=
google.golang.org/grpc v1.74.2/go.mod h1:CtQ+BGjaAIXHs/5YS3i473GqwBBa1zGQNevxdeBEXrM=
google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=
google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
pkg/rtc/participant.go
+24 -6
View File
@@ -363,6 +363,19 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
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.id.Store(params.SID)
p.dataChannelStats = telemetry.NewBytesTrackStats(
@@ -1168,19 +1181,24 @@ func (p *ParticipantImpl) onPublisherAnswer(answer webrtc.SessionDescription, an
return p.sendSdpAnswer(answer, answerId)
}
func (p *ParticipantImpl) GetAnswer() (webrtc.SessionDescription, error) {
func (p *ParticipantImpl) GetAnswer() (webrtc.SessionDescription, uint32, error) {
if p.IsClosed() || p.IsDisconnected() {
return webrtc.SessionDescription{}, ErrParticipantSessionClosed
return webrtc.SessionDescription{}, 0, ErrParticipantSessionClosed
}
answer, err := p.TransportManager.GetAnswer()
answer, answerId, err := p.TransportManager.GetAnswer()
if err != nil {
return answer, err
return answer, answerId, err
}
answer = p.configurePublisherAnswer(answer)
p.pubLogger.Debugw("returning answer", "transport", livekit.SignalTarget_PUBLISHER, "answer", answer)
return answer, nil
p.pubLogger.Debugw(
"returning answer",
"transport", livekit.SignalTarget_PUBLISHER,
"answer", answer,
"answerId", answerId,
)
return answer, answerId, nil
}
// HandleAnswer handles a client answer response, with subscriber PC, server initiates the
pkg/rtc/room.go
+8 -23
View File
@@ -621,7 +621,7 @@ func (r *Room) Joinv2(
participant types.LocalParticipant,
opts *ParticipantOptions,
iceServers []*livekit.ICEServer,
) (*livekit.ConnectResponse, error) {
) error {
connectResponse, err := func() (*livekit.ConnectResponse, error) {
r.lock.Lock()
defer r.lock.Unlock()
@@ -782,7 +782,7 @@ func (r *Room) Joinv2(
return connectResponse, nil
}()
if err != nil {
return connectResponse, err
return err
}
// SIGNALLING-V2-TODO
@@ -791,37 +791,22 @@ func (r *Room) Joinv2(
// 3. HandleOffer and get answer (publisher)
r.subscribeToExistingTracks(participant, true)
offer, err := participant.GetOffer()
offer, offerId, err := participant.GetOffer()
if err != nil {
participant.GetLogger().Warnw("could not get offer", err)
prometheus.ServiceOperationCounter.WithLabelValues("participant_join", "error", "get_subscriber_offer").Add(1)
return nil, err
return err
}
connectResponse.SubscriberSdp = protosignalling.ToProtoSessionDescription(offer, 0) // SIGNALLING-V2-TODO - need to proper offerId?
// SIGNALLING-V2-TODO: when moving to envelope response, do SendSdpOffer after sending connect response
connectResponse.SubscriberSdp = protosignalling.ToProtoSessionDescription(offer, offerId)
// for sync response, this does not actually send, only generates messageId and caches the message
if err := participant.SendConnectResponse(connectResponse); err != nil {
prometheus.ServiceOperationCounter.WithLabelValues("participant_join", "error", "send_response").Add(1)
return nil, err
}
if wireMessage := participant.SignalPendingMessages(); wireMessage != nil {
if wireMessage, ok := wireMessage.(*livekit.Signalv2WireMessage); ok {
switch msg := wireMessage.GetMessage().(type) {
case *livekit.Signalv2WireMessage_Envelope:
got_connect_response:
for _, innerMsg := range msg.Envelope.GetServerMessages() {
switch serverMessage := innerMsg.GetMessage().(type) {
case *livekit.Signalv2ServerMessage_ConnectResponse:
connectResponse = serverMessage.ConnectResponse
break got_connect_response
}
}
}
}
return err
}
prometheus.ServiceOperationCounter.WithLabelValues("participant_join", "success", "").Add(1)
return connectResponse, nil
return nil
}
func (r *Room) ReplaceParticipantRequestSource(identity livekit.ParticipantIdentity, reqSource routing.MessageSource) {
pkg/rtc/signalling/signalhandlerv2.go
+8
View File
@@ -105,6 +105,14 @@ func (s *signalhandlerv2) HandleRequest(msg proto.Message) error {
case *livekit.Signalv2ClientMessage_SubscriberSdp:
s.params.Participant.HandleAnswer(protosignalling.FromProtoSessionDescription(payload.SubscriberSdp))
case *livekit.Signalv2ClientMessage_Trickle:
candidateInit, err := protosignalling.FromProtoTrickle(payload.Trickle)
if err != nil {
s.params.Logger.Warnw("could not decode trickle", err)
return err
}
s.params.Participant.AddICECandidate(candidateInit, payload.Trickle.Target)
}
/* SIGNALLING-V2-TODO: uncomment once sequencer is implemented on both sides
pkg/rtc/transport.go
+88 -22
View File
@@ -235,6 +235,14 @@ type PCTransport struct {
preferTCP atomic.Bool
isClosed atomic.Bool
// used to check for offer/answer pairing,
// i. e. every offer should have an answer before another offer can be sent
localOfferId atomic.Uint32
remoteAnswerId atomic.Uint32
remoteOfferId atomic.Uint32
localAnswerId atomic.Uint32
eventsQueue *utils.TypedOpsQueue[event]
// the following should be accessed only in event processing go routine
@@ -248,8 +256,6 @@ type PCTransport struct {
signalStateCheckTimer *time.Timer
currentOfferIceCredential string // ice user:pwd, for publish side ice restart checking
pendingRestartIceOffer *webrtc.SessionDescription
activeOfferId uint32
activeAnswerId uint32
connectionDetails *types.ICEConnectionDetails
selectedPair atomic.Pointer[webrtc.ICECandidatePair]
@@ -486,8 +492,8 @@ func NewPCTransport(params TransportParams) (*PCTransport, error) {
canReuseTransceiver: true,
connectionDetails: types.NewICEConnectionDetails(params.Transport, params.Logger),
lastNegotiate: time.Now(),
activeOfferId: uint32(rand.Intn(1<<8) + 1),
}
t.localOfferId.Store(uint32(rand.Intn(1<<8) + 1))
bwe, err := t.createPeerConnection()
if err != nil {
@@ -1294,6 +1300,24 @@ func (t *PCTransport) clearConnTimer() {
func (t *PCTransport) HandleRemoteDescription(sd webrtc.SessionDescription, remoteId uint32) error {
if t.params.UseOneShotSignallingMode || t.params.SynchronousLocalCandidatesMode {
if sd.Type == webrtc.SDPTypeOffer {
remoteOfferId := t.remoteOfferId.Load()
if remoteOfferId != 0 && remoteOfferId != t.localAnswerId.Load() {
t.params.Logger.Warnw(
"sdp state: multiple offers without answer", nil,
"remoteOfferId", remoteOfferId,
"localAnswerId", t.localAnswerId.Load(),
"receivedRemoteOfferId", remoteId,
)
}
t.remoteOfferId.Store(remoteId)
} else {
if remoteId != 0 && remoteId != t.localOfferId.Load() {
t.params.Logger.Warnw("sdp state: answer id mismatch", nil, "expected", t.localOfferId.Load(), "got", remoteId)
}
t.remoteAnswerId.Store(remoteId)
}
// SIGNALLING-V2-TODO: need to support filtering candidates for transport fallback
// add remote candidates to ICE connection details
parsed, err := sd.Unmarshal()
@@ -1342,23 +1366,23 @@ func (t *PCTransport) HandleRemoteDescription(sd webrtc.SessionDescription, remo
return nil
}
func (t *PCTransport) GetAnswer() (webrtc.SessionDescription, error) {
func (t *PCTransport) GetAnswer() (webrtc.SessionDescription, uint32, error) {
if !t.params.UseOneShotSignallingMode && !t.params.SynchronousLocalCandidatesMode {
return webrtc.SessionDescription{}, ErrNotSynchronousLocalCandidatesMode
return webrtc.SessionDescription{}, 0, ErrNotSynchronousLocalCandidatesMode
}
prd := t.pc.PendingRemoteDescription()
if prd == nil || prd.Type != webrtc.SDPTypeOffer {
return webrtc.SessionDescription{}, ErrNoRemoteDescription
return webrtc.SessionDescription{}, 0, ErrNoRemoteDescription
}
answer, err := t.pc.CreateAnswer(nil)
if err != nil {
return webrtc.SessionDescription{}, err
return webrtc.SessionDescription{}, 0, err
}
if err = t.pc.SetLocalDescription(answer); err != nil {
return webrtc.SessionDescription{}, err
return webrtc.SessionDescription{}, 0, err
}
// wait for gathering to complete to include all candidates in the answer
@@ -1387,21 +1411,24 @@ func (t *PCTransport) GetAnswer() (webrtc.SessionDescription, error) {
}
}
return *cld, nil
answerId := t.remoteOfferId.Load()
t.localAnswerId.Store(answerId)
return *cld, answerId, nil
}
func (t *PCTransport) GetOffer() (webrtc.SessionDescription, error) {
func (t *PCTransport) GetOffer() (webrtc.SessionDescription, uint32, error) {
if !t.params.SynchronousLocalCandidatesMode {
return webrtc.SessionDescription{}, ErrNotSynchronousLocalCandidatesMode
return webrtc.SessionDescription{}, 0, ErrNotSynchronousLocalCandidatesMode
}
offer, err := t.pc.CreateOffer(nil)
if err != nil {
return webrtc.SessionDescription{}, err
return webrtc.SessionDescription{}, 0, err
}
if err = t.pc.SetLocalDescription(offer); err != nil {
return webrtc.SessionDescription{}, err
return webrtc.SessionDescription{}, 0, err
}
// wait for gathering to complete to include all candidates in the answer
@@ -1430,7 +1457,7 @@ func (t *PCTransport) GetOffer() (webrtc.SessionDescription, error) {
}
}
return *pld, nil
return *pld, t.localOfferId.Inc(), nil
}
func (t *PCTransport) GetICESessionUfrag() (string, error) {
@@ -2260,8 +2287,16 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error {
t.setupSignalStateCheckTimer()
t.activeOfferId++
if err := t.params.Handler.OnOffer(offer, t.activeOfferId); err != nil {
remoteAnswerId := t.remoteAnswerId.Load()
if remoteAnswerId != 0 && remoteAnswerId != t.localOfferId.Load() {
t.params.Logger.Warnw(
"sdp state: sending offer before receiving answer", nil,
"localOfferId", t.localOfferId.Load(),
"remoteAnswerId", remoteAnswerId,
)
}
if err := t.params.Handler.OnOffer(offer, t.localOfferId.Inc()); err != nil {
prometheus.ServiceOperationCounter.WithLabelValues("offer", "error", "write_message").Add(1)
return errors.Wrap(err, "could not send offer")
}
@@ -2378,10 +2413,21 @@ func (t *PCTransport) createAndSendAnswer() error {
t.params.Logger.Debugw("local answer (filtered)", "sdp", answer.SDP)
}
if err := t.params.Handler.OnAnswer(answer, t.activeAnswerId); err != nil {
localAnswerId := t.localAnswerId.Load()
if localAnswerId != 0 && localAnswerId >= t.remoteOfferId.Load() {
t.params.Logger.Warnw(
"sdp state: duplicate answer", nil,
"localAnswerId", localAnswerId,
"remoteOfferId", t.remoteOfferId.Load(),
)
}
answerId := t.remoteOfferId.Load()
if err := t.params.Handler.OnAnswer(answer, answerId); err != nil {
prometheus.ServiceOperationCounter.WithLabelValues("answer", "error", "write_message").Add(1)
return errors.Wrap(err, "could not send answer")
}
t.localAnswerId.Store(answerId)
prometheus.ServiceOperationCounter.WithLabelValues("answer", "success", "").Add(1)
return t.localDescriptionSent()
@@ -2389,7 +2435,16 @@ func (t *PCTransport) createAndSendAnswer() error {
func (t *PCTransport) handleRemoteOfferReceived(sd *webrtc.SessionDescription, offerId uint32) error {
t.params.Logger.Debugw("processing offer", "offerId", offerId)
t.activeAnswerId = offerId
remoteOfferId := t.remoteOfferId.Load()
if remoteOfferId != 0 && remoteOfferId != t.localAnswerId.Load() {
t.params.Logger.Warnw(
"sdp state: multiple offers without answer", nil,
"remoteOfferId", remoteOfferId,
"localAnswerId", t.localAnswerId.Load(),
"receivedRemoteOfferId", offerId,
)
}
t.remoteOfferId.Store(offerId)
parsed, err := sd.Unmarshal()
if err != nil {
@@ -2453,9 +2508,11 @@ func (t *PCTransport) handleRemoteOfferReceived(sd *webrtc.SessionDescription, o
func (t *PCTransport) handleRemoteAnswerReceived(sd *webrtc.SessionDescription, answerId uint32) error {
t.params.Logger.Debugw("processing answer", "answerId", answerId)
if answerId != 0 && answerId != t.activeOfferId {
t.params.Logger.Warnw("answer id mismatch", nil, "expected", t.activeOfferId, "got", answerId)
if answerId != 0 && answerId != t.localOfferId.Load() {
t.params.Logger.Warnw("sdp state: answer id mismatch", nil, "expected", t.localOfferId.Load(), "got", answerId)
}
t.remoteAnswerId.Store(answerId)
t.clearSignalStateCheckTimer()
if err := t.setRemoteDescription(*sd); err != nil {
@@ -2520,8 +2577,17 @@ func (t *PCTransport) doICERestart() error {
t.params.Logger.Infow("deferring ice restart to next offer")
t.setNegotiationState(transport.NegotiationStateRetry)
t.restartAtNextOffer = true
t.activeOfferId++
err := t.params.Handler.OnOffer(*offer, t.activeOfferId)
remoteAnswerId := t.remoteAnswerId.Load()
if remoteAnswerId != 0 && remoteAnswerId != t.localOfferId.Load() {
t.params.Logger.Warnw(
"sdp state: answer not received in ICE restart", nil,
"localOfferId", t.localOfferId.Load(),
"remoteAnswerId", remoteAnswerId,
)
}
err := t.params.Handler.OnOffer(*offer, t.localOfferId.Inc())
if err != nil {
prometheus.ServiceOperationCounter.WithLabelValues("offer", "error", "write_message").Add(1)
} else {
pkg/rtc/transportmanager.go
+4 -4
View File
@@ -458,12 +458,12 @@ func (t *TransportManager) HandleOffer(offer webrtc.SessionDescription, offerId
return t.publisher.HandleRemoteDescription(offer, offerId)
}
func (t *TransportManager) GetAnswer() (webrtc.SessionDescription, error) {
answer, err := t.publisher.GetAnswer()
func (t *TransportManager) GetAnswer() (webrtc.SessionDescription, uint32, error) {
answer, answerId, err := t.publisher.GetAnswer()
if err == nil {
t.lastPublisherAnswer.Store(answer)
}
return answer, err
return answer, answerId, err
}
func (t *TransportManager) GetPublisherICESessionUfrag() (string, error) {
@@ -496,7 +496,7 @@ func (t *TransportManager) HandleAnswer(answer webrtc.SessionDescription, answer
t.subscriber.HandleRemoteDescription(answer, answerId)
}
func (t *TransportManager) GetOffer() (webrtc.SessionDescription, error) {
func (t *TransportManager) GetOffer() (webrtc.SessionDescription, uint32, error) {
return t.subscriber.GetOffer()
}
pkg/rtc/types/interfaces.go
+2 -2
View File
@@ -403,14 +403,14 @@ type LocalParticipant interface {
// PeerConnection
AddICECandidate(candidate webrtc.ICECandidateInit, target livekit.SignalTarget)
HandleOffer(sdp webrtc.SessionDescription, offerId uint32) error
GetAnswer() (webrtc.SessionDescription, error)
GetAnswer() (webrtc.SessionDescription, uint32, error)
HandleICETrickleSDPFragment(sdpFragment string) error
HandleICERestartSDPFragment(sdpFragment string) (string, error)
AddTrack(req *livekit.AddTrackRequest)
SetTrackMuted(trackID livekit.TrackID, muted bool, fromAdmin bool) *livekit.TrackInfo
HandleAnswer(sdp webrtc.SessionDescription, answerId uint32)
GetOffer() (webrtc.SessionDescription, error)
GetOffer() (webrtc.SessionDescription, uint32, error)
Negotiate(force bool)
ICERestart(iceConfig *livekit.ICEConfig)
AddTrackLocal(trackLocal webrtc.TrackLocal, params AddTrackParams) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error)
pkg/rtc/types/typesfakes/fake_local_participant.go
+38 -28
View File
@@ -213,17 +213,19 @@ type FakeLocalParticipant struct {
getAdaptiveStreamReturnsOnCall map[int]struct {
result1 bool
}
GetAnswerStub func() (webrtc.SessionDescription, error)
GetAnswerStub func() (webrtc.SessionDescription, uint32, error)
getAnswerMutex sync.RWMutex
getAnswerArgsForCall []struct {
}
getAnswerReturns struct {
result1 webrtc.SessionDescription
result2 error
result2 uint32
result3 error
}
getAnswerReturnsOnCall map[int]struct {
result1 webrtc.SessionDescription
result2 error
result2 uint32
result3 error
}
GetAudioLevelStub func() (float64, bool)
getAudioLevelMutex sync.RWMutex
@@ -361,17 +363,19 @@ type FakeLocalParticipant struct {
getLoggerResolverReturnsOnCall map[int]struct {
result1 logger.DeferredFieldResolver
}
GetOfferStub func() (webrtc.SessionDescription, error)
GetOfferStub func() (webrtc.SessionDescription, uint32, error)
getOfferMutex sync.RWMutex
getOfferArgsForCall []struct {
}
getOfferReturns struct {
result1 webrtc.SessionDescription
result2 error
result2 uint32
result3 error
}
getOfferReturnsOnCall map[int]struct {
result1 webrtc.SessionDescription
result2 error
result2 uint32
result3 error
}
GetPacerStub func() pacer.Pacer
getPacerMutex sync.RWMutex
@@ -2437,7 +2441,7 @@ func (fake *FakeLocalParticipant) GetAdaptiveStreamReturnsOnCall(i int, result1
}{result1}
}
func (fake *FakeLocalParticipant) GetAnswer() (webrtc.SessionDescription, error) {
func (fake *FakeLocalParticipant) GetAnswer() (webrtc.SessionDescription, uint32, error) {
fake.getAnswerMutex.Lock()
ret, specificReturn := fake.getAnswerReturnsOnCall[len(fake.getAnswerArgsForCall)]
fake.getAnswerArgsForCall = append(fake.getAnswerArgsForCall, struct {
@@ -2450,9 +2454,9 @@ func (fake *FakeLocalParticipant) GetAnswer() (webrtc.SessionDescription, error)
return stub()
}
if specificReturn {
return ret.result1, ret.result2
return ret.result1, ret.result2, ret.result3
}
return fakeReturns.result1, fakeReturns.result2
return fakeReturns.result1, fakeReturns.result2, fakeReturns.result3
}
func (fake *FakeLocalParticipant) GetAnswerCallCount() int {
@@ -2461,36 +2465,39 @@ func (fake *FakeLocalParticipant) GetAnswerCallCount() int {
return len(fake.getAnswerArgsForCall)
}
func (fake *FakeLocalParticipant) GetAnswerCalls(stub func() (webrtc.SessionDescription, error)) {
func (fake *FakeLocalParticipant) GetAnswerCalls(stub func() (webrtc.SessionDescription, uint32, error)) {
fake.getAnswerMutex.Lock()
defer fake.getAnswerMutex.Unlock()
fake.GetAnswerStub = stub
}
func (fake *FakeLocalParticipant) GetAnswerReturns(result1 webrtc.SessionDescription, result2 error) {
func (fake *FakeLocalParticipant) GetAnswerReturns(result1 webrtc.SessionDescription, result2 uint32, result3 error) {
fake.getAnswerMutex.Lock()
defer fake.getAnswerMutex.Unlock()
fake.GetAnswerStub = nil
fake.getAnswerReturns = struct {
result1 webrtc.SessionDescription
result2 error
}{result1, result2}
result2 uint32
result3 error
}{result1, result2, result3}
}
func (fake *FakeLocalParticipant) GetAnswerReturnsOnCall(i int, result1 webrtc.SessionDescription, result2 error) {
func (fake *FakeLocalParticipant) GetAnswerReturnsOnCall(i int, result1 webrtc.SessionDescription, result2 uint32, result3 error) {
fake.getAnswerMutex.Lock()
defer fake.getAnswerMutex.Unlock()
fake.GetAnswerStub = nil
if fake.getAnswerReturnsOnCall == nil {
fake.getAnswerReturnsOnCall = make(map[int]struct {
result1 webrtc.SessionDescription
result2 error
result2 uint32
result3 error
})
}
fake.getAnswerReturnsOnCall[i] = struct {
result1 webrtc.SessionDescription
result2 error
}{result1, result2}
result2 uint32
result3 error
}{result1, result2, result3}
}
func (fake *FakeLocalParticipant) GetAudioLevel() (float64, bool) {
@@ -3204,7 +3211,7 @@ func (fake *FakeLocalParticipant) GetLoggerResolverReturnsOnCall(i int, result1
}{result1}
}
func (fake *FakeLocalParticipant) GetOffer() (webrtc.SessionDescription, error) {
func (fake *FakeLocalParticipant) GetOffer() (webrtc.SessionDescription, uint32, error) {
fake.getOfferMutex.Lock()
ret, specificReturn := fake.getOfferReturnsOnCall[len(fake.getOfferArgsForCall)]
fake.getOfferArgsForCall = append(fake.getOfferArgsForCall, struct {
@@ -3217,9 +3224,9 @@ func (fake *FakeLocalParticipant) GetOffer() (webrtc.SessionDescription, error)
return stub()
}
if specificReturn {
return ret.result1, ret.result2
return ret.result1, ret.result2, ret.result3
}
return fakeReturns.result1, fakeReturns.result2
return fakeReturns.result1, fakeReturns.result2, fakeReturns.result3
}
func (fake *FakeLocalParticipant) GetOfferCallCount() int {
@@ -3228,36 +3235,39 @@ func (fake *FakeLocalParticipant) GetOfferCallCount() int {
return len(fake.getOfferArgsForCall)
}
func (fake *FakeLocalParticipant) GetOfferCalls(stub func() (webrtc.SessionDescription, error)) {
func (fake *FakeLocalParticipant) GetOfferCalls(stub func() (webrtc.SessionDescription, uint32, error)) {
fake.getOfferMutex.Lock()
defer fake.getOfferMutex.Unlock()
fake.GetOfferStub = stub
}
func (fake *FakeLocalParticipant) GetOfferReturns(result1 webrtc.SessionDescription, result2 error) {
func (fake *FakeLocalParticipant) GetOfferReturns(result1 webrtc.SessionDescription, result2 uint32, result3 error) {
fake.getOfferMutex.Lock()
defer fake.getOfferMutex.Unlock()
fake.GetOfferStub = nil
fake.getOfferReturns = struct {
result1 webrtc.SessionDescription
result2 error
}{result1, result2}
result2 uint32
result3 error
}{result1, result2, result3}
}
func (fake *FakeLocalParticipant) GetOfferReturnsOnCall(i int, result1 webrtc.SessionDescription, result2 error) {
func (fake *FakeLocalParticipant) GetOfferReturnsOnCall(i int, result1 webrtc.SessionDescription, result2 uint32, result3 error) {
fake.getOfferMutex.Lock()
defer fake.getOfferMutex.Unlock()
fake.GetOfferStub = nil
if fake.getOfferReturnsOnCall == nil {
fake.getOfferReturnsOnCall = make(map[int]struct {
result1 webrtc.SessionDescription
result2 error
result2 uint32
result3 error
})
}
fake.getOfferReturnsOnCall[i] = struct {
result1 webrtc.SessionDescription
result2 error
}{result1, result2}
result2 uint32
result3 error
}{result1, result2, result3}
}
func (fake *FakeLocalParticipant) GetPacer() pacer.Pacer {
pkg/service/errors.go
+2
View File
@@ -46,4 +46,6 @@ var (
ErrSIPDispatchRuleNotFound = psrpc.NewErrorf(psrpc.NotFound, "requested sip dispatch rule does not exist")
ErrSIPParticipantNotFound = psrpc.NewErrorf(psrpc.NotFound, "requested sip participant does not exist")
ErrInvalidMessageType = psrpc.NewErrorf(psrpc.Internal, "invalid message type")
ErrNoConnectRequest = psrpc.NewErrorf(psrpc.InvalidArgument, "no connect request")
ErrNoConnectResponse = psrpc.NewErrorf(psrpc.InvalidArgument, "no connect response")
)
pkg/service/roommanager.go
+35 -13
View File
@@ -34,6 +34,7 @@ import (
"github.com/livekit/protocol/observability"
"github.com/livekit/protocol/observability/roomobs"
"github.com/livekit/protocol/rpc"
"github.com/livekit/protocol/signalling"
"github.com/livekit/protocol/utils"
"github.com/livekit/protocol/utils/guid"
"github.com/livekit/protocol/utils/must"
@@ -569,11 +570,17 @@ func (r *RoomManager) StartSession(
func (r *RoomManager) HandleConnect(
ctx context.Context,
grants *auth.ClaimGrants,
rscr *rpc.RelaySignalv2ConnectRequest,
createRoom *livekit.CreateRoomRequest,
wireMessage *livekit.Signalv2WireMessage, // contains livekit.ConnectRequest
) (*rpc.RelaySignalv2ConnectResponse, error) {
sessionStartTime := time.Now()
createRoom := rscr.CreateRoom
// find connect request in wire message
connectRequest := signalling.GetConnectRequest(wireMessage)
if connectRequest == nil {
return nil, ErrNoConnectRequest
}
room, err := r.getOrCreateRoom(ctx, createRoom)
if err != nil {
return nil, err
@@ -603,15 +610,15 @@ func (r *RoomManager) HandleConnect(
"nodeID", r.currentNode.NodeID(),
"numParticipants", room.GetParticipantCount(),
"grants", grants,
"connectRequest", logger.Proto(rscr),
"wireMessage", logger.Proto(wireMessage),
)
clientInfo := rscr.ConnectRequest.ClientInfo
clientInfo := connectRequest.ClientInfo
clientConf := r.clientConfManager.GetConfiguration(clientInfo)
rtcConf := *r.rtcConfig
rtcConf.SetBufferFactory(room.GetBufferFactory())
if rscr.ConnectRequest.ConnectionSettings.DisableIceLite {
if connectRequest.ConnectionSettings.DisableIceLite {
rtcConf.SettingEngine.SetLite(false)
}
@@ -640,8 +647,8 @@ func (r *RoomManager) HandleConnect(
}
subscriberAllowPause := r.config.RTC.CongestionControl.AllowPause
if rscr.ConnectRequest.ConnectionSettings.SubscriberAllowPause != nil {
subscriberAllowPause = *rscr.ConnectRequest.ConnectionSettings.SubscriberAllowPause
if connectRequest.ConnectionSettings.SubscriberAllowPause != nil {
subscriberAllowPause = *connectRequest.ConnectionSettings.SubscriberAllowPause
}
participant, err := rtc.NewParticipant(rtc.ParticipantParams{
@@ -666,7 +673,7 @@ func (r *RoomManager) HandleConnect(
ClientConf: clientConf,
ClientInfo: rtc.ClientInfo{ClientInfo: clientInfo},
// SIGNALLING-V@-TODO Region: pi.Region,
AdaptiveStream: rscr.ConnectRequest.ConnectionSettings.AdaptiveStream,
AdaptiveStream: connectRequest.ConnectionSettings.AdaptiveStream,
AllowTCPFallback: allowFallback,
TURNSEnabled: r.config.IsTURNSEnabled(),
ParticipantHelper: &roomManagerParticipantHelper{
@@ -696,20 +703,36 @@ func (r *RoomManager) HandleConnect(
// join room
opts := rtc.ParticipantOptions{
AutoSubscribe: rscr.ConnectRequest.ConnectionSettings.AutoSubscribe,
AutoSubscribe: connectRequest.ConnectionSettings.AutoSubscribe,
}
iceServers := r.iceServersForParticipant(
apiKey,
participant,
iceConfig.PreferenceSubscriber == livekit.ICECandidateType_ICT_TLS,
)
connectResponse, err := room.Joinv2(participant, &opts, iceServers)
if err != nil {
if err := room.Joinv2(participant, &opts, iceServers); err != nil {
pLogger.Errorw("could not join room", err)
_ = participant.Close(true, types.ParticipantCloseReasonJoinFailed, false)
return nil, err
}
// SIGNALLING-V2-TODO: process messages other than ConnecRequest in envelope.
var wireMessageResponse *livekit.Signalv2WireMessage
if pending := participant.SignalPendingMessages(); pending != nil {
var ok bool
if wireMessageResponse, ok = pending.(*livekit.Signalv2WireMessage); !ok {
pLogger.Errorw("could not join room", ErrInvalidMessageType)
_ = participant.Close(true, types.ParticipantCloseReasonJoinFailed, false)
return nil, ErrInvalidMessageType
}
}
if signalling.GetConnectResponse(wireMessageResponse) == nil {
pLogger.Errorw("could not join room", ErrNoConnectResponse)
_ = participant.Close(true, types.ParticipantCloseReasonJoinFailed, false)
return nil, ErrNoConnectResponse
}
var participantServerClosers utils.Closers
participantTopic := rpc.FormatParticipantTopic(room.Name(), participant.Identity())
participantServer := must.Get(rpc.NewTypedParticipantServer(r, r.bus))
@@ -721,7 +744,6 @@ func (r *RoomManager) HandleConnect(
return nil, err
}
// SIGNALLING-V2-TODO: register HTTP
httpSignalParticipantServer := must.Get(rpc.NewTypedSignalv2ParticipantServer(signalv2ParticipantService{r}, r.bus))
participantServerClosers = append(
participantServerClosers,
@@ -776,7 +798,7 @@ func (r *RoomManager) HandleConnect(
})
return &rpc.RelaySignalv2ConnectResponse{
ConnectResponse: connectResponse,
WireMessage: wireMessageResponse,
}, nil
}
pkg/service/roommanager_service.go
+1 -1
View File
@@ -78,7 +78,7 @@ func (s rtcRestService) Create(ctx context.Context, req *rpc.RTCRestCreateReques
return nil, err
}
answer, err := lp.GetAnswer()
answer, _, err := lp.GetAnswer()
if err != nil {
lp.GetLogger().Errorw("rtcRest service: could not get answer", err)
return nil, err
pkg/service/rtcv2service.go
+54 -96
View File
@@ -28,6 +28,7 @@ import (
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/rpc"
"github.com/livekit/protocol/signalling"
"github.com/livekit/psrpc"
"google.golang.org/protobuf/proto"
)
@@ -79,8 +80,13 @@ func (s *RTCv2Service) SetupRoutes(mux *http.ServeMux) {
func (s *RTCv2Service) validateInternal(
lgr logger.Logger,
r *http.Request,
connectRequest *livekit.ConnectRequest,
wireMessage *livekit.Signalv2WireMessage,
) (livekit.RoomName, livekit.ParticipantIdentity, *rpc.RelaySignalv2ConnectRequest, int, error) {
connectRequest := signalling.GetConnectRequest(wireMessage)
if connectRequest == nil {
return "", "", nil, http.StatusBadRequest, ErrNoConnectRequest
}
params := ValidateConnectRequestParams{
metadata: connectRequest.Metadata,
attributes: connectRequest.ParticipantAttributes,
@@ -108,9 +114,9 @@ func (s *RTCv2Service) validateInternal(
return res.roomName,
livekit.ParticipantIdentity(res.grants.Identity),
&rpc.RelaySignalv2ConnectRequest{
GrantsJson: string(grantsJson),
CreateRoom: res.createRoomRequest,
ConnectRequest: connectRequest,
GrantsJson: string(grantsJson),
CreateRoom: res.createRoomRequest,
WireMessage: wireMessage,
},
code,
err
@@ -128,80 +134,52 @@ func (s *RTCv2Service) handlePost(w http.ResponseWriter, r *http.Request) {
return
}
switch msg := wireMessage.GetMessage().(type) {
case *livekit.Signalv2WireMessage_Envelope:
for _, innerMsg := range msg.Envelope.GetClientMessages() {
switch clientMessage := innerMsg.GetMessage().(type) {
case *livekit.Signalv2ClientMessage_ConnectRequest:
roomName, participantIdentity, rscr, code, err := s.validateInternal(
utils.GetLogger(r.Context()),
r,
clientMessage.ConnectRequest,
)
if err != nil {
HandleErrorJson(w, r, code, err)
return
}
if err := s.roomAllocator.SelectRoomNode(r.Context(), roomName, ""); err != nil {
HandleErrorJson(w, r, http.StatusInternalServerError, err)
return
}
resp, err := s.router.HandleParticipantConnectRequest(r.Context(), roomName, participantIdentity, rscr)
if err != nil {
HandleErrorJson(w, r, http.StatusInternalServerError, err)
return
}
// SIGNALLING-V2-TODO: this needs to be in signal cache and get messageId
wireMessage := &livekit.Signalv2WireMessage{
Message: &livekit.Signalv2WireMessage_Envelope{
Envelope: &livekit.Envelope{
ServerMessages: []*livekit.Signalv2ServerMessage{
&livekit.Signalv2ServerMessage{
Message: &livekit.Signalv2ServerMessage_ConnectResponse{
ConnectResponse: resp.ConnectResponse,
},
},
},
},
},
}
marshalled, err := proto.Marshal(wireMessage)
if err != nil {
HandleErrorJson(w, r, http.StatusInternalServerError, err)
return
}
w.Header().Add("Content-type", "application/x-protobuf")
w.Write(marshalled)
logger.Debugw(
"connect response",
"room", roomName,
"roomID", resp.ConnectResponse.Room.Sid, // SIGNALLING-V2-TODO: roomID may not be resolved
"participant", participantIdentity,
"pID", resp.ConnectResponse.Participant.Sid,
"connectResponse", logger.Proto(resp.ConnectResponse),
)
default:
HandleErrorJson(
w,
r,
http.StatusBadRequest,
fmt.Errorf("%w, message: %T", errUnknownMessageType, clientMessage),
)
}
}
case *livekit.Signalv2WireMessage_Fragment:
utils.GetLogger(r.Context()).Errorw("signalv2 bad request", errFragmentsInHTTP)
HandleErrorJson(w, r, http.StatusBadRequest, errFragmentsInHTTP)
// only connect requests should be coming in here and there should not be fragments
roomName, participantIdentity, rscr, code, err := s.validateInternal(
utils.GetLogger(r.Context()),
r,
wireMessage,
)
if err != nil {
HandleErrorJson(w, r, code, err)
return
}
if err := s.roomAllocator.SelectRoomNode(r.Context(), roomName, ""); err != nil {
HandleErrorJson(w, r, http.StatusInternalServerError, err)
return
}
resp, err := s.router.HandleParticipantConnectRequest(r.Context(), roomName, participantIdentity, rscr)
if err != nil {
HandleErrorJson(w, r, http.StatusInternalServerError, err)
return
}
connectResponse := signalling.GetConnectResponse(resp.WireMessage)
if connectResponse == nil {
HandleErrorJson(w, r, http.StatusInternalServerError, ErrNoConnectResponse)
return
}
marshalled, err := proto.Marshal(resp.WireMessage)
if err != nil {
HandleErrorJson(w, r, http.StatusInternalServerError, err)
return
}
w.Header().Add("Content-type", "application/x-protobuf")
w.Write(marshalled)
logger.Debugw(
"connect response",
"room", roomName,
"roomID", connectResponse.Room.Sid, // SIGNALLING-V2-TODO: roomID may not be resolved
"participant", participantIdentity,
"pID", connectResponse.Participant.Sid,
"wireMessage", logger.Proto(resp.WireMessage),
)
w.WriteHeader(http.StatusOK)
}
@@ -217,13 +195,7 @@ func (s *RTCv2Service) validate(w http.ResponseWriter, r *http.Request) {
return
}
connectRequest := getConnectRequest(wireMessage)
if connectRequest == nil {
HandleErrorJson(w, r, http.StatusBadRequest, errors.New("no connect request"))
return
}
_, _, _, code, err := s.validateInternal(utils.GetLogger(r.Context()), r, connectRequest)
_, _, _, code, err := s.validateInternal(utils.GetLogger(r.Context()), r, wireMessage)
if err != nil {
HandleErrorJson(w, r, code, err)
return
@@ -337,17 +309,3 @@ func getWireMessage(r *http.Request) (*livekit.Signalv2WireMessage, error) {
return wireMessage, nil
}
func getConnectRequest(wireMessage *livekit.Signalv2WireMessage) *livekit.ConnectRequest {
switch msg := wireMessage.GetMessage().(type) {
case *livekit.Signalv2WireMessage_Envelope:
for _, innerMsg := range msg.Envelope.GetClientMessages() {
switch clientMessage := innerMsg.GetMessage().(type) {
case *livekit.Signalv2ClientMessage_ConnectRequest:
return clientMessage.ConnectRequest
}
}
}
return nil
}
pkg/service/signalv2.go
+1 -1
View File
@@ -123,7 +123,7 @@ func (s *defaultSignalv2Handler) HandleConnect(
return nil, err
}
return s.roomManager.HandleConnect(ctx, grants, rscr)
return s.roomManager.HandleConnect(ctx, grants, rscr.CreateRoom, rscr.WireMessage)
}
// ------------------------------------------