diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index e2821c574..65fe01c80 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -612,7 +612,7 @@ func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseRea p.updateState(livekit.ParticipantInfo_DISCONNECTED) // ensure this is synchronized - p.closeSignalConnection() + p.CloseSignalConnection() p.lock.RLock() onClose := p.onClose p.lock.RUnlock() @@ -1206,7 +1206,7 @@ func (p *ParticipantImpl) setupDisconnectTimer() { func (p *ParticipantImpl) onAnyTransportFailed() { // clients support resuming of connections when websocket becomes disconnected - p.closeSignalConnection() + p.CloseSignalConnection() // detect when participant has actually left. p.setupDisconnectTimer() @@ -1891,7 +1891,7 @@ func (p *ParticipantImpl) handleSubscriberNegotiationFailed() { }, }, }) - p.closeSignalConnection() + p.CloseSignalConnection() } func (p *ParticipantImpl) EnqueueSubscribeTrack(trackID livekit.TrackID, f func(sub types.LocalParticipant) error) { diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 263eab650..d46aad178 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -184,7 +184,7 @@ func TestTrackPublishing(t *testing.T) { func TestOutOfOrderUpdates(t *testing.T) { p := newParticipantForTest("test") p.SetMetadata("initial metadata") - sink := p.GetResponseSink().(*routingfakes.FakeMessageSink) + sink := p.getResponseSink().(*routingfakes.FakeMessageSink) pi1 := p.ToProto() p.SetMetadata("second update") pi2 := p.ToProto() diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 3aef563ef..afe30bedc 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -10,7 +10,7 @@ import ( "github.com/livekit/livekit-server/pkg/routing" ) -func (p *ParticipantImpl) GetResponseSink() routing.MessageSink { +func (p *ParticipantImpl) getResponseSink() routing.MessageSink { if !p.resSinkValid.Load() { return nil } @@ -174,7 +174,7 @@ func (p *ParticipantImpl) writeMessage(msg *livekit.SignalResponse) error { if p.State() == livekit.ParticipantInfo_DISCONNECTED { return nil } - sink := p.GetResponseSink() + sink := p.getResponseSink() if sink == nil { return nil } @@ -188,9 +188,10 @@ func (p *ParticipantImpl) writeMessage(msg *livekit.SignalResponse) error { } // closes signal connection to notify client to resume/reconnect -func (p *ParticipantImpl) closeSignalConnection() { - sink := p.GetResponseSink() +func (p *ParticipantImpl) CloseSignalConnection() { + sink := p.getResponseSink() if sink != nil { + p.params.Logger.Infow("closing signal connection") sink.Close() p.SetResponseSink(nil) } diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 38b8d8d5c..672db4436 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -333,9 +333,7 @@ func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions func (r *Room) ResumeParticipant(p types.LocalParticipant, responseSink routing.MessageSink) error { // close previous sink, and link to new one - if prevSink := p.GetResponseSink(); prevSink != nil { - prevSink.Close() - } + p.CloseSignalConnection() p.SetResponseSink(responseSink) updates := ToProtoParticipants(r.GetParticipants()) diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 96083ba0e..9ec0d78a4 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -214,8 +214,8 @@ type LocalParticipant interface { SubscriberAsPrimary() bool GetClientConfiguration() *livekit.ClientConfiguration - GetResponseSink() routing.MessageSink SetResponseSink(sink routing.MessageSink) + CloseSignalConnection() // permissions ClaimGrants() *auth.ClaimGrants diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 83ede51d7..02a7b71f7 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -150,6 +150,10 @@ type FakeLocalParticipant struct { closeReturnsOnCall map[int]struct { result1 error } + CloseSignalConnectionStub func() + closeSignalConnectionMutex sync.RWMutex + closeSignalConnectionArgsForCall []struct { + } ConnectedAtStub func() time.Time connectedAtMutex sync.RWMutex connectedAtArgsForCall []struct { @@ -269,16 +273,6 @@ type FakeLocalParticipant struct { getPublishedTracksReturnsOnCall map[int]struct { result1 []types.MediaTrack } - GetResponseSinkStub func() routing.MessageSink - getResponseSinkMutex sync.RWMutex - getResponseSinkArgsForCall []struct { - } - getResponseSinkReturns struct { - result1 routing.MessageSink - } - getResponseSinkReturnsOnCall map[int]struct { - result1 routing.MessageSink - } GetSubscribedParticipantsStub func() []livekit.ParticipantID getSubscribedParticipantsMutex sync.RWMutex getSubscribedParticipantsArgsForCall []struct { @@ -1461,6 +1455,30 @@ func (fake *FakeLocalParticipant) CloseReturnsOnCall(i int, result1 error) { }{result1} } +func (fake *FakeLocalParticipant) CloseSignalConnection() { + fake.closeSignalConnectionMutex.Lock() + fake.closeSignalConnectionArgsForCall = append(fake.closeSignalConnectionArgsForCall, struct { + }{}) + stub := fake.CloseSignalConnectionStub + fake.recordInvocation("CloseSignalConnection", []interface{}{}) + fake.closeSignalConnectionMutex.Unlock() + if stub != nil { + fake.CloseSignalConnectionStub() + } +} + +func (fake *FakeLocalParticipant) CloseSignalConnectionCallCount() int { + fake.closeSignalConnectionMutex.RLock() + defer fake.closeSignalConnectionMutex.RUnlock() + return len(fake.closeSignalConnectionArgsForCall) +} + +func (fake *FakeLocalParticipant) CloseSignalConnectionCalls(stub func()) { + fake.closeSignalConnectionMutex.Lock() + defer fake.closeSignalConnectionMutex.Unlock() + fake.CloseSignalConnectionStub = stub +} + func (fake *FakeLocalParticipant) ConnectedAt() time.Time { fake.connectedAtMutex.Lock() ret, specificReturn := fake.connectedAtReturnsOnCall[len(fake.connectedAtArgsForCall)] @@ -2080,59 +2098,6 @@ func (fake *FakeLocalParticipant) GetPublishedTracksReturnsOnCall(i int, result1 }{result1} } -func (fake *FakeLocalParticipant) GetResponseSink() routing.MessageSink { - fake.getResponseSinkMutex.Lock() - ret, specificReturn := fake.getResponseSinkReturnsOnCall[len(fake.getResponseSinkArgsForCall)] - fake.getResponseSinkArgsForCall = append(fake.getResponseSinkArgsForCall, struct { - }{}) - stub := fake.GetResponseSinkStub - fakeReturns := fake.getResponseSinkReturns - fake.recordInvocation("GetResponseSink", []interface{}{}) - fake.getResponseSinkMutex.Unlock() - if stub != nil { - return stub() - } - if specificReturn { - return ret.result1 - } - return fakeReturns.result1 -} - -func (fake *FakeLocalParticipant) GetResponseSinkCallCount() int { - fake.getResponseSinkMutex.RLock() - defer fake.getResponseSinkMutex.RUnlock() - return len(fake.getResponseSinkArgsForCall) -} - -func (fake *FakeLocalParticipant) GetResponseSinkCalls(stub func() routing.MessageSink) { - fake.getResponseSinkMutex.Lock() - defer fake.getResponseSinkMutex.Unlock() - fake.GetResponseSinkStub = stub -} - -func (fake *FakeLocalParticipant) GetResponseSinkReturns(result1 routing.MessageSink) { - fake.getResponseSinkMutex.Lock() - defer fake.getResponseSinkMutex.Unlock() - fake.GetResponseSinkStub = nil - fake.getResponseSinkReturns = struct { - result1 routing.MessageSink - }{result1} -} - -func (fake *FakeLocalParticipant) GetResponseSinkReturnsOnCall(i int, result1 routing.MessageSink) { - fake.getResponseSinkMutex.Lock() - defer fake.getResponseSinkMutex.Unlock() - fake.GetResponseSinkStub = nil - if fake.getResponseSinkReturnsOnCall == nil { - fake.getResponseSinkReturnsOnCall = make(map[int]struct { - result1 routing.MessageSink - }) - } - fake.getResponseSinkReturnsOnCall[i] = struct { - result1 routing.MessageSink - }{result1} -} - func (fake *FakeLocalParticipant) GetSubscribedParticipants() []livekit.ParticipantID { fake.getSubscribedParticipantsMutex.Lock() ret, specificReturn := fake.getSubscribedParticipantsReturnsOnCall[len(fake.getSubscribedParticipantsArgsForCall)] @@ -4837,6 +4802,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.RUnlock() fake.closeMutex.RLock() defer fake.closeMutex.RUnlock() + fake.closeSignalConnectionMutex.RLock() + defer fake.closeSignalConnectionMutex.RUnlock() fake.connectedAtMutex.RLock() defer fake.connectedAtMutex.RUnlock() fake.debugInfoMutex.RLock() @@ -4861,8 +4828,6 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.getPublishedTrackMutex.RUnlock() fake.getPublishedTracksMutex.RLock() defer fake.getPublishedTracksMutex.RUnlock() - fake.getResponseSinkMutex.RLock() - defer fake.getResponseSinkMutex.RUnlock() fake.getSubscribedParticipantsMutex.RLock() defer fake.getSubscribedParticipantsMutex.RUnlock() fake.getSubscribedTracksMutex.RLock()