Adjust interfaces to better work with cloud (#349)

* Adjust interfaces to better work with cloud

* force argument to quality update

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)

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)

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

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()

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()

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()

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