diff --git a/go.mod b/go.mod index f9a3e9019..2429276e4 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 40d44ddb1..c9b97427d 100644 --- a/go.sum +++ b/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= diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 573511d2e..7a2b07710 100644 --- a/pkg/rtc/mediatrack.go +++ b/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) + } +} diff --git a/pkg/rtc/mediatrack_test.go b/pkg/rtc/mediatrack_test.go index 20b9cdcb1..fb0e84b9f 100644 --- a/pkg/rtc/mediatrack_test.go +++ b/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) + }) +} diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index feb3cc670..201e52896 100644 --- a/pkg/rtc/participant.go +++ b/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, diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index 763945ffe..f6dfdd3dd 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/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() { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index e0b06f6b7..6b70523a1 100644 --- a/pkg/rtc/types/interfaces.go +++ b/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 diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index 2b44870f1..4f41fba30 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/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() diff --git a/pkg/rtc/types/typesfakes/fake_published_track.go b/pkg/rtc/types/typesfakes/fake_published_track.go index 1b975c905..414ed06d2 100644 --- a/pkg/rtc/types/typesfakes/fake_published_track.go +++ b/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() diff --git a/pkg/service/wire.go b/pkg/service/wire.go index 11ebce294..0f0e20a37 100644 --- a/pkg/service/wire.go +++ b/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" diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index 0570b3e6f..7c2853612 100644 --- a/pkg/service/wire_gen.go +++ b/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 diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 046f164ae..c223466af 100644 --- a/pkg/sfu/forwarder_test.go +++ b/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, } diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 37ba438a8..8a5ff59c6 100644 --- a/pkg/sfu/receiver.go +++ b/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) diff --git a/test/integration_helpers.go b/test/integration_helpers.go index 6bd5c095e..1fd93430e 100644 --- a/test/integration_helpers.go +++ b/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) diff --git a/tools/tools.go b/tools/tools.go index 102931637..5a341066b 100644 --- a/tools/tools.go +++ b/tools/tools.go @@ -1,3 +1,4 @@ +//go:build tools // +build tools package tools