SubscribedQualityUpdate message (#270)

* WIP commit

* SubscribedQualityUpdate message to send list of currently subscribed
qualities for a simulcast video publisher

* Correct subscriberID

* goimports

* Do quality update on add/remove of subscribed track

* do not update quality when admin mute is active

* update quality on admin unmute

* Update protocol version

* Simplify max subscribed quality loop per David's suggestion

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/google/wire v0.5.0
 	github.com/gorilla/websocket v1.4.2
 	github.com/hashicorp/golang-lru v0.5.4
-	github.com/livekit/protocol v0.11.1-0.20211210234141-81dc05762739
+	github.com/livekit/protocol v0.11.1
 	github.com/magefile/mage v1.11.0
 	github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0
 	github.com/mitchellh/go-homedir v1.1.0

go.sum

@@ -132,8 +132,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY/mKrcvA=
 github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts=
-github.com/livekit/protocol v0.11.1-0.20211210234141-81dc05762739 h1:/y/ve7kJ+2aJkhzn5n4jwzPMI4Jow8TUFp63uXsUP64=
-github.com/livekit/protocol v0.11.1-0.20211210234141-81dc05762739/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg=
+github.com/livekit/protocol v0.11.1 h1:SY9oZlbHD9s2fNjus5zSmTapI3uOIbb6YLkZYtIJESs=
+github.com/livekit/protocol v0.11.1/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg=
 github.com/magefile/mage v1.11.0 h1:C/55Ywp9BpgVVclD3lRnSYCwXTYxmSppIgLeDYlNuls=
 github.com/magefile/mage v1.11.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A=
 github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0=

pkg/rtc/mediatrack.go

@@ -65,6 +65,13 @@ type MediaTrack struct {
 	maxUpFracLost     uint8
 	maxUpFracLostTs   time.Time
 
+	// quality level enable/disable
+	maxQualityLock               sync.Mutex
+	maxSubscriberQuality         map[string]livekit.VideoQuality
+	maxSubscribedQuality         livekit.VideoQuality
+	allSubscribersMuted          bool
+	onSubscribedMaxQualityChange func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error
+
 	onClose []func()
 }
 
@@ -84,10 +91,11 @@ type MediaTrackParams struct {
 
 func NewMediaTrack(track *webrtc.TrackRemote, params MediaTrackParams) *MediaTrack {
 	t := &MediaTrack{
-		params:   params,
-		ssrc:     track.SSRC(),
-		streamID: track.StreamID(),
-		codec:    track.Codec(),
+		params:               params,
+		ssrc:                 track.SSRC(),
+		streamID:             track.StreamID(),
+		codec:                track.Codec(),
+		maxSubscriberQuality: make(map[string]livekit.VideoQuality),
 	}
 
 	if params.TrackInfo.Muted {
@@ -149,6 +157,11 @@ func (t *MediaTrack) SetMuted(muted bool) {
 		}
 		return true
 	})
+
+	// update quality based on subscription if unmuting
+	if !muted {
+		t.updateQualityChange()
+	}
 }
 
 func (t *MediaTrack) AddOnClose(f func()) {
@@ -201,7 +214,7 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
 	if err != nil {
 		return err
 	}
-	subTrack := NewSubscribedTrack(t, t.params.ParticipantIdentity, downTrack)
+	subTrack := NewSubscribedTrack(t, sub.ID(), t.params.ParticipantIdentity, downTrack)
 
 	var transceiver *webrtc.RTPTransceiver
 	var sender *webrtc.RTPSender
@@ -295,6 +308,7 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
 				}
 			}
 
+			t.NotifySubscriberMute(sub.ID())
 			sub.RemoveSubscribedTrack(subTrack)
 			sub.Negotiate()
 		}()
@@ -309,6 +323,7 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
 	t.receiver.AddDownTrack(downTrack)
 	// since sub will lock, run it in a goroutine to avoid deadlocks
 	go func() {
+		t.NotifySubscriberMaxQuality(sub.ID(), livekit.VideoQuality_HIGH) // start with HIGH, let subscription change it later
 		sub.AddSubscribedTrack(subTrack)
 		sub.Negotiate()
 	}()
@@ -652,3 +667,99 @@ func (t *MediaTrack) DebugInfo() map[string]interface{} {
 func (t *MediaTrack) Receiver() sfu.TrackReceiver {
 	return t.receiver
 }
+
+func (t *MediaTrack) OnSubscribedMaxQualityChange(f func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error) {
+	t.onSubscribedMaxQualityChange = f
+}
+
+func (t *MediaTrack) NotifySubscriberMute(subscriberID string) {
+	if !t.IsSimulcast() {
+		return
+	}
+
+	t.maxQualityLock.Lock()
+	_, ok := t.maxSubscriberQuality[subscriberID]
+	if !ok {
+		t.maxQualityLock.Unlock()
+		return
+	}
+
+	delete(t.maxSubscriberQuality, subscriberID)
+	t.maxQualityLock.Unlock()
+
+	t.updateQualityChange()
+}
+
+func (t *MediaTrack) NotifySubscriberMaxQuality(subscriberID string, quality livekit.VideoQuality) {
+	if !t.IsSimulcast() {
+		return
+	}
+
+	t.maxQualityLock.Lock()
+	maxQuality, ok := t.maxSubscriberQuality[subscriberID]
+	if ok && maxQuality == quality {
+		t.maxQualityLock.Unlock()
+		return
+	}
+
+	t.maxSubscriberQuality[subscriberID] = quality
+	t.maxQualityLock.Unlock()
+
+	t.updateQualityChange()
+}
+
+func (t *MediaTrack) updateQualityChange() {
+	if t.IsMuted() {
+		return
+	}
+
+	var subscribedQualities []*livekit.SubscribedQuality
+
+	t.maxQualityLock.Lock()
+	allSubscribersMuted := false
+	maxSubscribedQuality := livekit.VideoQuality_LOW
+	if len(t.maxSubscriberQuality) == 0 {
+		allSubscribersMuted = true
+	} else {
+		for _, subQuality := range t.maxSubscriberQuality {
+			if subQuality > maxSubscribedQuality {
+				maxSubscribedQuality = subQuality
+			}
+		}
+	}
+
+	if allSubscribersMuted {
+		if !t.allSubscribersMuted {
+			t.allSubscribersMuted = true
+			subscribedQualities = []*livekit.SubscribedQuality{
+				&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: false},
+				&livekit.SubscribedQuality{Quality: livekit.VideoQuality_MEDIUM, Enabled: false},
+				&livekit.SubscribedQuality{Quality: livekit.VideoQuality_HIGH, Enabled: false},
+			}
+		}
+	} else {
+		t.allSubscribersMuted = false
+		if maxSubscribedQuality != t.maxSubscribedQuality {
+			t.maxSubscribedQuality = maxSubscribedQuality
+
+			subscribedQualities = append(subscribedQualities, &livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: true})
+
+			if t.maxSubscribedQuality == livekit.VideoQuality_LOW {
+				subscribedQualities = append(subscribedQualities, &livekit.SubscribedQuality{Quality: livekit.VideoQuality_MEDIUM, Enabled: false})
+			} else {
+				subscribedQualities = append(subscribedQualities, &livekit.SubscribedQuality{Quality: livekit.VideoQuality_MEDIUM, Enabled: true})
+			}
+
+			if t.maxSubscribedQuality != livekit.VideoQuality_HIGH {
+				subscribedQualities = append(subscribedQualities, &livekit.SubscribedQuality{Quality: livekit.VideoQuality_HIGH, Enabled: false})
+			} else {
+				subscribedQualities = append(subscribedQualities, &livekit.SubscribedQuality{Quality: livekit.VideoQuality_HIGH, Enabled: true})
+			}
+		}
+	}
+	t.maxQualityLock.Unlock()
+
+	if len(subscribedQualities) != 0 && t.onSubscribedMaxQualityChange != nil {
+		t.onSubscribedMaxQualityChange(t.ID(), subscribedQualities)
+	}
+}

pkg/rtc/mediatrack_test.go

@@ -99,3 +99,144 @@ func TestGetQualityForDimension(t *testing.T) {
 		require.Equal(t, livekit.VideoQuality_HIGH, mt.GetQualityForDimension(1000, 700))
 	})
 }
+
+func TestSubscribedMaxQuality(t *testing.T) {
+	t.Run("subscribers muted", func(t *testing.T) {
+		mt := NewMediaTrack(&webrtc.TrackRemote{}, MediaTrackParams{TrackInfo: &livekit.TrackInfo{
+			Sid:    "v1",
+			Type:   livekit.TrackType_VIDEO,
+			Width:  1080,
+			Height: 720,
+			Layers: []*livekit.VideoLayer{
+				{
+					Quality: livekit.VideoQuality_LOW,
+					Width:   480,
+					Height:  270,
+				},
+				{
+					Quality: livekit.VideoQuality_MEDIUM,
+					Width:   960,
+					Height:  540,
+				},
+				{
+					Quality: livekit.VideoQuality_HIGH,
+					Width:   1080,
+					Height:  720,
+				},
+			},
+		}})
+		mt.simulcasted.TrySet(true)
+
+		mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_HIGH)
+
+		actualTrackSid := ""
+		actualSubscribedQualities := []*livekit.SubscribedQuality{}
+		mt.OnSubscribedMaxQualityChange(func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error {
+			actualTrackSid = trackSid
+			actualSubscribedQualities = subscribedQualities
+			return nil
+		})
+
+		// mute all subscribers
+		mt.NotifySubscriberMute("s1")
+
+		expectedSubscribedQualities := []*livekit.SubscribedQuality{
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: false},
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_MEDIUM, Enabled: false},
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_HIGH, Enabled: false},
+		}
+		require.Equal(t, "v1", actualTrackSid)
+		require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities)
+	})
+
+	t.Run("subscribers max quality", func(t *testing.T) {
+		mt := NewMediaTrack(&webrtc.TrackRemote{}, MediaTrackParams{TrackInfo: &livekit.TrackInfo{
+			Sid:    "v1",
+			Type:   livekit.TrackType_VIDEO,
+			Width:  1080,
+			Height: 720,
+			Layers: []*livekit.VideoLayer{
+				{
+					Quality: livekit.VideoQuality_LOW,
+					Width:   480,
+					Height:  270,
+				},
+				{
+					Quality: livekit.VideoQuality_MEDIUM,
+					Width:   960,
+					Height:  540,
+				},
+				{
+					Quality: livekit.VideoQuality_HIGH,
+					Width:   1080,
+					Height:  720,
+				},
+			},
+		}})
+		mt.simulcasted.TrySet(true)
+
+		actualTrackSid := ""
+		actualSubscribedQualities := []*livekit.SubscribedQuality{}
+		mt.OnSubscribedMaxQualityChange(func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error {
+			actualTrackSid = trackSid
+			actualSubscribedQualities = subscribedQualities
+			return nil
+		})
+
+		mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_HIGH)
+		mt.NotifySubscriberMaxQuality("s2", livekit.VideoQuality_MEDIUM)
+
+		expectedSubscribedQualities := []*livekit.SubscribedQuality{
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: true},
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_MEDIUM, Enabled: true},
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_HIGH, Enabled: true},
+		}
+		require.Equal(t, "v1", actualTrackSid)
+		require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities)
+
+		// "s1" dropping to MEDIUM should disable HIGH layer
+		mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_MEDIUM)
+
+		expectedSubscribedQualities = []*livekit.SubscribedQuality{
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: true},
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_MEDIUM, Enabled: true},
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_HIGH, Enabled: false},
+		}
+		require.Equal(t, "v1", actualTrackSid)
+		require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities)
+
+		// "s1" and "s1" dropping to LOW should disable HIGH & MEDIUM
+		mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_LOW)
+		mt.NotifySubscriberMaxQuality("s2", livekit.VideoQuality_LOW)
+
+		expectedSubscribedQualities = []*livekit.SubscribedQuality{
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: true},
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_MEDIUM, Enabled: false},
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_HIGH, Enabled: false},
+		}
+		require.Equal(t, "v1", actualTrackSid)
+		require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities)
+
+		// muting one should still produce LOW
+		mt.NotifySubscriberMute("s1")
+
+		expectedSubscribedQualities = []*livekit.SubscribedQuality{
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: true},
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_MEDIUM, Enabled: false},
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_HIGH, Enabled: false},
+		}
+		require.Equal(t, "v1", actualTrackSid)
+		require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities)
+
+		// muting "s2" should disable all qualities
+		mt.NotifySubscriberMute("s2")
+
+		expectedSubscribedQualities = []*livekit.SubscribedQuality{
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: false},
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_MEDIUM, Enabled: false},
+			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_HIGH, Enabled: false},
+		}
+		require.Equal(t, "v1", actualTrackSid)
+		require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities)
+	})
+}

pkg/rtc/participant.go

@@ -938,6 +938,8 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w
 			Logger:              p.params.Logger,
 		})
 
+		mt.OnSubscribedMaxQualityChange(p.onSubscribedMaxQualityChange)
+
 		// add to published and clean up pending
 		p.publishedTracks[mt.ID()] = mt
 		delete(p.pendingTracks, signalCid)
@@ -1312,6 +1314,23 @@ func (p *ParticipantImpl) onStreamStateChange(update *sfu.StreamStateUpdate) err
 	})
 }
 
+func (p *ParticipantImpl) onSubscribedMaxQualityChange(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error {
+	if len(subscribedQualities) == 0 {
+		return nil
+	}
+
+	subscribedQualityUpdate := &livekit.SubscribedQualityUpdate{
+		TrackSid:            trackSid,
+		SubscribedQualities: subscribedQualities,
+	}
+
+	return p.writeMessage(&livekit.SignalResponse{
+		Message: &livekit.SignalResponse_SubscribedQualityUpdate{
+			SubscribedQualityUpdate: subscribedQualityUpdate,
+		},
+	})
+}
+
 func (p *ParticipantImpl) DebugInfo() map[string]interface{} {
 	info := map[string]interface{}{
 		"ID":    p.params.SID,

pkg/rtc/subscribedtrack.go

@@ -21,6 +21,7 @@ type SubscribedTrack struct {
 	publishedTrack    types.MediaTrack
 	dt                *sfu.DownTrack
 	publisherIdentity string
+	subscriberID      string
 	subMuted          utils.AtomicFlag
 	pubMuted          utils.AtomicFlag
 	settings          atomic.Value // *livekit.UpdateTrackSettings
@@ -30,10 +31,11 @@ type SubscribedTrack struct {
 	debouncer func(func())
 }
 
-func NewSubscribedTrack(mediaTrack types.MediaTrack, publisherIdentity string, dt *sfu.DownTrack) *SubscribedTrack {
+func NewSubscribedTrack(mediaTrack types.MediaTrack, publisherIdentity string, subscriberID string, dt *sfu.DownTrack) *SubscribedTrack {
 	return &SubscribedTrack{
 		publishedTrack:    mediaTrack,
 		publisherIdentity: publisherIdentity,
+		subscriberID:      subscriberID,
 		dt:                dt,
 		debouncer:         debounce.New(subscriptionDebounceInterval),
 	}
@@ -92,7 +94,11 @@ func (t *SubscribedTrack) UpdateSubscriberSettings(settings *livekit.UpdateTrack
 
 func (t *SubscribedTrack) UpdateVideoLayer() {
 	t.updateDownTrackMute()
-	if t.subMuted.Get() || t.dt.Kind() != webrtc.RTPCodecTypeVideo {
+	if t.dt.Kind() != webrtc.RTPCodecTypeVideo {
+		return
+	}
+	if t.subMuted.Get() {
+		t.publishedTrack.NotifySubscriberMute(t.subscriberID)
 		return
 	}
 	settings, ok := t.settings.Load().(*livekit.UpdateTrackSettings)
@@ -105,6 +111,8 @@ func (t *SubscribedTrack) UpdateVideoLayer() {
 		quality = t.publishedTrack.GetQualityForDimension(settings.Width, settings.Height)
 	}
 	t.dt.SetMaxSpatialLayer(spatialLayerForQuality(quality))
+
+	t.publishedTrack.NotifySubscriberMaxQuality(t.subscriberID, quality)
 }
 
 func (t *SubscribedTrack) updateDownTrackMute() {

pkg/rtc/types/interfaces.go

@@ -116,6 +116,10 @@ type MediaTrack interface {
 	RemoveAllSubscribers()
 	// returns quality information that's appropriate for width & height
 	GetQualityForDimension(width, height uint32) livekit.VideoQuality
+
+	NotifySubscriberMute(subscriberID string)
+	NotifySubscriberMaxQuality(subscriberID string, quality livekit.VideoQuality)
+	OnSubscribedMaxQualityChange(f func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error)
 }
 
 // PublishedTrack is the main interface representing a track published to the room

pkg/rtc/types/typesfakes/fake_media_track.go

@@ -93,6 +93,22 @@ type FakeMediaTrack struct {
 	nameReturnsOnCall map[int]struct {
 		result1 string
 	}
+	NotifySubscriberMaxQualityStub        func(string, livekit.VideoQuality)
+	notifySubscriberMaxQualityMutex       sync.RWMutex
+	notifySubscriberMaxQualityArgsForCall []struct {
+		arg1 string
+		arg2 livekit.VideoQuality
+	}
+	NotifySubscriberMuteStub        func(string)
+	notifySubscriberMuteMutex       sync.RWMutex
+	notifySubscriberMuteArgsForCall []struct {
+		arg1 string
+	}
+	OnSubscribedMaxQualityChangeStub        func(func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error)
+	onSubscribedMaxQualityChangeMutex       sync.RWMutex
+	onSubscribedMaxQualityChangeArgsForCall []struct {
+		arg1 func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error
+	}
 	RemoveAllSubscribersStub        func()
 	removeAllSubscribersMutex       sync.RWMutex
 	removeAllSubscribersArgsForCall []struct {
@@ -575,6 +591,103 @@ func (fake *FakeMediaTrack) NameReturnsOnCall(i int, result1 string) {
 	}{result1}
 }
 
+func (fake *FakeMediaTrack) NotifySubscriberMaxQuality(arg1 string, arg2 livekit.VideoQuality) {
+	fake.notifySubscriberMaxQualityMutex.Lock()
+	fake.notifySubscriberMaxQualityArgsForCall = append(fake.notifySubscriberMaxQualityArgsForCall, struct {
+		arg1 string
+		arg2 livekit.VideoQuality
+	}{arg1, arg2})
+	stub := fake.NotifySubscriberMaxQualityStub
+	fake.recordInvocation("NotifySubscriberMaxQuality", []interface{}{arg1, arg2})
+	fake.notifySubscriberMaxQualityMutex.Unlock()
+	if stub != nil {
+		fake.NotifySubscriberMaxQualityStub(arg1, arg2)
+	}
+}
+
+func (fake *FakeMediaTrack) NotifySubscriberMaxQualityCallCount() int {
+	fake.notifySubscriberMaxQualityMutex.RLock()
+	defer fake.notifySubscriberMaxQualityMutex.RUnlock()
+	return len(fake.notifySubscriberMaxQualityArgsForCall)
+}
+
+func (fake *FakeMediaTrack) NotifySubscriberMaxQualityCalls(stub func(string, livekit.VideoQuality)) {
+	fake.notifySubscriberMaxQualityMutex.Lock()
+	defer fake.notifySubscriberMaxQualityMutex.Unlock()
+	fake.NotifySubscriberMaxQualityStub = stub
+}
+
+func (fake *FakeMediaTrack) NotifySubscriberMaxQualityArgsForCall(i int) (string, livekit.VideoQuality) {
+	fake.notifySubscriberMaxQualityMutex.RLock()
+	defer fake.notifySubscriberMaxQualityMutex.RUnlock()
+	argsForCall := fake.notifySubscriberMaxQualityArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2
+}
+
+func (fake *FakeMediaTrack) NotifySubscriberMute(arg1 string) {
+	fake.notifySubscriberMuteMutex.Lock()
+	fake.notifySubscriberMuteArgsForCall = append(fake.notifySubscriberMuteArgsForCall, struct {
+		arg1 string
+	}{arg1})
+	stub := fake.NotifySubscriberMuteStub
+	fake.recordInvocation("NotifySubscriberMute", []interface{}{arg1})
+	fake.notifySubscriberMuteMutex.Unlock()
+	if stub != nil {
+		fake.NotifySubscriberMuteStub(arg1)
+	}
+}
+
+func (fake *FakeMediaTrack) NotifySubscriberMuteCallCount() int {
+	fake.notifySubscriberMuteMutex.RLock()
+	defer fake.notifySubscriberMuteMutex.RUnlock()
+	return len(fake.notifySubscriberMuteArgsForCall)
+}
+
+func (fake *FakeMediaTrack) NotifySubscriberMuteCalls(stub func(string)) {
+	fake.notifySubscriberMuteMutex.Lock()
+	defer fake.notifySubscriberMuteMutex.Unlock()
+	fake.NotifySubscriberMuteStub = stub
+}
+
+func (fake *FakeMediaTrack) NotifySubscriberMuteArgsForCall(i int) string {
+	fake.notifySubscriberMuteMutex.RLock()
+	defer fake.notifySubscriberMuteMutex.RUnlock()
+	argsForCall := fake.notifySubscriberMuteArgsForCall[i]
+	return argsForCall.arg1
+}
+
+func (fake *FakeMediaTrack) OnSubscribedMaxQualityChange(arg1 func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error) {
+	fake.onSubscribedMaxQualityChangeMutex.Lock()
+	fake.onSubscribedMaxQualityChangeArgsForCall = append(fake.onSubscribedMaxQualityChangeArgsForCall, struct {
+		arg1 func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error
+	}{arg1})
+	stub := fake.OnSubscribedMaxQualityChangeStub
+	fake.recordInvocation("OnSubscribedMaxQualityChange", []interface{}{arg1})
+	fake.onSubscribedMaxQualityChangeMutex.Unlock()
+	if stub != nil {
+		fake.OnSubscribedMaxQualityChangeStub(arg1)
+	}
+}
+
+func (fake *FakeMediaTrack) OnSubscribedMaxQualityChangeCallCount() int {
+	fake.onSubscribedMaxQualityChangeMutex.RLock()
+	defer fake.onSubscribedMaxQualityChangeMutex.RUnlock()
+	return len(fake.onSubscribedMaxQualityChangeArgsForCall)
+}
+
+func (fake *FakeMediaTrack) OnSubscribedMaxQualityChangeCalls(stub func(func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error)) {
+	fake.onSubscribedMaxQualityChangeMutex.Lock()
+	defer fake.onSubscribedMaxQualityChangeMutex.Unlock()
+	fake.OnSubscribedMaxQualityChangeStub = stub
+}
+
+func (fake *FakeMediaTrack) OnSubscribedMaxQualityChangeArgsForCall(i int) func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error {
+	fake.onSubscribedMaxQualityChangeMutex.RLock()
+	defer fake.onSubscribedMaxQualityChangeMutex.RUnlock()
+	argsForCall := fake.onSubscribedMaxQualityChangeArgsForCall[i]
+	return argsForCall.arg1
+}
+
 func (fake *FakeMediaTrack) RemoveAllSubscribers() {
 	fake.removeAllSubscribersMutex.Lock()
 	fake.removeAllSubscribersArgsForCall = append(fake.removeAllSubscribersArgsForCall, struct {
@@ -772,6 +885,12 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} {
 	defer fake.kindMutex.RUnlock()
 	fake.nameMutex.RLock()
 	defer fake.nameMutex.RUnlock()
+	fake.notifySubscriberMaxQualityMutex.RLock()
+	defer fake.notifySubscriberMaxQualityMutex.RUnlock()
+	fake.notifySubscriberMuteMutex.RLock()
+	defer fake.notifySubscriberMuteMutex.RUnlock()
+	fake.onSubscribedMaxQualityChangeMutex.RLock()
+	defer fake.onSubscribedMaxQualityChangeMutex.RUnlock()
 	fake.removeAllSubscribersMutex.RLock()
 	defer fake.removeAllSubscribersMutex.RUnlock()
 	fake.removeSubscriberMutex.RLock()

pkg/rtc/types/typesfakes/fake_published_track.go

@@ -99,6 +99,17 @@ type FakePublishedTrack struct {
 	nameReturnsOnCall map[int]struct {
 		result1 string
 	}
+	NotifySubscriberMaxQualityStub        func(string, livekit.VideoQuality)
+	notifySubscriberMaxQualityMutex       sync.RWMutex
+	notifySubscriberMaxQualityArgsForCall []struct {
+		arg1 string
+		arg2 livekit.VideoQuality
+	}
+	NotifySubscriberMuteStub        func(string)
+	notifySubscriberMuteMutex       sync.RWMutex
+	notifySubscriberMuteArgsForCall []struct {
+		arg1 string
+	}
 	NumUpTracksStub        func() (uint32, uint32)
 	numUpTracksMutex       sync.RWMutex
 	numUpTracksArgsForCall []struct {
@@ -111,6 +122,11 @@ type FakePublishedTrack struct {
 		result1 uint32
 		result2 uint32
 	}
+	OnSubscribedMaxQualityChangeStub        func(func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error)
+	onSubscribedMaxQualityChangeMutex       sync.RWMutex
+	onSubscribedMaxQualityChangeArgsForCall []struct {
+		arg1 func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error
+	}
 	PublishLossPercentageStub        func() uint32
 	publishLossPercentageMutex       sync.RWMutex
 	publishLossPercentageArgsForCall []struct {
@@ -675,6 +691,71 @@ func (fake *FakePublishedTrack) NameReturnsOnCall(i int, result1 string) {
 	}{result1}
 }
 
+func (fake *FakePublishedTrack) NotifySubscriberMaxQuality(arg1 string, arg2 livekit.VideoQuality) {
+	fake.notifySubscriberMaxQualityMutex.Lock()
+	fake.notifySubscriberMaxQualityArgsForCall = append(fake.notifySubscriberMaxQualityArgsForCall, struct {
+		arg1 string
+		arg2 livekit.VideoQuality
+	}{arg1, arg2})
+	stub := fake.NotifySubscriberMaxQualityStub
+	fake.recordInvocation("NotifySubscriberMaxQuality", []interface{}{arg1, arg2})
+	fake.notifySubscriberMaxQualityMutex.Unlock()
+	if stub != nil {
+		fake.NotifySubscriberMaxQualityStub(arg1, arg2)
+	}
+}
+
+func (fake *FakePublishedTrack) NotifySubscriberMaxQualityCallCount() int {
+	fake.notifySubscriberMaxQualityMutex.RLock()
+	defer fake.notifySubscriberMaxQualityMutex.RUnlock()
+	return len(fake.notifySubscriberMaxQualityArgsForCall)
+}
+
+func (fake *FakePublishedTrack) NotifySubscriberMaxQualityCalls(stub func(string, livekit.VideoQuality)) {
+	fake.notifySubscriberMaxQualityMutex.Lock()
+	defer fake.notifySubscriberMaxQualityMutex.Unlock()
+	fake.NotifySubscriberMaxQualityStub = stub
+}
+
+func (fake *FakePublishedTrack) NotifySubscriberMaxQualityArgsForCall(i int) (string, livekit.VideoQuality) {
+	fake.notifySubscriberMaxQualityMutex.RLock()
+	defer fake.notifySubscriberMaxQualityMutex.RUnlock()
+	argsForCall := fake.notifySubscriberMaxQualityArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2
+}
+
+func (fake *FakePublishedTrack) NotifySubscriberMute(arg1 string) {
+	fake.notifySubscriberMuteMutex.Lock()
+	fake.notifySubscriberMuteArgsForCall = append(fake.notifySubscriberMuteArgsForCall, struct {
+		arg1 string
+	}{arg1})
+	stub := fake.NotifySubscriberMuteStub
+	fake.recordInvocation("NotifySubscriberMute", []interface{}{arg1})
+	fake.notifySubscriberMuteMutex.Unlock()
+	if stub != nil {
+		fake.NotifySubscriberMuteStub(arg1)
+	}
+}
+
+func (fake *FakePublishedTrack) NotifySubscriberMuteCallCount() int {
+	fake.notifySubscriberMuteMutex.RLock()
+	defer fake.notifySubscriberMuteMutex.RUnlock()
+	return len(fake.notifySubscriberMuteArgsForCall)
+}
+
+func (fake *FakePublishedTrack) NotifySubscriberMuteCalls(stub func(string)) {
+	fake.notifySubscriberMuteMutex.Lock()
+	defer fake.notifySubscriberMuteMutex.Unlock()
+	fake.NotifySubscriberMuteStub = stub
+}
+
+func (fake *FakePublishedTrack) NotifySubscriberMuteArgsForCall(i int) string {
+	fake.notifySubscriberMuteMutex.RLock()
+	defer fake.notifySubscriberMuteMutex.RUnlock()
+	argsForCall := fake.notifySubscriberMuteArgsForCall[i]
+	return argsForCall.arg1
+}
+
 func (fake *FakePublishedTrack) NumUpTracks() (uint32, uint32) {
 	fake.numUpTracksMutex.Lock()
 	ret, specificReturn := fake.numUpTracksReturnsOnCall[len(fake.numUpTracksArgsForCall)]
@@ -731,6 +812,38 @@ func (fake *FakePublishedTrack) NumUpTracksReturnsOnCall(i int, result1 uint32,
 	}{result1, result2}
 }
 
+func (fake *FakePublishedTrack) OnSubscribedMaxQualityChange(arg1 func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error) {
+	fake.onSubscribedMaxQualityChangeMutex.Lock()
+	fake.onSubscribedMaxQualityChangeArgsForCall = append(fake.onSubscribedMaxQualityChangeArgsForCall, struct {
+		arg1 func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error
+	}{arg1})
+	stub := fake.OnSubscribedMaxQualityChangeStub
+	fake.recordInvocation("OnSubscribedMaxQualityChange", []interface{}{arg1})
+	fake.onSubscribedMaxQualityChangeMutex.Unlock()
+	if stub != nil {
+		fake.OnSubscribedMaxQualityChangeStub(arg1)
+	}
+}
+
+func (fake *FakePublishedTrack) OnSubscribedMaxQualityChangeCallCount() int {
+	fake.onSubscribedMaxQualityChangeMutex.RLock()
+	defer fake.onSubscribedMaxQualityChangeMutex.RUnlock()
+	return len(fake.onSubscribedMaxQualityChangeArgsForCall)
+}
+
+func (fake *FakePublishedTrack) OnSubscribedMaxQualityChangeCalls(stub func(func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error)) {
+	fake.onSubscribedMaxQualityChangeMutex.Lock()
+	defer fake.onSubscribedMaxQualityChangeMutex.Unlock()
+	fake.OnSubscribedMaxQualityChangeStub = stub
+}
+
+func (fake *FakePublishedTrack) OnSubscribedMaxQualityChangeArgsForCall(i int) func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error {
+	fake.onSubscribedMaxQualityChangeMutex.RLock()
+	defer fake.onSubscribedMaxQualityChangeMutex.RUnlock()
+	argsForCall := fake.onSubscribedMaxQualityChangeArgsForCall[i]
+	return argsForCall.arg1
+}
+
 func (fake *FakePublishedTrack) PublishLossPercentage() uint32 {
 	fake.publishLossPercentageMutex.Lock()
 	ret, specificReturn := fake.publishLossPercentageReturnsOnCall[len(fake.publishLossPercentageArgsForCall)]
@@ -1195,8 +1308,14 @@ func (fake *FakePublishedTrack) Invocations() map[string][][]interface{} {
 	defer fake.kindMutex.RUnlock()
 	fake.nameMutex.RLock()
 	defer fake.nameMutex.RUnlock()
+	fake.notifySubscriberMaxQualityMutex.RLock()
+	defer fake.notifySubscriberMaxQualityMutex.RUnlock()
+	fake.notifySubscriberMuteMutex.RLock()
+	defer fake.notifySubscriberMuteMutex.RUnlock()
 	fake.numUpTracksMutex.RLock()
 	defer fake.numUpTracksMutex.RUnlock()
+	fake.onSubscribedMaxQualityChangeMutex.RLock()
+	defer fake.onSubscribedMaxQualityChangeMutex.RUnlock()
 	fake.publishLossPercentageMutex.RLock()
 	defer fake.publishLossPercentageMutex.RUnlock()
 	fake.receiverMutex.RLock()

pkg/service/wire.go

@@ -10,12 +10,12 @@ import (
 
 	"github.com/go-redis/redis/v8"
 	"github.com/google/wire"
-	"github.com/livekit/protocol/livekit"
-	"github.com/pkg/errors"
 	"github.com/livekit/protocol/auth"
+	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/utils"
 	"github.com/livekit/protocol/webhook"
+	"github.com/pkg/errors"
 
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/routing"

pkg/service/wire_gen.go

@@ -1,7 +1,8 @@
 // Code generated by Wire. DO NOT EDIT.
 
 //go:generate go run github.com/google/wire/cmd/wire
-//+build !wireinject
+//go:build !wireinject
+// +build !wireinject
 
 package service
 

pkg/sfu/forwarder_test.go

@@ -493,7 +493,7 @@ func TestForwarderFinalizeAllocate(t *testing.T) {
 		bandwidthRequested: 0,
 		bandwidthDelta:     0,
 		availableLayers:    nil,
-		bitrates:           Bitrates{{0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}},
+		bitrates:           Bitrates{},
 		targetLayers:       InvalidLayers,
 		distanceToDesired:  0,
 	}

pkg/sfu/receiver.go

@@ -42,7 +42,7 @@ type Receiver interface {
 	NumAvailableSpatialLayers() int
 	GetBitrateTemporalCumulative() Bitrates
 	ReadRTP(buf []byte, layer uint8, sn uint16) (int, error)
-	DeleteDownTrack(ID string)
+	DeleteDownTrack(peerID string)
 	OnCloseHandler(fn func())
 	SendPLI(layer int32)
 	SetRTCPCh(ch chan []rtcp.Packet)
@@ -319,7 +319,7 @@ func (w *WebRTCReceiver) removeAvailableLayer(layer uint16) {
 		w.upTrackMu.Unlock()
 		return
 	}
-	newLayers := make([]uint16, 0, 3)
+	newLayers := make([]uint16, 0, DefaultMaxLayerSpatial+1)
 	for _, l := range layers {
 		if l != layer {
 			newLayers = append(newLayers, l)

test/integration_helpers.go

@@ -28,9 +28,9 @@ const (
 	testApiSecret     = "apiSecret"
 	testRoom          = "mytestroom"
 	defaultServerPort = 7880
-	secondServerPort = 8880
-	nodeID1 = "node-1"
-	nodeID2 = "node-2"
+	secondServerPort  = 8880
+	nodeID1           = "node-1"
+	nodeID2           = "node-2"
 
 	syncDelay = 100 * time.Millisecond
 	// if there are deadlocks, it's helpful to set a short test timeout (i.e. go test -timeout=30s)

tools/tools.go

@@ -1,3 +1,4 @@
+//go:build tools
 // +build tools
 
 package tools