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

LeaveRequest changes. (#2426)

Browse Source 
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.
This commit is contained in:
Raja Subramanian
2024-01-29 13:04:18 +05:30
committed by GitHub
parent 2a3de84351
commit d53f167b31
4 changed files with 38 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/participant.go
+6 -59
View File
@@ -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()
}
pkg/rtc/participant_signal.go
+31
View File
@@ -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,
},
})
}
pkg/rtc/types/interfaces.go
+1 -3
View File
@@ -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
pkg/rtc/types/typesfakes/fake_local_participant.go
-39
View File
@@ -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()