From 1216113b35a1cd1420dbfc3908e50425152c1fe8 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 24 Jun 2025 12:54:34 -0700 Subject: [PATCH] Do not need to just clean up receivers. Remove that interface. (#3760) The tracks are always closed. --- pkg/rtc/mediatrack.go | 1 - pkg/rtc/mediatrackreceiver.go | 4 +- pkg/rtc/participant.go | 2 +- pkg/rtc/room.go | 2 +- pkg/rtc/types/interfaces.go | 3 +- .../typesfakes/fake_local_media_track.go | 39 ------------------- .../typesfakes/fake_local_participant.go | 18 ++++----- pkg/rtc/types/typesfakes/fake_media_track.go | 39 ------------------- pkg/rtc/types/typesfakes/fake_participant.go | 18 ++++----- pkg/rtc/uptrackmanager.go | 9 ++--- 10 files changed, 25 insertions(+), 110 deletions(-) diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 595bc39f4..332899d09 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -507,7 +507,6 @@ func (t *MediaTrack) Close(isExpectedToResume bool) { if t.dynacastManager != nil { t.dynacastManager.Close() } - t.MediaTrackReceiver.ClearAllReceivers(isExpectedToResume) t.MediaTrackReceiver.Close(isExpectedToResume) } diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index 16140e92f..e7efab68b 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -376,7 +376,7 @@ func (t *MediaTrackReceiver) ClearReceiver(mime mime.MimeType, isExpectedToResum t.removeAllSubscribersForMime(mime, isExpectedToResume) } -func (t *MediaTrackReceiver) ClearAllReceivers(isExpectedToResume bool) { +func (t *MediaTrackReceiver) clearAllReceivers(isExpectedToResume bool) { t.params.Logger.Debugw("clearing all receivers", "isExpectedToResume", isExpectedToResume) t.lock.Lock() receivers := t.receivers @@ -448,6 +448,8 @@ func (t *MediaTrackReceiver) TryClose() bool { } func (t *MediaTrackReceiver) Close(isExpectedToResume bool) { + t.clearAllReceivers(isExpectedToResume) + t.lock.Lock() if t.state == mediaTrackReceiverStateClosed { t.lock.Unlock() diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index bc97939a1..9703e743d 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -2000,7 +2000,7 @@ func (p *ParticipantImpl) onSubscriberOffer(offer webrtc.SessionDescription, off } func (p *ParticipantImpl) removePublishedTrack(track types.MediaTrack) { - p.RemovePublishedTrack(track, false, true) + p.RemovePublishedTrack(track, false) if p.ProtocolVersion().SupportsUnpublish() { p.sendTrackUnpublished(track.ID()) } else { diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 3e8942f7e..282616ce3 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -742,7 +742,7 @@ func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, pID livek // remove all published tracks for _, t := range p.GetPublishedTracks() { - p.RemovePublishedTrack(t, false, true) + p.RemovePublishedTrack(t, false) r.trackManager.RemoveTrack(t) } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 0228f1e4c..7d88617f3 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -284,7 +284,7 @@ type Participant interface { IsPublisher() bool GetPublishedTrack(trackID livekit.TrackID) MediaTrack GetPublishedTracks() []MediaTrack - RemovePublishedTrack(track MediaTrack, isExpectedToResume bool, shouldClose bool) + RemovePublishedTrack(track MediaTrack, isExpectedToResume bool) GetAudioLevel() (smoothedLevel float64, active bool) @@ -574,7 +574,6 @@ type MediaTrack interface { GetTemporalLayerForSpatialFps(spatial int32, fps uint32, mime mime.MimeType) int32 Receivers() []sfu.TrackReceiver - ClearAllReceivers(isExpectedToResume bool) IsEncrypted() bool } diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index 39d18da94..14a452886 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -30,11 +30,6 @@ type FakeLocalMediaTrack struct { result1 types.SubscribedTrack result2 error } - ClearAllReceiversStub func(bool) - clearAllReceiversMutex sync.RWMutex - clearAllReceiversArgsForCall []struct { - arg1 bool - } ClearSubscriberNodesMaxQualityStub func() clearSubscriberNodesMaxQualityMutex sync.RWMutex clearSubscriberNodesMaxQualityArgsForCall []struct { @@ -467,38 +462,6 @@ func (fake *FakeLocalMediaTrack) AddSubscriberReturnsOnCall(i int, result1 types }{result1, result2} } -func (fake *FakeLocalMediaTrack) ClearAllReceivers(arg1 bool) { - fake.clearAllReceiversMutex.Lock() - fake.clearAllReceiversArgsForCall = append(fake.clearAllReceiversArgsForCall, struct { - arg1 bool - }{arg1}) - stub := fake.ClearAllReceiversStub - fake.recordInvocation("ClearAllReceivers", []interface{}{arg1}) - fake.clearAllReceiversMutex.Unlock() - if stub != nil { - fake.ClearAllReceiversStub(arg1) - } -} - -func (fake *FakeLocalMediaTrack) ClearAllReceiversCallCount() int { - fake.clearAllReceiversMutex.RLock() - defer fake.clearAllReceiversMutex.RUnlock() - return len(fake.clearAllReceiversArgsForCall) -} - -func (fake *FakeLocalMediaTrack) ClearAllReceiversCalls(stub func(bool)) { - fake.clearAllReceiversMutex.Lock() - defer fake.clearAllReceiversMutex.Unlock() - fake.ClearAllReceiversStub = stub -} - -func (fake *FakeLocalMediaTrack) ClearAllReceiversArgsForCall(i int) bool { - fake.clearAllReceiversMutex.RLock() - defer fake.clearAllReceiversMutex.RUnlock() - argsForCall := fake.clearAllReceiversArgsForCall[i] - return argsForCall.arg1 -} - func (fake *FakeLocalMediaTrack) ClearSubscriberNodesMaxQuality() { fake.clearSubscriberNodesMaxQualityMutex.Lock() fake.clearSubscriberNodesMaxQualityArgsForCall = append(fake.clearSubscriberNodesMaxQualityArgsForCall, struct { @@ -2306,8 +2269,6 @@ func (fake *FakeLocalMediaTrack) Invocations() map[string][][]interface{} { defer fake.addOnCloseMutex.RUnlock() fake.addSubscriberMutex.RLock() defer fake.addSubscriberMutex.RUnlock() - fake.clearAllReceiversMutex.RLock() - defer fake.clearAllReceiversMutex.RUnlock() fake.clearSubscriberNodesMaxQualityMutex.RLock() defer fake.clearSubscriberNodesMaxQualityMutex.RUnlock() fake.closeMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 5e0c1d28a..fc6257067 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -851,12 +851,11 @@ type FakeLocalParticipant struct { protocolVersionReturnsOnCall map[int]struct { result1 types.ProtocolVersion } - RemovePublishedTrackStub func(types.MediaTrack, bool, bool) + RemovePublishedTrackStub func(types.MediaTrack, bool) removePublishedTrackMutex sync.RWMutex removePublishedTrackArgsForCall []struct { arg1 types.MediaTrack arg2 bool - arg3 bool } RemoveTrackLocalStub func(*webrtc.RTPSender) error removeTrackLocalMutex sync.RWMutex @@ -5771,18 +5770,17 @@ func (fake *FakeLocalParticipant) ProtocolVersionReturnsOnCall(i int, result1 ty }{result1} } -func (fake *FakeLocalParticipant) RemovePublishedTrack(arg1 types.MediaTrack, arg2 bool, arg3 bool) { +func (fake *FakeLocalParticipant) RemovePublishedTrack(arg1 types.MediaTrack, arg2 bool) { fake.removePublishedTrackMutex.Lock() fake.removePublishedTrackArgsForCall = append(fake.removePublishedTrackArgsForCall, struct { arg1 types.MediaTrack arg2 bool - arg3 bool - }{arg1, arg2, arg3}) + }{arg1, arg2}) stub := fake.RemovePublishedTrackStub - fake.recordInvocation("RemovePublishedTrack", []interface{}{arg1, arg2, arg3}) + fake.recordInvocation("RemovePublishedTrack", []interface{}{arg1, arg2}) fake.removePublishedTrackMutex.Unlock() if stub != nil { - fake.RemovePublishedTrackStub(arg1, arg2, arg3) + fake.RemovePublishedTrackStub(arg1, arg2) } } @@ -5792,17 +5790,17 @@ func (fake *FakeLocalParticipant) RemovePublishedTrackCallCount() int { return len(fake.removePublishedTrackArgsForCall) } -func (fake *FakeLocalParticipant) RemovePublishedTrackCalls(stub func(types.MediaTrack, bool, bool)) { +func (fake *FakeLocalParticipant) RemovePublishedTrackCalls(stub func(types.MediaTrack, bool)) { fake.removePublishedTrackMutex.Lock() defer fake.removePublishedTrackMutex.Unlock() fake.RemovePublishedTrackStub = stub } -func (fake *FakeLocalParticipant) RemovePublishedTrackArgsForCall(i int) (types.MediaTrack, bool, bool) { +func (fake *FakeLocalParticipant) RemovePublishedTrackArgsForCall(i int) (types.MediaTrack, bool) { fake.removePublishedTrackMutex.RLock() defer fake.removePublishedTrackMutex.RUnlock() argsForCall := fake.removePublishedTrackArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 + return argsForCall.arg1, argsForCall.arg2 } func (fake *FakeLocalParticipant) RemoveTrackLocal(arg1 *webrtc.RTPSender) error { diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index 58415b3f2..8b32f21a9 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_media_track.go @@ -30,11 +30,6 @@ type FakeMediaTrack struct { result1 types.SubscribedTrack result2 error } - ClearAllReceiversStub func(bool) - clearAllReceiversMutex sync.RWMutex - clearAllReceiversArgsForCall []struct { - arg1 bool - } CloseStub func(bool) closeMutex sync.RWMutex closeArgsForCall []struct { @@ -399,38 +394,6 @@ func (fake *FakeMediaTrack) AddSubscriberReturnsOnCall(i int, result1 types.Subs }{result1, result2} } -func (fake *FakeMediaTrack) ClearAllReceivers(arg1 bool) { - fake.clearAllReceiversMutex.Lock() - fake.clearAllReceiversArgsForCall = append(fake.clearAllReceiversArgsForCall, struct { - arg1 bool - }{arg1}) - stub := fake.ClearAllReceiversStub - fake.recordInvocation("ClearAllReceivers", []interface{}{arg1}) - fake.clearAllReceiversMutex.Unlock() - if stub != nil { - fake.ClearAllReceiversStub(arg1) - } -} - -func (fake *FakeMediaTrack) ClearAllReceiversCallCount() int { - fake.clearAllReceiversMutex.RLock() - defer fake.clearAllReceiversMutex.RUnlock() - return len(fake.clearAllReceiversArgsForCall) -} - -func (fake *FakeMediaTrack) ClearAllReceiversCalls(stub func(bool)) { - fake.clearAllReceiversMutex.Lock() - defer fake.clearAllReceiversMutex.Unlock() - fake.ClearAllReceiversStub = stub -} - -func (fake *FakeMediaTrack) ClearAllReceiversArgsForCall(i int) bool { - fake.clearAllReceiversMutex.RLock() - defer fake.clearAllReceiversMutex.RUnlock() - argsForCall := fake.clearAllReceiversArgsForCall[i] - return argsForCall.arg1 -} - func (fake *FakeMediaTrack) Close(arg1 bool) { fake.closeMutex.Lock() fake.closeArgsForCall = append(fake.closeArgsForCall, struct { @@ -1864,8 +1827,6 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} { defer fake.addOnCloseMutex.RUnlock() fake.addSubscriberMutex.RLock() defer fake.addSubscriberMutex.RUnlock() - fake.clearAllReceiversMutex.RLock() - defer fake.clearAllReceiversMutex.RUnlock() fake.closeMutex.RLock() defer fake.closeMutex.RUnlock() fake.getAllSubscribersMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index 9ec45f43b..d305384cf 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/pkg/rtc/types/typesfakes/fake_participant.go @@ -194,12 +194,11 @@ type FakeParticipant struct { onMetricsArgsForCall []struct { arg1 func(types.Participant, *livekit.DataPacket) } - RemovePublishedTrackStub func(types.MediaTrack, bool, bool) + RemovePublishedTrackStub func(types.MediaTrack, bool) removePublishedTrackMutex sync.RWMutex removePublishedTrackArgsForCall []struct { arg1 types.MediaTrack arg2 bool - arg3 bool } StateStub func() livekit.ParticipantInfo_State stateMutex sync.RWMutex @@ -1223,18 +1222,17 @@ func (fake *FakeParticipant) OnMetricsArgsForCall(i int) func(types.Participant, return argsForCall.arg1 } -func (fake *FakeParticipant) RemovePublishedTrack(arg1 types.MediaTrack, arg2 bool, arg3 bool) { +func (fake *FakeParticipant) RemovePublishedTrack(arg1 types.MediaTrack, arg2 bool) { fake.removePublishedTrackMutex.Lock() fake.removePublishedTrackArgsForCall = append(fake.removePublishedTrackArgsForCall, struct { arg1 types.MediaTrack arg2 bool - arg3 bool - }{arg1, arg2, arg3}) + }{arg1, arg2}) stub := fake.RemovePublishedTrackStub - fake.recordInvocation("RemovePublishedTrack", []interface{}{arg1, arg2, arg3}) + fake.recordInvocation("RemovePublishedTrack", []interface{}{arg1, arg2}) fake.removePublishedTrackMutex.Unlock() if stub != nil { - fake.RemovePublishedTrackStub(arg1, arg2, arg3) + fake.RemovePublishedTrackStub(arg1, arg2) } } @@ -1244,17 +1242,17 @@ func (fake *FakeParticipant) RemovePublishedTrackCallCount() int { return len(fake.removePublishedTrackArgsForCall) } -func (fake *FakeParticipant) RemovePublishedTrackCalls(stub func(types.MediaTrack, bool, bool)) { +func (fake *FakeParticipant) RemovePublishedTrackCalls(stub func(types.MediaTrack, bool)) { fake.removePublishedTrackMutex.Lock() defer fake.removePublishedTrackMutex.Unlock() fake.RemovePublishedTrackStub = stub } -func (fake *FakeParticipant) RemovePublishedTrackArgsForCall(i int) (types.MediaTrack, bool, bool) { +func (fake *FakeParticipant) RemovePublishedTrackArgsForCall(i int) (types.MediaTrack, bool) { fake.removePublishedTrackMutex.RLock() defer fake.removePublishedTrackMutex.RUnlock() argsForCall := fake.removePublishedTrackArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 + return argsForCall.arg1, argsForCall.arg2 } func (fake *FakeParticipant) State() livekit.ParticipantInfo_State { diff --git a/pkg/rtc/uptrackmanager.go b/pkg/rtc/uptrackmanager.go index 45ad411f3..eef19fae9 100644 --- a/pkg/rtc/uptrackmanager.go +++ b/pkg/rtc/uptrackmanager.go @@ -274,12 +274,9 @@ func (u *UpTrackManager) AddPublishedTrack(track types.MediaTrack) { }) } -func (u *UpTrackManager) RemovePublishedTrack(track types.MediaTrack, isExpectedToResume bool, shouldClose bool) { - if shouldClose { - track.Close(isExpectedToResume) - } else { - track.ClearAllReceivers(isExpectedToResume) - } +func (u *UpTrackManager) RemovePublishedTrack(track types.MediaTrack, isExpectedToResume bool) { + track.Close(isExpectedToResume) + u.lock.Lock() delete(u.publishedTracks, track.ID()) u.lock.Unlock()