From 10714b3f6dcae9ecab47177e1b76d32a0726f3ad Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 19 Jan 2022 10:28:55 +0530 Subject: [PATCH] Adjust interfaces to better work with cloud (#349) * Adjust interfaces to better work with cloud * force argument to quality update --- pkg/rtc/mediatracksubscriptions.go | 58 +++---- pkg/rtc/participant.go | 32 +--- pkg/rtc/types/interfaces.go | 12 +- .../typesfakes/fake_local_media_track.go | 39 ----- pkg/rtc/types/typesfakes/fake_media_track.go | 58 +++---- pkg/rtc/types/typesfakes/fake_participant.go | 156 ++++++++++++++++++ pkg/rtc/uptrackmanager.go | 34 +++- 7 files changed, 247 insertions(+), 142 deletions(-) diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index 535637429..d8c9133b9 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -79,7 +79,7 @@ func (t *MediaTrackSubscriptions) SetMuted(muted bool) { // update quality based on subscription if unmuting if !muted { - t.updateQualityChange() + t.UpdateQualityChange(true) } } @@ -406,7 +406,7 @@ func (t *MediaTrackSubscriptions) NotifySubscriberMaxQuality(subscriberID liveki } t.maxQualityLock.Unlock() - t.updateQualityChange() + t.UpdateQualityChange(false) } func (t *MediaTrackSubscriptions) NotifySubscriberNodeMaxQuality(nodeID string, quality livekit.VideoQuality) { @@ -434,34 +434,10 @@ func (t *MediaTrackSubscriptions) NotifySubscriberNodeMaxQuality(nodeID string, } t.maxQualityLock.Unlock() - t.updateQualityChange() + t.UpdateQualityChange(false) } -func (t *MediaTrackSubscriptions) startMaxQualityTimer() { - t.maxQualityLock.Lock() - defer t.maxQualityLock.Unlock() - - if t.params.MediaTrack.Kind() != livekit.TrackType_VIDEO { - return - } - - t.maxQualityTimer = time.AfterFunc(initialQualityUpdateWait, func() { - t.stopMaxQualityTimer() - t.updateQualityChange() - }) -} - -func (t *MediaTrackSubscriptions) stopMaxQualityTimer() { - t.maxQualityLock.Lock() - defer t.maxQualityLock.Unlock() - - if t.maxQualityTimer != nil { - t.maxQualityTimer.Stop() - t.maxQualityTimer = nil - } -} - -func (t *MediaTrackSubscriptions) updateQualityChange() { +func (t *MediaTrackSubscriptions) UpdateQualityChange(force bool) { if t.params.MediaTrack.Kind() != livekit.TrackType_VIDEO { return } @@ -480,7 +456,7 @@ func (t *MediaTrackSubscriptions) updateQualityChange() { } } - if maxSubscribedQuality == t.maxSubscribedQuality { + if maxSubscribedQuality == t.maxSubscribedQuality && !force { t.maxQualityLock.Unlock() return } @@ -509,6 +485,30 @@ func (t *MediaTrackSubscriptions) updateQualityChange() { } } +func (t *MediaTrackSubscriptions) startMaxQualityTimer() { + t.maxQualityLock.Lock() + defer t.maxQualityLock.Unlock() + + if t.params.MediaTrack.Kind() != livekit.TrackType_VIDEO { + return + } + + t.maxQualityTimer = time.AfterFunc(initialQualityUpdateWait, func() { + t.stopMaxQualityTimer() + t.UpdateQualityChange(false) + }) +} + +func (t *MediaTrackSubscriptions) stopMaxQualityTimer() { + t.maxQualityLock.Lock() + defer t.maxQualityLock.Unlock() + + if t.maxQualityTimer != nil { + t.maxQualityTimer.Stop() + t.maxQualityTimer = nil + } +} + func (t *MediaTrackSubscriptions) numSubscribedLayers() uint32 { t.maxQualityLock.RLock() numSubscribedLayers := uint32(0) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 58aa0c65d..d45cf6343 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -848,34 +848,6 @@ func (p *ParticipantImpl) SubscriptionPermissionUpdate(publisherID livekit.Parti } } -func (p *ParticipantImpl) UpdateSubscribedQuality(nodeID string, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error { - track := p.GetPublishedTrack(trackID) - if track == nil { - p.params.Logger.Warnw("could not find track", nil, "trackID", trackID) - return errors.New("could not find track") - } - - if mt, ok := track.(*MediaTrack); ok { - mt.NotifySubscriberNodeMaxQuality(nodeID, maxQuality) - } - - return nil -} - -func (p *ParticipantImpl) UpdateMediaLoss(nodeID string, trackID livekit.TrackID, fractionalLoss uint32) error { - track := p.GetPublishedTrack(trackID) - if track == nil { - p.params.Logger.Warnw("could not find track", nil, "trackID", trackID) - return errors.New("could not find track") - } - - if mt, ok := track.(*MediaTrack); ok { - mt.NotifySubscriberNodeMediaLoss(nodeID, uint8(fractionalLoss)) - } - - return nil -} - func (p *ParticipantImpl) setupUptrackManager() { p.UptrackManager = NewUptrackManager(UptrackManagerParams{ SID: p.params.SID, @@ -892,8 +864,6 @@ func (p *ParticipantImpl) setupUptrackManager() { } }) - p.UptrackManager.OnSubscribedMaxQualityChange(p.onSubscribedMaxQualityChange) - p.UptrackManager.OnUptrackManagerClose(p.onUptrackManagerClose) } @@ -1375,6 +1345,8 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei } } + mt.OnSubscribedMaxQualityChange(p.onSubscribedMaxQualityChange) + // add to published and clean up pending p.UptrackManager.AddPublishedTrack(mt) delete(p.pendingTracks, signalCid) diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index b9e0c70aa..384e7acdd 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -53,8 +53,11 @@ type Participant interface { Start() Close(sendLeave bool) error + // updates from remotes UpdateSubscriptionPermissions(permissions *livekit.UpdateSubscriptionPermissions, resolver func(participantID livekit.ParticipantID) LocalParticipant) error UpdateVideoLayers(updateVideoLayers *livekit.UpdateVideoLayers) error + UpdateSubscribedQuality(nodeID string, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error + UpdateMediaLoss(nodeID string, trackID livekit.TrackID, fractionalLoss uint32) error DebugInfo() map[string]interface{} } @@ -125,10 +128,6 @@ type LocalParticipant interface { OnDataPacket(callback func(LocalParticipant, *livekit.DataPacket)) OnClose(_callback func(LocalParticipant, map[livekit.TrackID]livekit.ParticipantID)) - // updates from remotes - UpdateSubscribedQuality(nodeID string, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error - UpdateMediaLoss(nodeID string, trackID livekit.TrackID, fractionalLoss uint32) error - // session migration SetMigrateState(s MigrateState) MigrateState() MigrateState @@ -171,7 +170,6 @@ type MediaTrack interface { Receiver() sfu.TrackReceiver // callbacks - OnSubscribedMaxQualityChange(f func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxQuality livekit.VideoQuality) error) AddOnClose(func()) // subscribers @@ -187,6 +185,8 @@ type MediaTrack interface { NotifySubscriberMaxQuality(subscriberID livekit.ParticipantID, quality livekit.VideoQuality) NotifySubscriberNodeMaxQuality(nodeID string, quality livekit.VideoQuality) + + NotifySubscriberNodeMediaLoss(nodeID string, fractionalLoss uint8) } //counterfeiter:generate . LocalMediaTrack @@ -198,8 +198,6 @@ type LocalMediaTrack interface { GetAudioLevel() (level uint8, active bool) GetConnectionScore() float64 - - NotifySubscriberNodeMediaLoss(nodeID string, fractionalLoss uint8) } // MediaTrack is the main interface representing a track published to the room diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index ecb3b435d..3c1b38d2a 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -149,11 +149,6 @@ type FakeLocalMediaTrack struct { arg1 string arg2 uint8 } - OnSubscribedMaxQualityChangeStub func(func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxQuality livekit.VideoQuality) error) - onSubscribedMaxQualityChangeMutex sync.RWMutex - onSubscribedMaxQualityChangeArgsForCall []struct { - arg1 func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxQuality livekit.VideoQuality) error - } PublisherIDStub func() livekit.ParticipantID publisherIDMutex sync.RWMutex publisherIDArgsForCall []struct { @@ -1000,38 +995,6 @@ func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMediaLossArgsForCall(i int) return argsForCall.arg1, argsForCall.arg2 } -func (fake *FakeLocalMediaTrack) OnSubscribedMaxQualityChange(arg1 func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxQuality livekit.VideoQuality) error) { - fake.onSubscribedMaxQualityChangeMutex.Lock() - fake.onSubscribedMaxQualityChangeArgsForCall = append(fake.onSubscribedMaxQualityChangeArgsForCall, struct { - arg1 func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxQuality livekit.VideoQuality) error - }{arg1}) - stub := fake.OnSubscribedMaxQualityChangeStub - fake.recordInvocation("OnSubscribedMaxQualityChange", []interface{}{arg1}) - fake.onSubscribedMaxQualityChangeMutex.Unlock() - if stub != nil { - fake.OnSubscribedMaxQualityChangeStub(arg1) - } -} - -func (fake *FakeLocalMediaTrack) OnSubscribedMaxQualityChangeCallCount() int { - fake.onSubscribedMaxQualityChangeMutex.RLock() - defer fake.onSubscribedMaxQualityChangeMutex.RUnlock() - return len(fake.onSubscribedMaxQualityChangeArgsForCall) -} - -func (fake *FakeLocalMediaTrack) OnSubscribedMaxQualityChangeCalls(stub func(func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxQuality livekit.VideoQuality) error)) { - fake.onSubscribedMaxQualityChangeMutex.Lock() - defer fake.onSubscribedMaxQualityChangeMutex.Unlock() - fake.OnSubscribedMaxQualityChangeStub = stub -} - -func (fake *FakeLocalMediaTrack) OnSubscribedMaxQualityChangeArgsForCall(i int) func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxQuality livekit.VideoQuality) error { - fake.onSubscribedMaxQualityChangeMutex.RLock() - defer fake.onSubscribedMaxQualityChangeMutex.RUnlock() - argsForCall := fake.onSubscribedMaxQualityChangeArgsForCall[i] - return argsForCall.arg1 -} - func (fake *FakeLocalMediaTrack) PublisherID() livekit.ParticipantID { fake.publisherIDMutex.Lock() ret, specificReturn := fake.publisherIDReturnsOnCall[len(fake.publisherIDArgsForCall)] @@ -1627,8 +1590,6 @@ func (fake *FakeLocalMediaTrack) Invocations() map[string][][]interface{} { defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock() fake.notifySubscriberNodeMediaLossMutex.RLock() defer fake.notifySubscriberNodeMediaLossMutex.RUnlock() - fake.onSubscribedMaxQualityChangeMutex.RLock() - defer fake.onSubscribedMaxQualityChangeMutex.RUnlock() fake.publisherIDMutex.RLock() defer fake.publisherIDMutex.RUnlock() fake.publisherIdentityMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index 244332347..8efe8760d 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_media_track.go @@ -121,10 +121,11 @@ type FakeMediaTrack struct { arg1 string arg2 livekit.VideoQuality } - OnSubscribedMaxQualityChangeStub func(func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxQuality livekit.VideoQuality) error) - onSubscribedMaxQualityChangeMutex sync.RWMutex - onSubscribedMaxQualityChangeArgsForCall []struct { - arg1 func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxQuality livekit.VideoQuality) error + NotifySubscriberNodeMediaLossStub func(string, uint8) + notifySubscriberNodeMediaLossMutex sync.RWMutex + notifySubscriberNodeMediaLossArgsForCall []struct { + arg1 string + arg2 uint8 } PublisherIDStub func() livekit.ParticipantID publisherIDMutex sync.RWMutex @@ -810,36 +811,37 @@ func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityArgsForCall(i int) (st return argsForCall.arg1, argsForCall.arg2 } -func (fake *FakeMediaTrack) OnSubscribedMaxQualityChange(arg1 func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxQuality livekit.VideoQuality) error) { - fake.onSubscribedMaxQualityChangeMutex.Lock() - fake.onSubscribedMaxQualityChangeArgsForCall = append(fake.onSubscribedMaxQualityChangeArgsForCall, struct { - arg1 func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxQuality livekit.VideoQuality) error - }{arg1}) - stub := fake.OnSubscribedMaxQualityChangeStub - fake.recordInvocation("OnSubscribedMaxQualityChange", []interface{}{arg1}) - fake.onSubscribedMaxQualityChangeMutex.Unlock() +func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLoss(arg1 string, arg2 uint8) { + fake.notifySubscriberNodeMediaLossMutex.Lock() + fake.notifySubscriberNodeMediaLossArgsForCall = append(fake.notifySubscriberNodeMediaLossArgsForCall, struct { + arg1 string + arg2 uint8 + }{arg1, arg2}) + stub := fake.NotifySubscriberNodeMediaLossStub + fake.recordInvocation("NotifySubscriberNodeMediaLoss", []interface{}{arg1, arg2}) + fake.notifySubscriberNodeMediaLossMutex.Unlock() if stub != nil { - fake.OnSubscribedMaxQualityChangeStub(arg1) + fake.NotifySubscriberNodeMediaLossStub(arg1, arg2) } } -func (fake *FakeMediaTrack) OnSubscribedMaxQualityChangeCallCount() int { - fake.onSubscribedMaxQualityChangeMutex.RLock() - defer fake.onSubscribedMaxQualityChangeMutex.RUnlock() - return len(fake.onSubscribedMaxQualityChangeArgsForCall) +func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLossCallCount() int { + fake.notifySubscriberNodeMediaLossMutex.RLock() + defer fake.notifySubscriberNodeMediaLossMutex.RUnlock() + return len(fake.notifySubscriberNodeMediaLossArgsForCall) } -func (fake *FakeMediaTrack) OnSubscribedMaxQualityChangeCalls(stub func(func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxQuality livekit.VideoQuality) error)) { - fake.onSubscribedMaxQualityChangeMutex.Lock() - defer fake.onSubscribedMaxQualityChangeMutex.Unlock() - fake.OnSubscribedMaxQualityChangeStub = stub +func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLossCalls(stub func(string, uint8)) { + fake.notifySubscriberNodeMediaLossMutex.Lock() + defer fake.notifySubscriberNodeMediaLossMutex.Unlock() + fake.NotifySubscriberNodeMediaLossStub = stub } -func (fake *FakeMediaTrack) OnSubscribedMaxQualityChangeArgsForCall(i int) func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxQuality livekit.VideoQuality) error { - fake.onSubscribedMaxQualityChangeMutex.RLock() - defer fake.onSubscribedMaxQualityChangeMutex.RUnlock() - argsForCall := fake.onSubscribedMaxQualityChangeArgsForCall[i] - return argsForCall.arg1 +func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLossArgsForCall(i int) (string, uint8) { + fake.notifySubscriberNodeMediaLossMutex.RLock() + defer fake.notifySubscriberNodeMediaLossMutex.RUnlock() + argsForCall := fake.notifySubscriberNodeMediaLossArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 } func (fake *FakeMediaTrack) PublisherID() livekit.ParticipantID { @@ -1325,8 +1327,8 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} { defer fake.notifySubscriberMaxQualityMutex.RUnlock() fake.notifySubscriberNodeMaxQualityMutex.RLock() defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock() - fake.onSubscribedMaxQualityChangeMutex.RLock() - defer fake.onSubscribedMaxQualityChangeMutex.RUnlock() + fake.notifySubscriberNodeMediaLossMutex.RLock() + defer fake.notifySubscriberNodeMediaLossMutex.RUnlock() fake.publisherIDMutex.RLock() defer fake.publisherIDMutex.RUnlock() fake.publisherIdentityMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index 507b7b87e..d599431e9 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/pkg/rtc/types/typesfakes/fake_participant.go @@ -120,6 +120,32 @@ type FakeParticipant struct { toProtoReturnsOnCall map[int]struct { result1 *livekit.ParticipantInfo } + UpdateMediaLossStub func(string, livekit.TrackID, uint32) error + updateMediaLossMutex sync.RWMutex + updateMediaLossArgsForCall []struct { + arg1 string + arg2 livekit.TrackID + arg3 uint32 + } + updateMediaLossReturns struct { + result1 error + } + updateMediaLossReturnsOnCall map[int]struct { + result1 error + } + UpdateSubscribedQualityStub func(string, livekit.TrackID, livekit.VideoQuality) error + updateSubscribedQualityMutex sync.RWMutex + updateSubscribedQualityArgsForCall []struct { + arg1 string + arg2 livekit.TrackID + arg3 livekit.VideoQuality + } + updateSubscribedQualityReturns struct { + result1 error + } + updateSubscribedQualityReturnsOnCall map[int]struct { + result1 error + } UpdateSubscriptionPermissionsStub func(*livekit.UpdateSubscriptionPermissions, func(participantID livekit.ParticipantID) types.LocalParticipant) error updateSubscriptionPermissionsMutex sync.RWMutex updateSubscriptionPermissionsArgsForCall []struct { @@ -741,6 +767,132 @@ func (fake *FakeParticipant) ToProtoReturnsOnCall(i int, result1 *livekit.Partic }{result1} } +func (fake *FakeParticipant) UpdateMediaLoss(arg1 string, arg2 livekit.TrackID, arg3 uint32) error { + fake.updateMediaLossMutex.Lock() + ret, specificReturn := fake.updateMediaLossReturnsOnCall[len(fake.updateMediaLossArgsForCall)] + fake.updateMediaLossArgsForCall = append(fake.updateMediaLossArgsForCall, struct { + arg1 string + arg2 livekit.TrackID + arg3 uint32 + }{arg1, arg2, arg3}) + stub := fake.UpdateMediaLossStub + fakeReturns := fake.updateMediaLossReturns + fake.recordInvocation("UpdateMediaLoss", []interface{}{arg1, arg2, arg3}) + fake.updateMediaLossMutex.Unlock() + if stub != nil { + return stub(arg1, arg2, arg3) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeParticipant) UpdateMediaLossCallCount() int { + fake.updateMediaLossMutex.RLock() + defer fake.updateMediaLossMutex.RUnlock() + return len(fake.updateMediaLossArgsForCall) +} + +func (fake *FakeParticipant) UpdateMediaLossCalls(stub func(string, livekit.TrackID, uint32) error) { + fake.updateMediaLossMutex.Lock() + defer fake.updateMediaLossMutex.Unlock() + fake.UpdateMediaLossStub = stub +} + +func (fake *FakeParticipant) UpdateMediaLossArgsForCall(i int) (string, livekit.TrackID, uint32) { + fake.updateMediaLossMutex.RLock() + defer fake.updateMediaLossMutex.RUnlock() + argsForCall := fake.updateMediaLossArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 +} + +func (fake *FakeParticipant) UpdateMediaLossReturns(result1 error) { + fake.updateMediaLossMutex.Lock() + defer fake.updateMediaLossMutex.Unlock() + fake.UpdateMediaLossStub = nil + fake.updateMediaLossReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeParticipant) UpdateMediaLossReturnsOnCall(i int, result1 error) { + fake.updateMediaLossMutex.Lock() + defer fake.updateMediaLossMutex.Unlock() + fake.UpdateMediaLossStub = nil + if fake.updateMediaLossReturnsOnCall == nil { + fake.updateMediaLossReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.updateMediaLossReturnsOnCall[i] = struct { + result1 error + }{result1} +} + +func (fake *FakeParticipant) UpdateSubscribedQuality(arg1 string, arg2 livekit.TrackID, arg3 livekit.VideoQuality) error { + fake.updateSubscribedQualityMutex.Lock() + ret, specificReturn := fake.updateSubscribedQualityReturnsOnCall[len(fake.updateSubscribedQualityArgsForCall)] + fake.updateSubscribedQualityArgsForCall = append(fake.updateSubscribedQualityArgsForCall, struct { + arg1 string + arg2 livekit.TrackID + arg3 livekit.VideoQuality + }{arg1, arg2, arg3}) + stub := fake.UpdateSubscribedQualityStub + fakeReturns := fake.updateSubscribedQualityReturns + fake.recordInvocation("UpdateSubscribedQuality", []interface{}{arg1, arg2, arg3}) + fake.updateSubscribedQualityMutex.Unlock() + if stub != nil { + return stub(arg1, arg2, arg3) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeParticipant) UpdateSubscribedQualityCallCount() int { + fake.updateSubscribedQualityMutex.RLock() + defer fake.updateSubscribedQualityMutex.RUnlock() + return len(fake.updateSubscribedQualityArgsForCall) +} + +func (fake *FakeParticipant) UpdateSubscribedQualityCalls(stub func(string, livekit.TrackID, livekit.VideoQuality) error) { + fake.updateSubscribedQualityMutex.Lock() + defer fake.updateSubscribedQualityMutex.Unlock() + fake.UpdateSubscribedQualityStub = stub +} + +func (fake *FakeParticipant) UpdateSubscribedQualityArgsForCall(i int) (string, livekit.TrackID, livekit.VideoQuality) { + fake.updateSubscribedQualityMutex.RLock() + defer fake.updateSubscribedQualityMutex.RUnlock() + argsForCall := fake.updateSubscribedQualityArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 +} + +func (fake *FakeParticipant) UpdateSubscribedQualityReturns(result1 error) { + fake.updateSubscribedQualityMutex.Lock() + defer fake.updateSubscribedQualityMutex.Unlock() + fake.UpdateSubscribedQualityStub = nil + fake.updateSubscribedQualityReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeParticipant) UpdateSubscribedQualityReturnsOnCall(i int, result1 error) { + fake.updateSubscribedQualityMutex.Lock() + defer fake.updateSubscribedQualityMutex.Unlock() + fake.UpdateSubscribedQualityStub = nil + if fake.updateSubscribedQualityReturnsOnCall == nil { + fake.updateSubscribedQualityReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.updateSubscribedQualityReturnsOnCall[i] = struct { + result1 error + }{result1} +} + func (fake *FakeParticipant) UpdateSubscriptionPermissions(arg1 *livekit.UpdateSubscriptionPermissions, arg2 func(participantID livekit.ParticipantID) types.LocalParticipant) error { fake.updateSubscriptionPermissionsMutex.Lock() ret, specificReturn := fake.updateSubscriptionPermissionsReturnsOnCall[len(fake.updateSubscriptionPermissionsArgsForCall)] @@ -891,6 +1043,10 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} { defer fake.startMutex.RUnlock() fake.toProtoMutex.RLock() defer fake.toProtoMutex.RUnlock() + fake.updateMediaLossMutex.RLock() + defer fake.updateMediaLossMutex.RUnlock() + fake.updateSubscribedQualityMutex.RLock() + defer fake.updateSubscribedQualityMutex.RUnlock() fake.updateSubscriptionPermissionsMutex.RLock() defer fake.updateSubscriptionPermissionsMutex.RUnlock() fake.updateVideoLayersMutex.RLock() diff --git a/pkg/rtc/uptrackmanager.go b/pkg/rtc/uptrackmanager.go index f6d893d92..61ca2c55d 100644 --- a/pkg/rtc/uptrackmanager.go +++ b/pkg/rtc/uptrackmanager.go @@ -30,9 +30,8 @@ type UptrackManager struct { lock sync.RWMutex // callbacks & handlers - onClose func() - onTrackUpdated func(track types.MediaTrack, onlyIfReady bool) - onSubscribedMaxQualityChange func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxSubscribedQuality livekit.VideoQuality) error + onClose func() + onTrackUpdated func(track types.MediaTrack, onlyIfReady bool) } func NewUptrackManager(params UptrackManagerParams) *UptrackManager { @@ -83,10 +82,6 @@ func (u *UptrackManager) OnPublishedTrackUpdated(f func(track types.MediaTrack, u.onTrackUpdated = f } -func (u *UptrackManager) OnSubscribedMaxQualityChange(f func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxSubscribedQuality livekit.VideoQuality) error) { - u.onSubscribedMaxQualityChange = f -} - // AddSubscriber subscribes op to all publishedTracks func (u *UptrackManager) AddSubscriber(sub types.LocalParticipant, params types.AddSubscriberParams) (int, error) { var tracks []types.MediaTrack @@ -200,6 +195,7 @@ func (u *UptrackManager) UpdateSubscriptionPermissions( func (u *UptrackManager) UpdateVideoLayers(updateVideoLayers *livekit.UpdateVideoLayers) error { track := u.GetPublishedTrack(livekit.TrackID(updateVideoLayers.TrackSid)) if track == nil { + u.params.Logger.Warnw("could not find track", nil, "trackID", livekit.TrackID(updateVideoLayers.TrackSid)) return errors.New("could not find published track") } @@ -207,9 +203,29 @@ func (u *UptrackManager) UpdateVideoLayers(updateVideoLayers *livekit.UpdateVide return nil } -func (u *UptrackManager) AddPublishedTrack(track types.MediaTrack) { - track.OnSubscribedMaxQualityChange(u.onSubscribedMaxQualityChange) +func (u *UptrackManager) UpdateSubscribedQuality(nodeID string, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error { + track := u.GetPublishedTrack(trackID) + if track == nil { + u.params.Logger.Warnw("could not find track", nil, "trackID", trackID) + return errors.New("could not find published track") + } + track.NotifySubscriberNodeMaxQuality(nodeID, maxQuality) + return nil +} + +func (u *UptrackManager) UpdateMediaLoss(nodeID string, trackID livekit.TrackID, fractionalLoss uint32) error { + track := u.GetPublishedTrack(trackID) + if track == nil { + u.params.Logger.Warnw("could not find track", nil, "trackID", trackID) + return errors.New("could not find published track") + } + + track.NotifySubscriberNodeMediaLoss(nodeID, uint8(fractionalLoss)) + return nil +} + +func (u *UptrackManager) AddPublishedTrack(track types.MediaTrack) { u.lock.Lock() if _, ok := u.publishedTracks[track.ID()]; !ok { u.publishedTracks[track.ID()] = track