From 7d5c991d8d82c01ec2321062e2e17316c1c3ff7a Mon Sep 17 00:00:00 2001 From: Paul Wells Date: Sun, 14 Apr 2024 12:49:13 -0700 Subject: [PATCH] add disconnected chan to participant (#2650) --- pkg/rtc/participant.go | 11 +++- pkg/rtc/types/interfaces.go | 1 + .../typesfakes/fake_local_participant.go | 65 +++++++++++++++++++ pkg/service/roommanager.go | 9 +-- 4 files changed, 77 insertions(+), 9 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index e49e57f3a..fe55b7bc1 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -147,7 +147,8 @@ type ParticipantImpl struct { isClosed atomic.Bool closeReason atomic.Value // types.ParticipantCloseReason - state atomic.Value // livekit.ParticipantInfo_State + state atomic.Value // livekit.ParticipantInfo_State + disconnected chan struct{} resSinkMu sync.Mutex resSink routing.MessageSink @@ -241,7 +242,8 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { return nil, ErrMissingGrants } p := &ParticipantImpl{ - params: params, + params: params, + disconnected: make(chan struct{}), pubRTCPQueue: sutils.NewOpsQueue(sutils.OpsQueueParams{ Name: "pub-rtcp", MinSize: 64, @@ -372,6 +374,10 @@ func (p *ParticipantImpl) IsDisconnected() bool { return p.State() == livekit.ParticipantInfo_DISCONNECTED } +func (p *ParticipantImpl) Disconnected() <-chan struct{} { + return p.disconnected +} + func (p *ParticipantImpl) IsIdle() bool { // check if there are any published tracks that are subscribed for _, t := range p.GetPublishedTracks() { @@ -841,6 +847,7 @@ func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseRea p.UpTrackManager.Close(isExpectedToResume) p.updateState(livekit.ParticipantInfo_DISCONNECTED) + close(p.disconnected) // ensure this is synchronized p.CloseSignalConnection(types.SignallingCloseReasonParticipantClose) diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 57d6f3922..e072d1935 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -308,6 +308,7 @@ type LocalParticipant interface { IsClosed() bool IsReady() bool IsDisconnected() bool + Disconnected() <-chan struct{} IsIdle() bool SubscriberAsPrimary() bool GetClientInfo() *livekit.ClientInfo diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 351697226..b0843f428 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -168,6 +168,16 @@ type FakeLocalParticipant struct { debugInfoReturnsOnCall map[int]struct { result1 map[string]interface{} } + DisconnectedStub func() <-chan struct{} + disconnectedMutex sync.RWMutex + disconnectedArgsForCall []struct { + } + disconnectedReturns struct { + result1 <-chan struct{} + } + disconnectedReturnsOnCall map[int]struct { + result1 <-chan struct{} + } GetAdaptiveStreamStub func() bool getAdaptiveStreamMutex sync.RWMutex getAdaptiveStreamArgsForCall []struct { @@ -1770,6 +1780,59 @@ func (fake *FakeLocalParticipant) DebugInfoReturnsOnCall(i int, result1 map[stri }{result1} } +func (fake *FakeLocalParticipant) Disconnected() <-chan struct{} { + fake.disconnectedMutex.Lock() + ret, specificReturn := fake.disconnectedReturnsOnCall[len(fake.disconnectedArgsForCall)] + fake.disconnectedArgsForCall = append(fake.disconnectedArgsForCall, struct { + }{}) + stub := fake.DisconnectedStub + fakeReturns := fake.disconnectedReturns + fake.recordInvocation("Disconnected", []interface{}{}) + fake.disconnectedMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) DisconnectedCallCount() int { + fake.disconnectedMutex.RLock() + defer fake.disconnectedMutex.RUnlock() + return len(fake.disconnectedArgsForCall) +} + +func (fake *FakeLocalParticipant) DisconnectedCalls(stub func() <-chan struct{}) { + fake.disconnectedMutex.Lock() + defer fake.disconnectedMutex.Unlock() + fake.DisconnectedStub = stub +} + +func (fake *FakeLocalParticipant) DisconnectedReturns(result1 <-chan struct{}) { + fake.disconnectedMutex.Lock() + defer fake.disconnectedMutex.Unlock() + fake.DisconnectedStub = nil + fake.disconnectedReturns = struct { + result1 <-chan struct{} + }{result1} +} + +func (fake *FakeLocalParticipant) DisconnectedReturnsOnCall(i int, result1 <-chan struct{}) { + fake.disconnectedMutex.Lock() + defer fake.disconnectedMutex.Unlock() + fake.DisconnectedStub = nil + if fake.disconnectedReturnsOnCall == nil { + fake.disconnectedReturnsOnCall = make(map[int]struct { + result1 <-chan struct{} + }) + } + fake.disconnectedReturnsOnCall[i] = struct { + result1 <-chan struct{} + }{result1} +} + func (fake *FakeLocalParticipant) GetAdaptiveStream() bool { fake.getAdaptiveStreamMutex.Lock() ret, specificReturn := fake.getAdaptiveStreamReturnsOnCall[len(fake.getAdaptiveStreamArgsForCall)] @@ -6408,6 +6471,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.connectedAtMutex.RUnlock() fake.debugInfoMutex.RLock() defer fake.debugInfoMutex.RUnlock() + fake.disconnectedMutex.RLock() + defer fake.disconnectedMutex.RUnlock() fake.getAdaptiveStreamMutex.RLock() defer fake.getAdaptiveStreamMutex.RUnlock() fake.getAudioLevelMutex.RLock() diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 3d3b8dd13..9855e6b64 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -620,15 +620,10 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.LocalPa _ = r.refreshToken(participant) tokenTicker := time.NewTicker(tokenRefreshInterval) defer tokenTicker.Stop() - stateCheckTicker := time.NewTicker(time.Millisecond * 500) - defer stateCheckTicker.Stop() for { select { - case <-stateCheckTicker.C: - // periodic check to ensure participant didn't become disconnected - if participant.IsDisconnected() { - return - } + case <-participant.Disconnected(): + return case <-tokenTicker.C: // refresh token with the first API Key/secret pair if err := r.refreshToken(participant); err != nil {