dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-07-28 12:09:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

add reconnect response to update configuration while reconnecting (#1300)

Browse Source 
* add reconnect response to update configuration while reconnecting

* fix test
This commit is contained in:
cnderrauber
2023-01-11 17:40:12 +08:00
committed by GitHub
parent c022319845
commit 25debc6d35
7 changed files with 114 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
pkg/rtc/participant_signal.go
+8
View File
@@ -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
pkg/rtc/room.go
+8 -1
View File
@@ -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
pkg/rtc/types/interfaces.go
+1
View File
@@ -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))
pkg/rtc/types/typesfakes/fake_local_participant.go
+74
View File
@@ -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()
pkg/service/roommanager.go
+20 -7
View File
@@ -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
}