diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index aa3d4443d..7f92c2017 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -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)) { diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 544399a1e..106c8370a 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -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 } diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index ebdd590ef..b1109fab7 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -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 } diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index cc883a1ba..113fd99f0 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -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 { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index a83c6206f..0dcd9af88 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -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 diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 7ae08e75c..b63f167c4 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -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()