From 25debc6d35f5b82e65d76c13b653ee9308a73b05 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Wed, 11 Jan 2023 17:40:12 +0800 Subject: [PATCH] add reconnect response to update configuration while reconnecting (#1300) * add reconnect response to update configuration while reconnecting * fix test --- go.mod | 2 +- go.sum | 4 +- pkg/rtc/participant_signal.go | 8 ++ pkg/rtc/room.go | 9 ++- pkg/rtc/types/interfaces.go | 1 + .../typesfakes/fake_local_participant.go | 74 +++++++++++++++++++ pkg/service/roommanager.go | 27 +++++-- 7 files changed, 114 insertions(+), 11 deletions(-) diff --git a/go.mod b/go.mod index b86fd2484..bb68f54c6 100644 --- a/go.mod +++ b/go.mod @@ -18,7 +18,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20221221221243-f361fbe40290 github.com/livekit/mediatransportutil v0.0.0-20230111071722-904079e94a7c - github.com/livekit/protocol v1.3.2-0.20230110201647-34cae0997a36 + github.com/livekit/protocol v1.3.2-0.20230111091921-f637ca8e675f github.com/livekit/psrpc v0.2.1 github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995 github.com/mackerelio/go-osstat v0.2.3 diff --git a/go.sum b/go.sum index 5a1ddb922..ec27bf49a 100644 --- a/go.sum +++ b/go.sum @@ -233,8 +233,8 @@ github.com/livekit/mageutil v0.0.0-20221221221243-f361fbe40290 h1:ZVsQUuUOM9G7O3 github.com/livekit/mageutil v0.0.0-20221221221243-f361fbe40290/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20230111071722-904079e94a7c h1:wdzwTJjCpzy2FDmwdyVVGVa4+U9iv3E4Jy9qUDe/ubw= github.com/livekit/mediatransportutil v0.0.0-20230111071722-904079e94a7c/go.mod h1:1Dlx20JPoIKGP45eo+yuj0HjeE25zmyeX/EWHiPCjFw= -github.com/livekit/protocol v1.3.2-0.20230110201647-34cae0997a36 h1:SlWsB3XEt4fYYkcCeCES2V8CFkRYcX0ThdkNqWP4MPg= -github.com/livekit/protocol v1.3.2-0.20230110201647-34cae0997a36/go.mod h1:gwCG03nKlHlC9hTjL4pXQpn783ALhmbyhq65UZxqbb8= +github.com/livekit/protocol v1.3.2-0.20230111091921-f637ca8e675f h1:5trxeV2GknxRya2EgbE3BZeB+a8ULLDBRZffLxVq1x0= +github.com/livekit/protocol v1.3.2-0.20230111091921-f637ca8e675f/go.mod h1:gwCG03nKlHlC9hTjL4pXQpn783ALhmbyhq65UZxqbb8= github.com/livekit/psrpc v0.2.1 h1:ph/4egUMueUPoh5PZ/Aw4v6SH3wAbA+2t/GyCbpPKTg= github.com/livekit/psrpc v0.2.1/go.mod h1:MCe0xLdFPXmzogPiLrM94JIJbctb9+fAv5qYPkY2DXw= github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995 h1:vOaY2qvfLihDyeZtnGGN1Law9wRrw8BMGCr1TygTvMw= diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 136d38c11..5fa4d3fa7 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -174,6 +174,14 @@ func (p *ParticipantImpl) SendRefreshToken(token string) error { }) } +func (p *ParticipantImpl) SendReconnectResponse(reconnectResponse *livekit.ReconnectResponse) error { + return p.writeMessage(&livekit.SignalResponse{ + Message: &livekit.SignalResponse_Reconnect{ + Reconnect: reconnectResponse, + }, + }) +} + func (p *ParticipantImpl) sendICECandidate(c *webrtc.ICECandidate, target livekit.SignalTarget) error { trickle := ToProtoTrickle(c.ToJSON()) trickle.Target = target diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 92686fe24..c1b97837c 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -355,11 +355,18 @@ func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions return nil } -func (r *Room) ResumeParticipant(p types.LocalParticipant, responseSink routing.MessageSink) error { +func (r *Room) ResumeParticipant(p types.LocalParticipant, responseSink routing.MessageSink, iceServers []*livekit.ICEServer) error { // close previous sink, and link to new one p.CloseSignalConnection() p.SetResponseSink(responseSink) + if err := p.SendReconnectResponse(&livekit.ReconnectResponse{ + IceServers: iceServers, + ClientConfiguration: p.GetClientConfiguration(), + }); err != nil { + return err + } + updates := ToProtoParticipants(r.GetParticipants()) if err := p.SendParticipantUpdate(updates); err != nil { return err diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 7cd300e44..19097108b 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -286,6 +286,7 @@ type LocalParticipant interface { SendConnectionQualityUpdate(update *livekit.ConnectionQualityUpdate) error SubscriptionPermissionUpdate(publisherID livekit.ParticipantID, trackID livekit.TrackID, allowed bool) SendRefreshToken(token string) error + SendReconnectResponse(reconnectResponse *livekit.ReconnectResponse) error // callbacks OnStateChange(func(p LocalParticipant, oldState livekit.ParticipantInfo_State)) diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index b042940bd..0bd08be04 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -588,6 +588,17 @@ type FakeLocalParticipant struct { sendParticipantUpdateReturnsOnCall map[int]struct { result1 error } + SendReconnectResponseStub func(*livekit.ReconnectResponse) error + sendReconnectResponseMutex sync.RWMutex + sendReconnectResponseArgsForCall []struct { + arg1 *livekit.ReconnectResponse + } + sendReconnectResponseReturns struct { + result1 error + } + sendReconnectResponseReturnsOnCall map[int]struct { + result1 error + } SendRefreshTokenStub func(string) error sendRefreshTokenMutex sync.RWMutex sendRefreshTokenArgsForCall []struct { @@ -3862,6 +3873,67 @@ func (fake *FakeLocalParticipant) SendParticipantUpdateReturnsOnCall(i int, resu }{result1} } +func (fake *FakeLocalParticipant) SendReconnectResponse(arg1 *livekit.ReconnectResponse) error { + fake.sendReconnectResponseMutex.Lock() + ret, specificReturn := fake.sendReconnectResponseReturnsOnCall[len(fake.sendReconnectResponseArgsForCall)] + fake.sendReconnectResponseArgsForCall = append(fake.sendReconnectResponseArgsForCall, struct { + arg1 *livekit.ReconnectResponse + }{arg1}) + stub := fake.SendReconnectResponseStub + fakeReturns := fake.sendReconnectResponseReturns + fake.recordInvocation("SendReconnectResponse", []interface{}{arg1}) + fake.sendReconnectResponseMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) SendReconnectResponseCallCount() int { + fake.sendReconnectResponseMutex.RLock() + defer fake.sendReconnectResponseMutex.RUnlock() + return len(fake.sendReconnectResponseArgsForCall) +} + +func (fake *FakeLocalParticipant) SendReconnectResponseCalls(stub func(*livekit.ReconnectResponse) error) { + fake.sendReconnectResponseMutex.Lock() + defer fake.sendReconnectResponseMutex.Unlock() + fake.SendReconnectResponseStub = stub +} + +func (fake *FakeLocalParticipant) SendReconnectResponseArgsForCall(i int) *livekit.ReconnectResponse { + fake.sendReconnectResponseMutex.RLock() + defer fake.sendReconnectResponseMutex.RUnlock() + argsForCall := fake.sendReconnectResponseArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipant) SendReconnectResponseReturns(result1 error) { + fake.sendReconnectResponseMutex.Lock() + defer fake.sendReconnectResponseMutex.Unlock() + fake.SendReconnectResponseStub = nil + fake.sendReconnectResponseReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeLocalParticipant) SendReconnectResponseReturnsOnCall(i int, result1 error) { + fake.sendReconnectResponseMutex.Lock() + defer fake.sendReconnectResponseMutex.Unlock() + fake.SendReconnectResponseStub = nil + if fake.sendReconnectResponseReturnsOnCall == nil { + fake.sendReconnectResponseReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.sendReconnectResponseReturnsOnCall[i] = struct { + result1 error + }{result1} +} + func (fake *FakeLocalParticipant) SendRefreshToken(arg1 string) error { fake.sendRefreshTokenMutex.Lock() ret, specificReturn := fake.sendRefreshTokenReturnsOnCall[len(fake.sendRefreshTokenArgsForCall)] @@ -5169,6 +5241,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.sendJoinResponseMutex.RUnlock() fake.sendParticipantUpdateMutex.RLock() defer fake.sendParticipantUpdateMutex.RUnlock() + fake.sendReconnectResponseMutex.RLock() + defer fake.sendReconnectResponseMutex.RUnlock() fake.sendRefreshTokenMutex.RLock() defer fake.sendRefreshTokenMutex.RUnlock() fake.sendRoomUpdateMutex.RLock() diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index c2968c90d..1e28e1a65 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -219,6 +219,8 @@ func (r *RoomManager) StartSession( } defer room.Release() + protoRoom := room.ToProto() + // only create the room, but don't start a participant session if pi.Identity == "" { return nil @@ -233,7 +235,13 @@ func (r *RoomManager) StartSession( "nodeID", r.currentNode.Id, "participant", pi.Identity, ) - if err = room.ResumeParticipant(participant, responseSink); err != nil { + iceConfig := r.getIceConfig(participant) + if iceConfig == nil { + iceConfig = &livekit.ICEConfig{} + } + if err = room.ResumeParticipant(participant, responseSink, + r.iceServersForRoom(protoRoom, iceConfig.PreferenceSubscriber == livekit.ICECandidateType_ICT_TLS), + ); err != nil { logger.Warnw("could not resume participant", err, "participant", pi.Identity) return err } @@ -274,7 +282,6 @@ func (r *RoomManager) StartSession( rtcConf.SetBufferFactory(room.GetBufferFactory()) sid := livekit.ParticipantID(utils.NewGuid(utils.ParticipantPrefix)) pLogger := rtc.LoggerWithParticipant(room.Logger, pi.Identity, sid, false) - protoRoom := room.ToProto() // default allow forceTCP allowFallback := true if r.config.RTC.AllowTCPFallback != nil { @@ -695,16 +702,22 @@ func (r *RoomManager) refreshToken(participant types.LocalParticipant) error { } func (r *RoomManager) setIceConfig(participant types.LocalParticipant) *livekit.ICEConfig { + iceConfig := r.getIceConfig(participant) + if iceConfig == nil { + return &livekit.ICEConfig{} + } + participant.SetICEConfig(iceConfig) + return iceConfig +} + +func (r *RoomManager) getIceConfig(participant types.LocalParticipant) *livekit.ICEConfig { r.lock.Lock() + defer r.lock.Unlock() iceConfigCacheEntry, ok := r.iceConfigCache[participant.Identity()] if !ok || time.Since(iceConfigCacheEntry.modifiedAt) > iceConfigTTL { delete(r.iceConfigCache, participant.Identity()) - r.lock.Unlock() - return &livekit.ICEConfig{} + return nil } - r.lock.Unlock() - - participant.SetICEConfig(iceConfigCacheEntry.iceConfig) return iceConfigCacheEntry.iceConfig }