diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index ba72d5b5d..933d05fe9 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -249,7 +249,7 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, code // RemoveSubscriber removes participant from subscription // stop all forwarders to the client -func (t *MediaTrackSubscriptions) RemoveSubscriber(participantID livekit.ParticipantID) { +func (t *MediaTrackSubscriptions) RemoveSubscriber(participantID livekit.ParticipantID, resume bool) { subTrack := t.getSubscribedTrack(participantID) t.subscribedTracksMu.Lock() @@ -257,7 +257,7 @@ func (t *MediaTrackSubscriptions) RemoveSubscriber(participantID livekit.Partici t.subscribedTracksMu.Unlock() if subTrack != nil { - subTrack.DownTrack().Close() + subTrack.DownTrack().CloseWithFlush(!resume) } t.maybeNotifyNoSubscribers() diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 126d5d413..e1fd30e9b 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -174,7 +174,7 @@ type MediaTrack interface { // subscribers AddSubscriber(participant LocalParticipant) error - RemoveSubscriber(participantID livekit.ParticipantID) + RemoveSubscriber(participantID livekit.ParticipantID, resume bool) IsSubscriber(subID livekit.ParticipantID) bool GetAllSubscriberIDs() []livekit.ParticipantID RemoveAllSubscribers() diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index 3c1b38d2a..595a4c7ea 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -183,10 +183,11 @@ type FakeLocalMediaTrack struct { removeAllSubscribersMutex sync.RWMutex removeAllSubscribersArgsForCall []struct { } - RemoveSubscriberStub func(livekit.ParticipantID) + RemoveSubscriberStub func(livekit.ParticipantID, bool) removeSubscriberMutex sync.RWMutex removeSubscriberArgsForCall []struct { arg1 livekit.ParticipantID + arg2 bool } RevokeDisallowedSubscribersStub func([]livekit.ParticipantID) []livekit.ParticipantID revokeDisallowedSubscribersMutex sync.RWMutex @@ -1178,16 +1179,17 @@ func (fake *FakeLocalMediaTrack) RemoveAllSubscribersCalls(stub func()) { fake.RemoveAllSubscribersStub = stub } -func (fake *FakeLocalMediaTrack) RemoveSubscriber(arg1 livekit.ParticipantID) { +func (fake *FakeLocalMediaTrack) RemoveSubscriber(arg1 livekit.ParticipantID, arg2 bool) { fake.removeSubscriberMutex.Lock() fake.removeSubscriberArgsForCall = append(fake.removeSubscriberArgsForCall, struct { arg1 livekit.ParticipantID - }{arg1}) + arg2 bool + }{arg1, arg2}) stub := fake.RemoveSubscriberStub - fake.recordInvocation("RemoveSubscriber", []interface{}{arg1}) + fake.recordInvocation("RemoveSubscriber", []interface{}{arg1, arg2}) fake.removeSubscriberMutex.Unlock() if stub != nil { - fake.RemoveSubscriberStub(arg1) + fake.RemoveSubscriberStub(arg1, arg2) } } @@ -1197,17 +1199,17 @@ func (fake *FakeLocalMediaTrack) RemoveSubscriberCallCount() int { return len(fake.removeSubscriberArgsForCall) } -func (fake *FakeLocalMediaTrack) RemoveSubscriberCalls(stub func(livekit.ParticipantID)) { +func (fake *FakeLocalMediaTrack) RemoveSubscriberCalls(stub func(livekit.ParticipantID, bool)) { fake.removeSubscriberMutex.Lock() defer fake.removeSubscriberMutex.Unlock() fake.RemoveSubscriberStub = stub } -func (fake *FakeLocalMediaTrack) RemoveSubscriberArgsForCall(i int) livekit.ParticipantID { +func (fake *FakeLocalMediaTrack) RemoveSubscriberArgsForCall(i int) (livekit.ParticipantID, bool) { fake.removeSubscriberMutex.RLock() defer fake.removeSubscriberMutex.RUnlock() argsForCall := fake.removeSubscriberArgsForCall[i] - return argsForCall.arg1 + return argsForCall.arg1, argsForCall.arg2 } func (fake *FakeLocalMediaTrack) RevokeDisallowedSubscribers(arg1 []livekit.ParticipantID) []livekit.ParticipantID { diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index 8efe8760d..34988407f 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_media_track.go @@ -161,10 +161,11 @@ type FakeMediaTrack struct { removeAllSubscribersMutex sync.RWMutex removeAllSubscribersArgsForCall []struct { } - RemoveSubscriberStub func(livekit.ParticipantID) + RemoveSubscriberStub func(livekit.ParticipantID, bool) removeSubscriberMutex sync.RWMutex removeSubscriberArgsForCall []struct { arg1 livekit.ParticipantID + arg2 bool } RevokeDisallowedSubscribersStub func([]livekit.ParticipantID) []livekit.ParticipantID revokeDisallowedSubscribersMutex sync.RWMutex @@ -1027,16 +1028,17 @@ func (fake *FakeMediaTrack) RemoveAllSubscribersCalls(stub func()) { fake.RemoveAllSubscribersStub = stub } -func (fake *FakeMediaTrack) RemoveSubscriber(arg1 livekit.ParticipantID) { +func (fake *FakeMediaTrack) RemoveSubscriber(arg1 livekit.ParticipantID, arg2 bool) { fake.removeSubscriberMutex.Lock() fake.removeSubscriberArgsForCall = append(fake.removeSubscriberArgsForCall, struct { arg1 livekit.ParticipantID - }{arg1}) + arg2 bool + }{arg1, arg2}) stub := fake.RemoveSubscriberStub - fake.recordInvocation("RemoveSubscriber", []interface{}{arg1}) + fake.recordInvocation("RemoveSubscriber", []interface{}{arg1, arg2}) fake.removeSubscriberMutex.Unlock() if stub != nil { - fake.RemoveSubscriberStub(arg1) + fake.RemoveSubscriberStub(arg1, arg2) } } @@ -1046,17 +1048,17 @@ func (fake *FakeMediaTrack) RemoveSubscriberCallCount() int { return len(fake.removeSubscriberArgsForCall) } -func (fake *FakeMediaTrack) RemoveSubscriberCalls(stub func(livekit.ParticipantID)) { +func (fake *FakeMediaTrack) RemoveSubscriberCalls(stub func(livekit.ParticipantID, bool)) { fake.removeSubscriberMutex.Lock() defer fake.removeSubscriberMutex.Unlock() fake.RemoveSubscriberStub = stub } -func (fake *FakeMediaTrack) RemoveSubscriberArgsForCall(i int) livekit.ParticipantID { +func (fake *FakeMediaTrack) RemoveSubscriberArgsForCall(i int) (livekit.ParticipantID, bool) { fake.removeSubscriberMutex.RLock() defer fake.removeSubscriberMutex.RUnlock() argsForCall := fake.removeSubscriberArgsForCall[i] - return argsForCall.arg1 + return argsForCall.arg1, argsForCall.arg2 } func (fake *FakeMediaTrack) RevokeDisallowedSubscribers(arg1 []livekit.ParticipantID) []livekit.ParticipantID { diff --git a/pkg/rtc/uptrackmanager.go b/pkg/rtc/uptrackmanager.go index 37d68f34c..b25364d2a 100644 --- a/pkg/rtc/uptrackmanager.go +++ b/pkg/rtc/uptrackmanager.go @@ -130,7 +130,7 @@ func (u *UpTrackManager) AddSubscriber(sub types.LocalParticipant, params types. func (u *UpTrackManager) RemoveSubscriber(sub types.LocalParticipant, trackID livekit.TrackID) { track := u.GetPublishedTrack(trackID) if track != nil { - track.RemoveSubscriber(sub.ID()) + track.RemoveSubscriber(sub.ID(), false) } u.lock.Lock() diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 51bbfc4bd..f1ce7fe0c 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -430,15 +430,26 @@ func (d *DownTrack) Mute(val bool) { } } -// Close track func (d *DownTrack) Close() { + d.CloseWithFlush(true) +} + +// Close track, flush used to indicate whether send blank frame to flush +// decoder of client. +// 1. When transceiver is reused by other participant's video track, +// set flush=true to avoid previous video shows before previous stream is displayed. +// 2. in case of session migration, participant migrate from other node, video track should +// be resumed with same participant, set flush=false since we don't need flush decoder. +func (d *DownTrack) CloseWithFlush(flush bool) { d.forwarder.Mute(true) // write blank frames after disabling so that other frames do not interfere. // Idea here is to send blank 1x1 key frames to flush the decoder buffer at the remote end. // Otherwise, with transceiver re-use last frame from previous stream is held in the // display buffer and there could be a brief moment where the previous stream is displayed. - _ = d.writeBlankFrameRTP() + if flush { + _ = d.writeBlankFrameRTP() + } d.closeOnce.Do(func() { Logger.V(1).Info("Closing sender", "peer_id", d.peerID, "kind", d.kind)