diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index da41e7ff0..7492beaf7 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -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()) diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index e32a739be..9202ebb91 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -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 { diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 3ae984632..f2891e79a 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -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 } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 87b89ffe1..a8e3331df 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -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)) diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index b6f349e70..a417a5b31 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -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) {