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

Refine transport fallback for client resuming (#1597)

Browse Source 
* reset fallback after ice restart

* Configure ice for reconnect before send response
This commit is contained in:
cnderrauber
2023-04-10 15:12:05 +08:00
committed by GitHub
parent c56e37f3fe
commit c70a5c831f
6 changed files with 113 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/participant.go
+2 -2
View File
@@ -774,7 +774,7 @@ func (p *ParticipantImpl) MigrateState() types.MigrateState {
}
// ICERestart restarts subscriber ICE connections
func (p *ParticipantImpl) ICERestart(iceConfig *livekit.ICEConfig, reason livekit.ReconnectReason) {
func (p *ParticipantImpl) ICERestart(iceConfig *livekit.ICEConfig) {
p.clearDisconnectTimer()
p.clearMigrationTimer()
@@ -782,7 +782,7 @@ func (p *ParticipantImpl) ICERestart(iceConfig *livekit.ICEConfig, reason liveki
t.(types.LocalMediaTrack).Restart()
}
p.TransportManager.ICERestart(iceConfig, reason)
p.TransportManager.ICERestart(iceConfig)
}
func (p *ParticipantImpl) OnICEConfigChanged(f func(participant types.LocalParticipant, iceConfig *livekit.ICEConfig)) {
pkg/rtc/participant_signal.go
+3 -1
View File
@@ -172,7 +172,9 @@ func (p *ParticipantImpl) SendRefreshToken(token string) error {
})
}
func (p *ParticipantImpl) SendReconnectResponse(reconnectResponse *livekit.ReconnectResponse) error {
func (p *ParticipantImpl) HandleReconnectAndSendResponse(reconnectReason livekit.ReconnectReason, reconnectResponse *livekit.ReconnectResponse) error {
p.TransportManager.HandleClientReconnect(reconnectReason)
if !p.params.ClientInfo.CanHandleReconnectResponse() {
return nil
}
pkg/rtc/room.go
+3 -3
View File
@@ -407,7 +407,7 @@ func (r *Room) ResumeParticipant(p types.LocalParticipant, requestSource routing
p.SetSignalSourceValid(true)
if err := p.SendReconnectResponse(&livekit.ReconnectResponse{
if err := p.HandleReconnectAndSendResponse(reason, &livekit.ReconnectResponse{
IceServers: iceServers,
ClientConfiguration: p.GetClientConfiguration(),
}); err != nil {
@@ -423,7 +423,7 @@ func (r *Room) ResumeParticipant(p types.LocalParticipant, requestSource routing
p.SendRoomUpdate(r.protoRoom)
r.lock.RUnlock()
p.ICERestart(nil, reason)
p.ICERestart(nil)
return nil
}
@@ -729,7 +729,7 @@ func (r *Room) SimulateScenario(participant types.LocalParticipant, simulateScen
participant.ICERestart(&livekit.ICEConfig{
PreferenceSubscriber: livekit.ICECandidateType(scenario.SwitchCandidateProtocol),
PreferencePublisher: livekit.ICECandidateType(scenario.SwitchCandidateProtocol),
}, livekit.ReconnectReason_RR_SWITCH_CANDIDATE)
})
}
return nil
}
pkg/rtc/transportmanager.go
+19 -9
View File
@@ -457,11 +457,7 @@ func (t *TransportManager) NegotiateSubscriber(force bool) {
t.subscriber.Negotiate(force)
}
func (t *TransportManager) ICERestart(iceConfig *livekit.ICEConfig, reason livekit.ReconnectReason) {
if iceConfig != nil {
t.SetICEConfig(iceConfig)
}
func (t *TransportManager) HandleClientReconnect(reason livekit.ReconnectReason) {
var (
isShort bool
duration time.Duration
@@ -478,6 +474,9 @@ func (t *TransportManager) ICERestart(iceConfig *livekit.ICEConfig, reason livek
}
if isShort {
t.lock.Lock()
t.resetTransportConfigureLocked(false)
t.lock.Unlock()
t.params.Logger.Infow("short connection by client ice restart", "duration", duration, "reason", reason)
t.handleConnectionFailed(isShort)
}
@@ -486,6 +485,13 @@ func (t *TransportManager) ICERestart(iceConfig *livekit.ICEConfig, reason livek
t.publisher.ResetShortConnOnICERestart()
t.subscriber.ResetShortConnOnICERestart()
}
}
func (t *TransportManager) ICERestart(iceConfig *livekit.ICEConfig) {
if iceConfig != nil {
t.SetICEConfig(iceConfig)
}
t.subscriber.ICERestart()
}
@@ -499,14 +505,18 @@ func (t *TransportManager) SetICEConfig(iceConfig *livekit.ICEConfig) {
t.configureICE(iceConfig, true)
}
func (t *TransportManager) resetTransportConfigureLocked(reconfigured bool) {
t.failureCount = 0
t.isTransportReconfigured = reconfigured
t.udpLossUnstableCount = 0
t.lastFailure = time.Time{}
}
func (t *TransportManager) configureICE(iceConfig *livekit.ICEConfig, reset bool) {
t.lock.Lock()
isEqual := proto.Equal(t.iceConfig, iceConfig)
if reset || !isEqual {
t.failureCount = 0
t.isTransportReconfigured = !reset
t.udpLossUnstableCount = 0
t.lastFailure = time.Time{}
t.resetTransportConfigureLocked(!reset)
}
if isEqual {
pkg/rtc/types/interfaces.go
+2 -2
View File
@@ -263,7 +263,7 @@ type LocalParticipant interface {
HandleAnswer(sdp webrtc.SessionDescription)
Negotiate(force bool)
ICERestart(iceConfig *livekit.ICEConfig, reason livekit.ReconnectReason)
ICERestart(iceConfig *livekit.ICEConfig)
AddTrackToSubscriber(trackLocal webrtc.TrackLocal, params AddTrackParams) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error)
AddTransceiverFromTrackToSubscriber(trackLocal webrtc.TrackLocal, params AddTrackParams) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error)
RemoveTrackFromSubscriber(sender *webrtc.RTPSender) error
@@ -295,7 +295,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
HandleReconnectAndSendResponse(reconnectReason livekit.ReconnectReason, reconnectResponse *livekit.ReconnectResponse) error
IssueFullReconnect(reason ParticipantCloseReason)
// callbacks
pkg/rtc/types/typesfakes/fake_local_participant.go
+84 -84
View File
@@ -280,6 +280,18 @@ type FakeLocalParticipant struct {
handleOfferArgsForCall []struct {
arg1 webrtc.SessionDescription
}
HandleReconnectAndSendResponseStub func(livekit.ReconnectReason, *livekit.ReconnectResponse) error
handleReconnectAndSendResponseMutex sync.RWMutex
handleReconnectAndSendResponseArgsForCall []struct {
arg1 livekit.ReconnectReason
arg2 *livekit.ReconnectResponse
}
handleReconnectAndSendResponseReturns struct {
result1 error
}
handleReconnectAndSendResponseReturnsOnCall map[int]struct {
result1 error
}
HasPermissionStub func(livekit.TrackID, livekit.ParticipantIdentity) bool
hasPermissionMutex sync.RWMutex
hasPermissionArgsForCall []struct {
@@ -302,11 +314,10 @@ type FakeLocalParticipant struct {
hiddenReturnsOnCall map[int]struct {
result1 bool
}
ICERestartStub func(*livekit.ICEConfig, livekit.ReconnectReason)
ICERestartStub func(*livekit.ICEConfig)
iCERestartMutex sync.RWMutex
iCERestartArgsForCall []struct {
arg1 *livekit.ICEConfig
arg2 livekit.ReconnectReason
}
IDStub func() livekit.ParticipantID
iDMutex sync.RWMutex
@@ -565,17 +576,6 @@ 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 {
@@ -2195,6 +2195,68 @@ func (fake *FakeLocalParticipant) HandleOfferArgsForCall(i int) webrtc.SessionDe
return argsForCall.arg1
}
func (fake *FakeLocalParticipant) HandleReconnectAndSendResponse(arg1 livekit.ReconnectReason, arg2 *livekit.ReconnectResponse) error {
fake.handleReconnectAndSendResponseMutex.Lock()
ret, specificReturn := fake.handleReconnectAndSendResponseReturnsOnCall[len(fake.handleReconnectAndSendResponseArgsForCall)]
fake.handleReconnectAndSendResponseArgsForCall = append(fake.handleReconnectAndSendResponseArgsForCall, struct {
arg1 livekit.ReconnectReason
arg2 *livekit.ReconnectResponse
}{arg1, arg2})
stub := fake.HandleReconnectAndSendResponseStub
fakeReturns := fake.handleReconnectAndSendResponseReturns
fake.recordInvocation("HandleReconnectAndSendResponse", []interface{}{arg1, arg2})
fake.handleReconnectAndSendResponseMutex.Unlock()
if stub != nil {
return stub(arg1, arg2)
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseCallCount() int {
fake.handleReconnectAndSendResponseMutex.RLock()
defer fake.handleReconnectAndSendResponseMutex.RUnlock()
return len(fake.handleReconnectAndSendResponseArgsForCall)
}
func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseCalls(stub func(livekit.ReconnectReason, *livekit.ReconnectResponse) error) {
fake.handleReconnectAndSendResponseMutex.Lock()
defer fake.handleReconnectAndSendResponseMutex.Unlock()
fake.HandleReconnectAndSendResponseStub = stub
}
func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseArgsForCall(i int) (livekit.ReconnectReason, *livekit.ReconnectResponse) {
fake.handleReconnectAndSendResponseMutex.RLock()
defer fake.handleReconnectAndSendResponseMutex.RUnlock()
argsForCall := fake.handleReconnectAndSendResponseArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2
}
func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseReturns(result1 error) {
fake.handleReconnectAndSendResponseMutex.Lock()
defer fake.handleReconnectAndSendResponseMutex.Unlock()
fake.HandleReconnectAndSendResponseStub = nil
fake.handleReconnectAndSendResponseReturns = struct {
result1 error
}{result1}
}
func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseReturnsOnCall(i int, result1 error) {
fake.handleReconnectAndSendResponseMutex.Lock()
defer fake.handleReconnectAndSendResponseMutex.Unlock()
fake.HandleReconnectAndSendResponseStub = nil
if fake.handleReconnectAndSendResponseReturnsOnCall == nil {
fake.handleReconnectAndSendResponseReturnsOnCall = make(map[int]struct {
result1 error
})
}
fake.handleReconnectAndSendResponseReturnsOnCall[i] = struct {
result1 error
}{result1}
}
func (fake *FakeLocalParticipant) HasPermission(arg1 livekit.TrackID, arg2 livekit.ParticipantIdentity) bool {
fake.hasPermissionMutex.Lock()
ret, specificReturn := fake.hasPermissionReturnsOnCall[len(fake.hasPermissionArgsForCall)]
@@ -2310,17 +2372,16 @@ func (fake *FakeLocalParticipant) HiddenReturnsOnCall(i int, result1 bool) {
}{result1}
}
func (fake *FakeLocalParticipant) ICERestart(arg1 *livekit.ICEConfig, arg2 livekit.ReconnectReason) {
func (fake *FakeLocalParticipant) ICERestart(arg1 *livekit.ICEConfig) {
fake.iCERestartMutex.Lock()
fake.iCERestartArgsForCall = append(fake.iCERestartArgsForCall, struct {
arg1 *livekit.ICEConfig
arg2 livekit.ReconnectReason
}{arg1, arg2})
}{arg1})
stub := fake.ICERestartStub
fake.recordInvocation("ICERestart", []interface{}{arg1, arg2})
fake.recordInvocation("ICERestart", []interface{}{arg1})
fake.iCERestartMutex.Unlock()
if stub != nil {
fake.ICERestartStub(arg1, arg2)
fake.ICERestartStub(arg1)
}
}
@@ -2330,17 +2391,17 @@ func (fake *FakeLocalParticipant) ICERestartCallCount() int {
return len(fake.iCERestartArgsForCall)
}
func (fake *FakeLocalParticipant) ICERestartCalls(stub func(*livekit.ICEConfig, livekit.ReconnectReason)) {
func (fake *FakeLocalParticipant) ICERestartCalls(stub func(*livekit.ICEConfig)) {
fake.iCERestartMutex.Lock()
defer fake.iCERestartMutex.Unlock()
fake.ICERestartStub = stub
}
func (fake *FakeLocalParticipant) ICERestartArgsForCall(i int) (*livekit.ICEConfig, livekit.ReconnectReason) {
func (fake *FakeLocalParticipant) ICERestartArgsForCall(i int) *livekit.ICEConfig {
fake.iCERestartMutex.RLock()
defer fake.iCERestartMutex.RUnlock()
argsForCall := fake.iCERestartArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2
return argsForCall.arg1
}
func (fake *FakeLocalParticipant) ID() livekit.ParticipantID {
@@ -3795,67 +3856,6 @@ 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)]
@@ -5274,6 +5274,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
defer fake.handleAnswerMutex.RUnlock()
fake.handleOfferMutex.RLock()
defer fake.handleOfferMutex.RUnlock()
fake.handleReconnectAndSendResponseMutex.RLock()
defer fake.handleReconnectAndSendResponseMutex.RUnlock()
fake.hasPermissionMutex.RLock()
defer fake.hasPermissionMutex.RUnlock()
fake.hiddenMutex.RLock()
@@ -5344,8 +5346,6 @@ 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()