From d53f167b316e4a2b0c5ab00dbbd322d0c8a93870 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Mon, 29 Jan 2024 13:04:18 +0530 Subject: [PATCH] LeaveRequest changes. (#2426) Reworking this a bit 1. Send leave whenever the signal channel is closed to induce a resume. 2. Use a getter to get regions rather than setting. --- pkg/rtc/participant.go | 65 ++----------------- pkg/rtc/participant_signal.go | 31 +++++++++ pkg/rtc/types/interfaces.go | 4 +- .../typesfakes/fake_local_participant.go | 39 ----------- 4 files changed, 38 insertions(+), 101 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index ee4f20da8..65f558326 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -120,6 +120,7 @@ type ParticipantParams struct { AllowUDPUnstableFallback bool TURNSEnabled bool GetParticipantInfo func(pID livekit.ParticipantID) *livekit.ParticipantInfo + GetRegionSettings func(ip string) *livekit.RegionSettings DisableSupervisor bool ReconnectOnPublicationError bool ReconnectOnSubscriptionError bool @@ -221,8 +222,6 @@ type ParticipantImpl struct { // loggers for publisher and subscriber pubLogger logger.Logger subLogger logger.Logger - - regionSettings *livekit.RegionSettings } func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { @@ -766,35 +765,8 @@ func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseRea p.clearDisconnectTimer() p.clearMigrationTimer() - // send leave message - var leave *livekit.LeaveRequest - if p.ProtocolVersion().SupportsRegionsInLeaveRequest() { - leave = &livekit.LeaveRequest{ - Reason: reason.ToDisconnectReason(), - } - if isExpectedToResume { - leave.Action = livekit.LeaveRequest_RESUME - } else { - leave.Action = livekit.LeaveRequest_DISCONNECT - } - // although regions are not needed when resuming OR disconnecting, - // send it if available, just in case clients want to fall back. - p.lock.RLock() - if p.regionSettings != nil { - leave.Regions = proto.Clone(p.regionSettings).(*livekit.RegionSettings) - } - p.lock.RUnlock() - } else if sendLeave { - leave = &livekit.LeaveRequest{ - Reason: reason.ToDisconnectReason(), - } - } - if leave != nil { - _ = p.writeMessage(&livekit.SignalResponse{ - Message: &livekit.SignalResponse_Leave{ - Leave: leave, - }, - }) + if sendLeave { + p.sendLeaveRequest(reason, isExpectedToResume, false) } if p.supervisor != nil { @@ -870,6 +842,7 @@ func (p *ParticipantImpl) MaybeStartMigration(force bool, onStart func()) bool { onStart() } + p.sendLeaveRequest(types.ParticipantCloseReasonMigrationRequested, true, false) p.CloseSignalConnection(types.SignallingCloseReasonMigration) // @@ -1533,6 +1506,7 @@ func (p *ParticipantImpl) setupDisconnectTimer() { func (p *ParticipantImpl) onAnyTransportFailed() { // clients support resuming of connections when websocket becomes disconnected + p.sendLeaveRequest(types.ParticipantCloseReasonPeerConnectionDisconnected, true, false) p.CloseSignalConnection(types.SignallingCloseReasonTransportFailure) // detect when participant has actually left. @@ -2355,28 +2329,7 @@ func (p *ParticipantImpl) GetCachedDownTrack(trackID livekit.TrackID) (*webrtc.R } func (p *ParticipantImpl) IssueFullReconnect(reason types.ParticipantCloseReason) { - var leave *livekit.LeaveRequest - if p.ProtocolVersion().SupportsRegionsInLeaveRequest() { - leave = &livekit.LeaveRequest{ - Reason: reason.ToDisconnectReason(), - Action: livekit.LeaveRequest_RECONNECT, - } - p.lock.RLock() - if p.regionSettings != nil { - leave.Regions = proto.Clone(p.regionSettings).(*livekit.RegionSettings) - } - p.lock.RUnlock() - } else { - leave = &livekit.LeaveRequest{ - CanReconnect: true, - Reason: reason.ToDisconnectReason(), - } - } - _ = p.writeMessage(&livekit.SignalResponse{ - Message: &livekit.SignalResponse_Leave{ - Leave: leave, - }, - }) + p.sendLeaveRequest(reason, false, true) scr := types.SignallingCloseReasonUnknown switch reason { @@ -2538,9 +2491,3 @@ func (p *ParticipantImpl) setupEnabledCodecs(publishEnabledCodecs []*livekit.Cod } p.enabledSubscribeCodecs = subscribeCodecs } - -func (p *ParticipantImpl) SetRegionSettings(regionSettings *livekit.RegionSettings) { - p.lock.Lock() - p.regionSettings = proto.Clone(regionSettings).(*livekit.RegionSettings) - p.lock.Unlock() -} diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 0a1bb31f1..04529f59b 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -311,3 +311,34 @@ func (p *ParticipantImpl) CloseSignalConnection(reason types.SignallingCloseReas p.SetResponseSink(nil) } } + +func (p *ParticipantImpl) sendLeaveRequest(reason types.ParticipantCloseReason, isExpectedToResume bool, isExpectedToReconnect bool) { + var leave *livekit.LeaveRequest + if p.ProtocolVersion().SupportsRegionsInLeaveRequest() { + leave = &livekit.LeaveRequest{ + Reason: reason.ToDisconnectReason(), + } + switch { + case isExpectedToResume: + leave.Action = livekit.LeaveRequest_RESUME + case isExpectedToReconnect: + leave.Action = livekit.LeaveRequest_RECONNECT + default: + leave.Action = livekit.LeaveRequest_DISCONNECT + } + if leave.Action != livekit.LeaveRequest_DISCONNECT && p.params.GetRegionSettings != nil { + // sending region settings even for RESUME just in case client wants to a full reconnect despite server saying RESUME + leave.Regions = p.params.GetRegionSettings(p.params.ClientInfo.Address) + } + } else { + leave = &livekit.LeaveRequest{ + CanReconnect: isExpectedToReconnect, + Reason: reason.ToDisconnectReason(), + } + } + _ = p.writeMessage(&livekit.SignalResponse{ + Message: &livekit.SignalResponse_Leave{ + Leave: leave, + }, + }) +} diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 4cbcb117a..af18021d0 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -178,7 +178,7 @@ func (p ParticipantCloseReason) ToDisconnectReason() livekit.DisconnectReason { return livekit.DisconnectReason_STATE_MISMATCH case ParticipantCloseReasonDuplicateIdentity, ParticipantCloseReasonStale: return livekit.DisconnectReason_DUPLICATE_IDENTITY - case ParticipantCloseReasonMigrationComplete, ParticipantCloseReasonSimulateMigration: + case ParticipantCloseReasonMigrationRequested, ParticipantCloseReasonMigrationComplete, ParticipantCloseReasonSimulateMigration: return livekit.DisconnectReason_MIGRATION case ParticipantCloseReasonServiceRequestRemoveParticipant: return livekit.DisconnectReason_PARTICIPANT_REMOVED @@ -422,8 +422,6 @@ type LocalParticipant interface { GetPacer() pacer.Pacer GetTrafficLoad() *TrafficLoad - - SetRegionSettings(regionSettings *livekit.RegionSettings) } // Room is a container of participants, and can provide room-level actions diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 32c1c2869..f101e4398 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -753,11 +753,6 @@ type FakeLocalParticipant struct { setPermissionReturnsOnCall map[int]struct { result1 bool } - SetRegionSettingsStub func(*livekit.RegionSettings) - setRegionSettingsMutex sync.RWMutex - setRegionSettingsArgsForCall []struct { - arg1 *livekit.RegionSettings - } SetResponseSinkStub func(routing.MessageSink) setResponseSinkMutex sync.RWMutex setResponseSinkArgsForCall []struct { @@ -4982,38 +4977,6 @@ func (fake *FakeLocalParticipant) SetPermissionReturnsOnCall(i int, result1 bool }{result1} } -func (fake *FakeLocalParticipant) SetRegionSettings(arg1 *livekit.RegionSettings) { - fake.setRegionSettingsMutex.Lock() - fake.setRegionSettingsArgsForCall = append(fake.setRegionSettingsArgsForCall, struct { - arg1 *livekit.RegionSettings - }{arg1}) - stub := fake.SetRegionSettingsStub - fake.recordInvocation("SetRegionSettings", []interface{}{arg1}) - fake.setRegionSettingsMutex.Unlock() - if stub != nil { - fake.SetRegionSettingsStub(arg1) - } -} - -func (fake *FakeLocalParticipant) SetRegionSettingsCallCount() int { - fake.setRegionSettingsMutex.RLock() - defer fake.setRegionSettingsMutex.RUnlock() - return len(fake.setRegionSettingsArgsForCall) -} - -func (fake *FakeLocalParticipant) SetRegionSettingsCalls(stub func(*livekit.RegionSettings)) { - fake.setRegionSettingsMutex.Lock() - defer fake.setRegionSettingsMutex.Unlock() - fake.SetRegionSettingsStub = stub -} - -func (fake *FakeLocalParticipant) SetRegionSettingsArgsForCall(i int) *livekit.RegionSettings { - fake.setRegionSettingsMutex.RLock() - defer fake.setRegionSettingsMutex.RUnlock() - argsForCall := fake.setRegionSettingsArgsForCall[i] - return argsForCall.arg1 -} - func (fake *FakeLocalParticipant) SetResponseSink(arg1 routing.MessageSink) { fake.setResponseSinkMutex.Lock() fake.setResponseSinkArgsForCall = append(fake.setResponseSinkArgsForCall, struct { @@ -6352,8 +6315,6 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.setNameMutex.RUnlock() fake.setPermissionMutex.RLock() defer fake.setPermissionMutex.RUnlock() - fake.setRegionSettingsMutex.RLock() - defer fake.setRegionSettingsMutex.RUnlock() fake.setResponseSinkMutex.RLock() defer fake.setResponseSinkMutex.RUnlock() fake.setSignalSourceValidMutex.RLock()