dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-10-06 12:08:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add a local participant method to get recently disconnected (#4911)

Browse Source 
participants.

While migrating, the old node will have the recently disconnected
participants in the migrating out participant's cache and needs to sent
out to the participant from the migrating in node via the new WebSocket.
This method can be used to exchange that data between the two nodes.

Keeping it as `recently` disconnected and leaving the semantics upto how
recent for implementation.
This commit is contained in:
Raja Subramanian
2026-09-26 01:33:38 +05:30
committed by GitHub
parent 9d8ee5a319
commit 93f5b3eb61
5 changed files with 107 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/participant_internal_test.go
+5
View File
@@ -917,6 +917,7 @@ func TestMigratingInParticipantWaitsForReconnectResponse(t *testing.T) {
require.NoError(t, p.HandleReconnectAndSendResponse(
livekit.ReconnectReason_RR_UNKNOWN,
&livekit.ReconnectResponse{LastMessageSeq: 21},
nil,
))
require.True(t, p.IsReady())
@@ -942,6 +943,7 @@ func TestMigratingInParticipantWaitsForReconnectResponse(t *testing.T) {
require.NoError(t, p.HandleReconnectAndSendResponse(
livekit.ReconnectReason_RR_UNKNOWN,
&livekit.ReconnectResponse{},
nil,
))
require.Equal(t, 2, sink.WriteMessageCallCount())
@@ -964,6 +966,7 @@ func TestMigratingInParticipantWaitsForReconnectResponse(t *testing.T) {
require.NoError(t, p.HandleReconnectAndSendResponse(
livekit.ReconnectReason_RR_UNKNOWN,
&livekit.ReconnectResponse{},
nil,
))
require.True(t, p.IsReady())
require.Zero(t, sink.WriteMessageCallCount())
@@ -1004,6 +1007,7 @@ func TestResumedParticipantWaitsForReconnectResponse(t *testing.T) {
require.NoError(t, p.HandleReconnectAndSendResponse(
livekit.ReconnectReason_RR_SIGNAL_DISCONNECTED,
&livekit.ReconnectResponse{LastMessageSeq: 7},
nil,
))
require.Equal(t, 2, sink.WriteMessageCallCount())
@@ -1040,6 +1044,7 @@ func TestResumedParticipantWaitsForReconnectResponse(t *testing.T) {
require.NoError(t, p.HandleReconnectAndSendResponse(
livekit.ReconnectReason_RR_SIGNAL_DISCONNECTED,
&livekit.ReconnectResponse{},
nil,
))
require.Zero(t, sink.WriteMessageCallCount())
pkg/rtc/participant_signal.go
+21 -4
View File
@@ -175,7 +175,11 @@ func (p *ParticipantImpl) SendRoomMovedResponse(roomMovedResponse *livekit.RoomM
return p.signaller.WriteMessage(p.signalling.SignalRoomMovedResponse(roomMovedResponse))
}
func (p *ParticipantImpl) HandleReconnectAndSendResponse(reconnectReason livekit.ReconnectReason, reconnectResponse *livekit.ReconnectResponse) error {
func (p *ParticipantImpl) HandleReconnectAndSendResponse(
reconnectReason livekit.ReconnectReason,
reconnectResponse *livekit.ReconnectResponse,
recentlyDisconnectedParticipants []*livekit.ParticipantInfo, // when migrating in, these are the participants disconnected received from migrate out node
) error {
p.TransportManager.HandleClientReconnect(reconnectReason)
if !p.params.ClientInfo.CanHandleReconnectResponse() {
@@ -198,6 +202,11 @@ func (p *ParticipantImpl) HandleReconnectAndSendResponse(reconnectReason livekit
}
if p.params.ProtocolVersion.SupportsDisconnectedUpdate() {
if len(recentlyDisconnectedParticipants) != 0 {
if err := p.signaller.WriteMessage(p.signalling.SignalParticipantUpdate(recentlyDisconnectedParticipants)); err != nil {
return err
}
}
return p.sendDisconnectUpdatesForReconnect()
}
@@ -243,11 +252,19 @@ func (p *ParticipantImpl) flushQueuedUpdates() {
}
func (p *ParticipantImpl) sendDisconnectUpdatesForReconnect() error {
recentlyDisconnectedParticipants := p.GetRecentlyDisconnectedParticipants()
if len(recentlyDisconnectedParticipants) != 0 {
return p.signaller.WriteMessage(p.signalling.SignalParticipantUpdate(recentlyDisconnectedParticipants))
}
return nil
}
func (p *ParticipantImpl) GetRecentlyDisconnectedParticipants() []*livekit.ParticipantInfo {
// look back a little more than last signal receive time as WebSocket close
// on client side might have flushed messages and client application may have
// lost connectivity earlier.
lastSignalAt := p.TransportManager.LastSeenSignalAt().Add(-3 * time.Second)
var disconnectedParticipants []*livekit.ParticipantInfo
var recentlyDisconnectedParticipants []*livekit.ParticipantInfo
p.updateLock.Lock()
keys := p.updateCache.Keys()
for i := len(keys) - 1; i >= 0; i-- {
@@ -257,7 +274,7 @@ func (p *ParticipantImpl) sendDisconnectUpdatesForReconnect() error {
}
if info.state == livekit.ParticipantInfo_DISCONNECTED {
disconnectedParticipants = append(disconnectedParticipants, &livekit.ParticipantInfo{
recentlyDisconnectedParticipants = append(recentlyDisconnectedParticipants, &livekit.ParticipantInfo{
Sid: string(keys[i]),
Identity: string(info.identity),
Version: info.version,
@@ -268,7 +285,7 @@ func (p *ParticipantImpl) sendDisconnectUpdatesForReconnect() error {
}
p.updateLock.Unlock()
return p.signaller.WriteMessage(p.signalling.SignalParticipantUpdate(disconnectedParticipants))
return recentlyDisconnectedParticipants
}
func (p *ParticipantImpl) sendICECandidate(ic *webrtc.ICECandidate, target livekit.SignalTarget) error {
pkg/rtc/room.go
+1 -1
View File
@@ -586,7 +586,7 @@ func (r *Room) ResumeParticipant(
ClientConfiguration: p.GetClientConfiguration(),
ServerInfo: r.serverInfo,
LastMessageSeq: p.GetLastReliableSequence(false),
}); err != nil {
}, nil); err != nil {
return err
}
pkg/rtc/types/interfaces.go
+2 -1
View File
@@ -583,10 +583,11 @@ type LocalParticipant interface {
SendConnectionQualityUpdate(update *livekit.ConnectionQualityUpdate) error
SendSubscriptionPermissionUpdate(publisherID livekit.ParticipantID, trackID livekit.TrackID, allowed bool) error
SendRefreshToken(token string) error
HandleReconnectAndSendResponse(reconnectReason livekit.ReconnectReason, reconnectResponse *livekit.ReconnectResponse) error
HandleReconnectAndSendResponse(reconnectReason livekit.ReconnectReason, reconnectResponse *livekit.ReconnectResponse, recentlyDisconnectedParticipants []*livekit.ParticipantInfo) error
IssueFullReconnect(reason ParticipantCloseReason)
SendRoomMovedResponse(moved *livekit.RoomMovedResponse) error
SendDataTrackSubscriberHandles(handles map[uint32]*livekit.DataTrackSubscriberHandles_PublishedDataTrack) error
GetRecentlyDisconnectedParticipants() []*livekit.ParticipantInfo
AddOnClose(key string, callback func(LocalParticipant))
OnClaimsChanged(callback func(LocalParticipant))
pkg/rtc/types/typesfakes/fake_local_participant.go
+78 -8
View File
@@ -527,6 +527,16 @@ type FakeLocalParticipant struct {
result1 string
result2 error
}
GetRecentlyDisconnectedParticipantsStub func() []*livekit.ParticipantInfo
getRecentlyDisconnectedParticipantsMutex sync.RWMutex
getRecentlyDisconnectedParticipantsArgsForCall []struct {
}
getRecentlyDisconnectedParticipantsReturns struct {
result1 []*livekit.ParticipantInfo
}
getRecentlyDisconnectedParticipantsReturnsOnCall map[int]struct {
result1 []*livekit.ParticipantInfo
}
GetReporterStub func() roomobs.ParticipantSessionReporter
getReporterMutex sync.RWMutex
getReporterArgsForCall []struct {
@@ -692,11 +702,12 @@ type FakeLocalParticipant struct {
arg1 *sfu.DownTrack
arg2 *rtcp.ReceiverReport
}
HandleReconnectAndSendResponseStub func(livekit.ReconnectReason, *livekit.ReconnectResponse) error
HandleReconnectAndSendResponseStub func(livekit.ReconnectReason, *livekit.ReconnectResponse, []*livekit.ParticipantInfo) error
handleReconnectAndSendResponseMutex sync.RWMutex
handleReconnectAndSendResponseArgsForCall []struct {
arg1 livekit.ReconnectReason
arg2 *livekit.ReconnectResponse
arg3 []*livekit.ParticipantInfo
}
handleReconnectAndSendResponseReturns struct {
result1 error
@@ -4264,6 +4275,59 @@ func (fake *FakeLocalParticipant) GetPublisherICESessionUfragReturnsOnCall(i int
}{result1, result2}
}
func (fake *FakeLocalParticipant) GetRecentlyDisconnectedParticipants() []*livekit.ParticipantInfo {
fake.getRecentlyDisconnectedParticipantsMutex.Lock()
ret, specificReturn := fake.getRecentlyDisconnectedParticipantsReturnsOnCall[len(fake.getRecentlyDisconnectedParticipantsArgsForCall)]
fake.getRecentlyDisconnectedParticipantsArgsForCall = append(fake.getRecentlyDisconnectedParticipantsArgsForCall, struct {
}{})
stub := fake.GetRecentlyDisconnectedParticipantsStub
fakeReturns := fake.getRecentlyDisconnectedParticipantsReturns
fake.recordInvocation("GetRecentlyDisconnectedParticipants", []interface{}{})
fake.getRecentlyDisconnectedParticipantsMutex.Unlock()
if stub != nil {
return stub()
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakeLocalParticipant) GetRecentlyDisconnectedParticipantsCallCount() int {
fake.getRecentlyDisconnectedParticipantsMutex.RLock()
defer fake.getRecentlyDisconnectedParticipantsMutex.RUnlock()
return len(fake.getRecentlyDisconnectedParticipantsArgsForCall)
}
func (fake *FakeLocalParticipant) GetRecentlyDisconnectedParticipantsCalls(stub func() []*livekit.ParticipantInfo) {
fake.getRecentlyDisconnectedParticipantsMutex.Lock()
defer fake.getRecentlyDisconnectedParticipantsMutex.Unlock()
fake.GetRecentlyDisconnectedParticipantsStub = stub
}
func (fake *FakeLocalParticipant) GetRecentlyDisconnectedParticipantsReturns(result1 []*livekit.ParticipantInfo) {
fake.getRecentlyDisconnectedParticipantsMutex.Lock()
defer fake.getRecentlyDisconnectedParticipantsMutex.Unlock()
fake.GetRecentlyDisconnectedParticipantsStub = nil
fake.getRecentlyDisconnectedParticipantsReturns = struct {
result1 []*livekit.ParticipantInfo
}{result1}
}
func (fake *FakeLocalParticipant) GetRecentlyDisconnectedParticipantsReturnsOnCall(i int, result1 []*livekit.ParticipantInfo) {
fake.getRecentlyDisconnectedParticipantsMutex.Lock()
defer fake.getRecentlyDisconnectedParticipantsMutex.Unlock()
fake.GetRecentlyDisconnectedParticipantsStub = nil
if fake.getRecentlyDisconnectedParticipantsReturnsOnCall == nil {
fake.getRecentlyDisconnectedParticipantsReturnsOnCall = make(map[int]struct {
result1 []*livekit.ParticipantInfo
})
}
fake.getRecentlyDisconnectedParticipantsReturnsOnCall[i] = struct {
result1 []*livekit.ParticipantInfo
}{result1}
}
func (fake *FakeLocalParticipant) GetReporter() roomobs.ParticipantSessionReporter {
fake.getReporterMutex.Lock()
ret, specificReturn := fake.getReporterReturnsOnCall[len(fake.getReporterArgsForCall)]
@@ -5168,19 +5232,25 @@ func (fake *FakeLocalParticipant) HandleReceiverReportArgsForCall(i int) (*sfu.D
return argsForCall.arg1, argsForCall.arg2
}
func (fake *FakeLocalParticipant) HandleReconnectAndSendResponse(arg1 livekit.ReconnectReason, arg2 *livekit.ReconnectResponse) error {
func (fake *FakeLocalParticipant) HandleReconnectAndSendResponse(arg1 livekit.ReconnectReason, arg2 *livekit.ReconnectResponse, arg3 []*livekit.ParticipantInfo) error {
var arg3Copy []*livekit.ParticipantInfo
if arg3 != nil {
arg3Copy = make([]*livekit.ParticipantInfo, len(arg3))
copy(arg3Copy, arg3)
}
fake.handleReconnectAndSendResponseMutex.Lock()
ret, specificReturn := fake.handleReconnectAndSendResponseReturnsOnCall[len(fake.handleReconnectAndSendResponseArgsForCall)]
fake.handleReconnectAndSendResponseArgsForCall = append(fake.handleReconnectAndSendResponseArgsForCall, struct {
arg1 livekit.ReconnectReason
arg2 *livekit.ReconnectResponse
}{arg1, arg2})
arg3 []*livekit.ParticipantInfo
}{arg1, arg2, arg3Copy})
stub := fake.HandleReconnectAndSendResponseStub
fakeReturns := fake.handleReconnectAndSendResponseReturns
fake.recordInvocation("HandleReconnectAndSendResponse", []interface{}{arg1, arg2})
fake.recordInvocation("HandleReconnectAndSendResponse", []interface{}{arg1, arg2, arg3Copy})
fake.handleReconnectAndSendResponseMutex.Unlock()
if stub != nil {
return stub(arg1, arg2)
return stub(arg1, arg2, arg3)
}
if specificReturn {
return ret.result1
@@ -5194,17 +5264,17 @@ func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseCallCount() int
return len(fake.handleReconnectAndSendResponseArgsForCall)
}
func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseCalls(stub func(livekit.ReconnectReason, *livekit.ReconnectResponse) error) {
func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseCalls(stub func(livekit.ReconnectReason, *livekit.ReconnectResponse, []*livekit.ParticipantInfo) error) {
fake.handleReconnectAndSendResponseMutex.Lock()
defer fake.handleReconnectAndSendResponseMutex.Unlock()
fake.HandleReconnectAndSendResponseStub = stub
}
func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseArgsForCall(i int) (livekit.ReconnectReason, *livekit.ReconnectResponse) {
func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseArgsForCall(i int) (livekit.ReconnectReason, *livekit.ReconnectResponse, []*livekit.ParticipantInfo) {
fake.handleReconnectAndSendResponseMutex.RLock()
defer fake.handleReconnectAndSendResponseMutex.RUnlock()
argsForCall := fake.handleReconnectAndSendResponseArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
}
func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseReturns(result1 error) {