diff --git a/pkg/rtc/mediatrack_test.go b/pkg/rtc/mediatrack_test.go index 3cfc441f0..26dcdfd4a 100644 --- a/pkg/rtc/mediatrack_test.go +++ b/pkg/rtc/mediatrack_test.go @@ -126,7 +126,7 @@ func TestSubscribedMaxQuality(t *testing.T) { }, }}) - mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_HIGH) + mt.notifySubscriberMaxQuality("s1", livekit.VideoQuality_HIGH) actualTrackID := livekit.TrackID("") actualSubscribedQualities := make([]*livekit.SubscribedQuality, 0) @@ -137,7 +137,7 @@ func TestSubscribedMaxQuality(t *testing.T) { }) // mute all subscribers - mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_OFF) + mt.notifySubscriberMaxQuality("s1", livekit.VideoQuality_OFF) expectedSubscribedQualities := []*livekit.SubscribedQuality{ {Quality: livekit.VideoQuality_LOW, Enabled: false}, @@ -181,8 +181,8 @@ func TestSubscribedMaxQuality(t *testing.T) { return nil }) - mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_HIGH) - mt.NotifySubscriberMaxQuality("s2", livekit.VideoQuality_MEDIUM) + mt.notifySubscriberMaxQuality("s1", livekit.VideoQuality_HIGH) + mt.notifySubscriberMaxQuality("s2", livekit.VideoQuality_MEDIUM) expectedSubscribedQualities := []*livekit.SubscribedQuality{ {Quality: livekit.VideoQuality_LOW, Enabled: true}, @@ -193,7 +193,7 @@ func TestSubscribedMaxQuality(t *testing.T) { require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) // "s1" dropping to MEDIUM should disable HIGH layer - mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_MEDIUM) + mt.notifySubscriberMaxQuality("s1", livekit.VideoQuality_MEDIUM) expectedSubscribedQualities = []*livekit.SubscribedQuality{ {Quality: livekit.VideoQuality_LOW, Enabled: true}, @@ -204,8 +204,8 @@ func TestSubscribedMaxQuality(t *testing.T) { 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) + mt.notifySubscriberMaxQuality("s1", livekit.VideoQuality_LOW) + mt.notifySubscriberMaxQuality("s2", livekit.VideoQuality_LOW) expectedSubscribedQualities = []*livekit.SubscribedQuality{ {Quality: livekit.VideoQuality_LOW, Enabled: true}, @@ -216,7 +216,7 @@ func TestSubscribedMaxQuality(t *testing.T) { require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) // muting "s2" only should not disable all qualities - mt.NotifySubscriberMaxQuality("s2", livekit.VideoQuality_OFF) + mt.notifySubscriberMaxQuality("s2", livekit.VideoQuality_OFF) expectedSubscribedQualities = []*livekit.SubscribedQuality{ {Quality: livekit.VideoQuality_LOW, Enabled: true}, @@ -227,7 +227,7 @@ func TestSubscribedMaxQuality(t *testing.T) { require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) // muting "s1" also should disable all qualities - mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_OFF) + mt.notifySubscriberMaxQuality("s1", livekit.VideoQuality_OFF) expectedSubscribedQualities = []*livekit.SubscribedQuality{ {Quality: livekit.VideoQuality_LOW, Enabled: false}, @@ -238,12 +238,25 @@ func TestSubscribedMaxQuality(t *testing.T) { require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) // unmuting "s1" should enable previously set max quality - mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_LOW) + mt.notifySubscriberMaxQuality("s1", livekit.VideoQuality_LOW) + expectedSubscribedQualities = []*livekit.SubscribedQuality{ {Quality: livekit.VideoQuality_LOW, Enabled: true}, {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, {Quality: livekit.VideoQuality_HIGH, Enabled: false}, } + require.Equal(t, livekit.TrackID("v1"), actualTrackID) + require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) + + // a higher quality from a different node should trigger that quality + mt.NotifySubscriberNodeMaxQuality("n1", livekit.VideoQuality_HIGH) + + expectedSubscribedQualities = []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: true}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: true}, + {Quality: livekit.VideoQuality_HIGH, Enabled: true}, + } + require.Equal(t, livekit.TrackID("v1"), actualTrackID) require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) }) } diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index 01b58c800..280098a15 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -396,6 +396,21 @@ func SpatialLayerForQuality(quality livekit.VideoQuality) int32 { } } +func QualityForSpatialLayer(layer int32) livekit.VideoQuality { + switch layer { + case 0: + return livekit.VideoQuality_LOW + case 1: + return livekit.VideoQuality_MEDIUM + case 2: + return livekit.VideoQuality_HIGH + case sfu.InvalidLayerSpatial: + return livekit.VideoQuality_OFF + default: + return livekit.VideoQuality_OFF + } +} + func VideoQualityToRID(q livekit.VideoQuality) string { switch q { case livekit.VideoQuality_HIGH: diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index 8a849c590..87a43b0a5 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -131,6 +131,7 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, code if err != nil { return nil, err } + subTrack := NewSubscribedTrack(SubscribedTrackParams{ PublisherID: t.params.MediaTrack.PublisherID(), PublisherIdentity: t.params.MediaTrack.PublisherIdentity(), @@ -182,17 +183,22 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, code downTrack.SetRTPHeaderExtensions(sendParameters.HeaderExtensions) downTrack.SetTransceiver(transceiver) + // when outtrack is bound, start loop to send reports downTrack.OnBind(func() { go subTrack.Bound() go t.sendDownTrackBindingReports(sub) }) - trackID := t.params.MediaTrack.ID() + trackID := t.params.MediaTrack.ID() downTrack.OnStatsUpdate(func(_ *sfu.DownTrack, stat *livekit.AnalyticsStat) { t.params.Telemetry.TrackStats(livekit.StreamType_DOWNSTREAM, subscriberID, trackID, stat) }) + downTrack.OnMaxLayerChanged(func(dt *sfu.DownTrack, layer int32) { + t.notifySubscriberMaxQuality(subscriberID, QualityForSpatialLayer(layer)) + }) + downTrack.OnCloseHandler(func() { t.subscribedTracksMu.Lock() delete(t.subscribedTracks, subscriberID) @@ -233,7 +239,6 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, code } } - t.NotifySubscriberMaxQuality(subscriberID, livekit.VideoQuality_OFF) sub.RemoveSubscribedTrack(subTrack) sub.Negotiate() }) @@ -243,7 +248,6 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, code // since sub will lock, run it in a goroutine to avoid deadlocks go func() { - t.NotifySubscriberMaxQuality(subscriberID, livekit.VideoQuality_HIGH) // start with HIGH, let subscription change it later sub.AddSubscribedTrack(subTrack) sub.Negotiate() }() @@ -393,7 +397,7 @@ func (t *MediaTrackSubscriptions) OnSubscribedMaxQualityChange(f func(subscribed t.onSubscribedMaxQualityChange = f } -func (t *MediaTrackSubscriptions) NotifySubscriberMaxQuality(subscriberID livekit.ParticipantID, quality livekit.VideoQuality) { +func (t *MediaTrackSubscriptions) notifySubscriberMaxQuality(subscriberID livekit.ParticipantID, quality livekit.VideoQuality) { if t.params.MediaTrack.Kind() != livekit.TrackType_VIDEO { return } diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index fafe20280..5f3951e3a 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -102,10 +102,7 @@ func (t *SubscribedTrack) UpdateVideoLayer() { if t.DownTrack().Kind() != webrtc.RTPCodecTypeVideo { return } - if t.subMuted.Get() { - t.MediaTrack().NotifySubscriberMaxQuality(t.params.SubscriberID, livekit.VideoQuality_OFF) - return - } + settings, ok := t.settings.Load().(*livekit.UpdateTrackSettings) if !ok { return @@ -116,8 +113,6 @@ func (t *SubscribedTrack) UpdateVideoLayer() { quality = t.MediaTrack().GetQualityForDimension(settings.Width, settings.Height) } t.DownTrack().SetMaxSpatialLayer(SpatialLayerForQuality(quality)) - - t.MediaTrack().NotifySubscriberMaxQuality(t.params.SubscriberID, quality) } func (t *SubscribedTrack) updateDownTrackMute() { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index b19906187..39667493f 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -199,9 +199,7 @@ type MediaTrack interface { // returns quality information that's appropriate for width & height GetQualityForDimension(width, height uint32) livekit.VideoQuality - NotifySubscriberMaxQuality(subscriberID livekit.ParticipantID, quality livekit.VideoQuality) NotifySubscriberNodeMaxQuality(nodeID string, quality livekit.VideoQuality) - NotifySubscriberNodeMediaLoss(nodeID string, fractionalLoss uint8) } diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index f5189610f..d91159979 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -121,12 +121,6 @@ type FakeLocalMediaTrack struct { nameReturnsOnCall map[int]struct { result1 string } - NotifySubscriberMaxQualityStub func(livekit.ParticipantID, livekit.VideoQuality) - notifySubscriberMaxQualityMutex sync.RWMutex - notifySubscriberMaxQualityArgsForCall []struct { - arg1 livekit.ParticipantID - arg2 livekit.VideoQuality - } NotifySubscriberNodeMaxQualityStub func(string, livekit.VideoQuality) notifySubscriberNodeMaxQualityMutex sync.RWMutex notifySubscriberNodeMaxQualityArgsForCall []struct { @@ -834,39 +828,6 @@ func (fake *FakeLocalMediaTrack) NameReturnsOnCall(i int, result1 string) { }{result1} } -func (fake *FakeLocalMediaTrack) NotifySubscriberMaxQuality(arg1 livekit.ParticipantID, arg2 livekit.VideoQuality) { - fake.notifySubscriberMaxQualityMutex.Lock() - fake.notifySubscriberMaxQualityArgsForCall = append(fake.notifySubscriberMaxQualityArgsForCall, struct { - arg1 livekit.ParticipantID - 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 *FakeLocalMediaTrack) NotifySubscriberMaxQualityCallCount() int { - fake.notifySubscriberMaxQualityMutex.RLock() - defer fake.notifySubscriberMaxQualityMutex.RUnlock() - return len(fake.notifySubscriberMaxQualityArgsForCall) -} - -func (fake *FakeLocalMediaTrack) NotifySubscriberMaxQualityCalls(stub func(livekit.ParticipantID, livekit.VideoQuality)) { - fake.notifySubscriberMaxQualityMutex.Lock() - defer fake.notifySubscriberMaxQualityMutex.Unlock() - fake.NotifySubscriberMaxQualityStub = stub -} - -func (fake *FakeLocalMediaTrack) NotifySubscriberMaxQualityArgsForCall(i int) (livekit.ParticipantID, livekit.VideoQuality) { - fake.notifySubscriberMaxQualityMutex.RLock() - defer fake.notifySubscriberMaxQualityMutex.RUnlock() - argsForCall := fake.notifySubscriberMaxQualityArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2 -} - func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMaxQuality(arg1 string, arg2 livekit.VideoQuality) { fake.notifySubscriberNodeMaxQualityMutex.Lock() fake.notifySubscriberNodeMaxQualityArgsForCall = append(fake.notifySubscriberNodeMaxQualityArgsForCall, struct { @@ -1521,8 +1482,6 @@ func (fake *FakeLocalMediaTrack) Invocations() map[string][][]interface{} { defer fake.kindMutex.RUnlock() fake.nameMutex.RLock() defer fake.nameMutex.RUnlock() - fake.notifySubscriberMaxQualityMutex.RLock() - defer fake.notifySubscriberMaxQualityMutex.RUnlock() fake.notifySubscriberNodeMaxQualityMutex.RLock() defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock() fake.notifySubscriberNodeMediaLossMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index fb4f1560e..0cc53898c 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_media_track.go @@ -99,12 +99,6 @@ type FakeMediaTrack struct { nameReturnsOnCall map[int]struct { result1 string } - NotifySubscriberMaxQualityStub func(livekit.ParticipantID, livekit.VideoQuality) - notifySubscriberMaxQualityMutex sync.RWMutex - notifySubscriberMaxQualityArgsForCall []struct { - arg1 livekit.ParticipantID - arg2 livekit.VideoQuality - } NotifySubscriberNodeMaxQualityStub func(string, livekit.VideoQuality) notifySubscriberNodeMaxQualityMutex sync.RWMutex notifySubscriberNodeMaxQualityArgsForCall []struct { @@ -683,39 +677,6 @@ func (fake *FakeMediaTrack) NameReturnsOnCall(i int, result1 string) { }{result1} } -func (fake *FakeMediaTrack) NotifySubscriberMaxQuality(arg1 livekit.ParticipantID, arg2 livekit.VideoQuality) { - fake.notifySubscriberMaxQualityMutex.Lock() - fake.notifySubscriberMaxQualityArgsForCall = append(fake.notifySubscriberMaxQualityArgsForCall, struct { - arg1 livekit.ParticipantID - 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(livekit.ParticipantID, livekit.VideoQuality)) { - fake.notifySubscriberMaxQualityMutex.Lock() - defer fake.notifySubscriberMaxQualityMutex.Unlock() - fake.NotifySubscriberMaxQualityStub = stub -} - -func (fake *FakeMediaTrack) NotifySubscriberMaxQualityArgsForCall(i int) (livekit.ParticipantID, livekit.VideoQuality) { - fake.notifySubscriberMaxQualityMutex.RLock() - defer fake.notifySubscriberMaxQualityMutex.RUnlock() - argsForCall := fake.notifySubscriberMaxQualityArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2 -} - func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQuality(arg1 string, arg2 livekit.VideoQuality) { fake.notifySubscriberNodeMaxQualityMutex.Lock() fake.notifySubscriberNodeMaxQualityArgsForCall = append(fake.notifySubscriberNodeMaxQualityArgsForCall, struct { @@ -1260,8 +1221,6 @@ 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.notifySubscriberNodeMaxQualityMutex.RLock() defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock() fake.notifySubscriberNodeMediaLossMutex.RLock() diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 12b14763b..2e7365d81 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -134,6 +134,9 @@ type DownTrack struct { // update stats onStatsUpdate func(dt *DownTrack, stat *livekit.AnalyticsStat) + + // when max subscribed layer changes + onMaxLayerChanged func(dt *DownTrack, layer int32) } // NewDownTrack returns a DownTrack. @@ -336,6 +339,10 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { f(d, pktSize) } + if tp.isSwitchingToMaxLayer && d.onMaxLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { + d.onMaxLayerChanged(d, layer) + } + d.UpdatePrimaryStats(uint32(pktSize)) } else { d.logger.Errorw("writing rtp packet err", err) @@ -369,7 +376,7 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int) int { // can be sent only on frame boundaries, writing on disabled tracks // will give more options. // LK-TODO-END - if d.forwarder.Muted() { + if d.forwarder.IsMuted() { return 0 } @@ -440,12 +447,25 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int) int { } // Mute enables or disables media forwarding -func (d *DownTrack) Mute(val bool) { - changed := d.forwarder.Mute(val) +func (d *DownTrack) Mute(muted bool) { + changed, maxLayers := d.forwarder.Mute(muted) if !changed { return } + if d.onMaxLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { + if muted { + d.onMaxLayerChanged(d, InvalidLayerSpatial) + } else { + // + // When unmuting, don't wait for layer lock as + // client might need to be notified to start layers + // before locking can happen in the forwarder. + // + d.onMaxLayerChanged(d, maxLayers.spatial) + } + } + if d.onSubscriptionChanged != nil { d.onSubscriptionChanged(d) } @@ -478,6 +498,10 @@ func (d *DownTrack) CloseWithFlush(flush bool) { d.connectionStats.Close() + if d.onMaxLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { + d.onMaxLayerChanged(d, InvalidLayerSpatial) + } + if d.onCloseHandler != nil { d.onCloseHandler() } @@ -485,18 +509,29 @@ func (d *DownTrack) CloseWithFlush(flush bool) { } func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32) { - changed, maxLayers := d.forwarder.SetMaxSpatialLayer(spatialLayer) + changed, maxLayers, currentLayers := d.forwarder.SetMaxSpatialLayer(spatialLayer) if !changed { return } + if d.onMaxLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo && maxLayers.SpatialGreaterThanOrEqual(currentLayers) { + // + // Notify when new max is + // 1. Equal to current -> already locked to the new max + // 2. Greater than current -> two scenarios + // a. is higher than previous max -> client may need to start higher layer before forwarder can lock + // b. is lower than previous max -> client can stop higher layer(s) + // + d.onMaxLayerChanged(d, maxLayers.spatial) + } + if d.onSubscribedLayersChanged != nil { d.onSubscribedLayersChanged(d, maxLayers) } } func (d *DownTrack) SetMaxTemporalLayer(temporalLayer int32) { - changed, maxLayers := d.forwarder.SetMaxTemporalLayer(temporalLayer) + changed, maxLayers, _ := d.forwarder.SetMaxTemporalLayer(temporalLayer) if !changed { return } @@ -570,6 +605,15 @@ func (d *DownTrack) OnStatsUpdate(fn func(dt *DownTrack, stat *livekit.Analytics d.onStatsUpdate = fn } +func (d *DownTrack) OnMaxLayerChanged(fn func(dt *DownTrack, layer int32)) { + d.onMaxLayerChanged = fn + + // have to send this immediately to set initial values + if fn != nil && d.kind == webrtc.RTPCodecTypeVideo { + go fn(d, d.forwarder.MaxLayers().spatial) + } +} + func (d *DownTrack) IsDeficient() bool { return d.forwarder.IsDeficient() } @@ -1041,7 +1085,7 @@ func (d *DownTrack) DebugInfo() map[string]interface{} { "SSRC": d.ssrc, "MimeType": d.codec.MimeType, "Bound": d.bound.get(), - "Muted": d.forwarder.Muted(), + "Muted": d.forwarder.IsMuted(), "CurrentSpatialLayer": d.forwarder.CurrentLayers().spatial, "Stats": stats, } diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index ce92e9d65..b107813c6 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -117,11 +117,12 @@ const ( ) type TranslationParams struct { - shouldDrop bool - isDroppingRelevant bool - shouldSendPLI bool - rtp *TranslationParamsRTP - vp8 *TranslationParamsVP8 + shouldDrop bool + isDroppingRelevant bool + shouldSendPLI bool + isSwitchingToMaxLayer bool + rtp *TranslationParamsRTP + vp8 *TranslationParamsVP8 } type VideoLayers struct { @@ -137,6 +138,10 @@ func (v VideoLayers) GreaterThan(v2 VideoLayers) bool { return v.spatial > v2.spatial || (v.spatial == v2.spatial && v.temporal > v2.temporal) } +func (v VideoLayers) SpatialGreaterThanOrEqual(v2 VideoLayers) bool { + return v.spatial >= v2.spatial +} + const ( InvalidLayerSpatial = int32(-1) InvalidLayerTemporal = int32(-1) @@ -211,49 +216,49 @@ func NewForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType, log return f } -func (f *Forwarder) Mute(val bool) bool { +func (f *Forwarder) Mute(val bool) (bool, VideoLayers) { f.lock.Lock() defer f.lock.Unlock() if f.muted == val { - return false + return false, f.maxLayers } f.muted = val - return true + return true, f.maxLayers } -func (f *Forwarder) Muted() bool { +func (f *Forwarder) IsMuted() bool { f.lock.RLock() defer f.lock.RUnlock() return f.muted } -func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, VideoLayers) { +func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, VideoLayers, VideoLayers) { f.lock.Lock() defer f.lock.Unlock() if f.kind == webrtc.RTPCodecTypeAudio || spatialLayer == f.maxLayers.spatial { - return false, InvalidLayers + return false, f.maxLayers, f.currentLayers } f.maxLayers.spatial = spatialLayer - return true, f.maxLayers + return true, f.maxLayers, f.currentLayers } -func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, VideoLayers) { +func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, VideoLayers, VideoLayers) { f.lock.Lock() defer f.lock.Unlock() if f.kind == webrtc.RTPCodecTypeAudio || temporalLayer == f.maxLayers.temporal { - return false, InvalidLayers + return false, f.maxLayers, f.currentLayers } f.maxLayers.temporal = temporalLayer - return true, f.maxLayers + return true, f.maxLayers, f.currentLayers } func (f *Forwarder) MaxLayers() VideoLayers { @@ -1118,6 +1123,9 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in if extPkt.KeyFrame { // lock to target layer f.currentLayers.spatial = f.targetLayers.spatial + if f.currentLayers.spatial == f.maxLayers.spatial { + tp.isSwitchingToMaxLayer = true + } } else { tp.shouldSendPLI = true } diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 5314fe972..11b4bae18 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -24,13 +24,16 @@ func newForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType) *Fo func TestForwarderMute(t *testing.T) { f := newForwarder(testutils.TestOpusCodec, webrtc.RTPCodecTypeAudio) - require.False(t, f.Muted()) - require.False(t, f.Mute(false)) // no change in mute state - require.False(t, f.Muted()) - require.True(t, f.Mute(true)) - require.True(t, f.Muted()) - require.True(t, f.Mute(false)) - require.False(t, f.Muted()) + require.False(t, f.IsMuted()) + muted, _ := f.Mute(false) + require.False(t, muted) // no change in mute state + require.False(t, f.IsMuted()) + muted, _ = f.Mute(true) + require.True(t, muted) + require.True(t, f.IsMuted()) + muted, _ = f.Mute(false) + require.True(t, muted) + require.False(t, f.IsMuted()) } func TestForwarderLayersAudio(t *testing.T) { @@ -41,13 +44,15 @@ func TestForwarderLayersAudio(t *testing.T) { require.Equal(t, InvalidLayers, f.CurrentLayers()) require.Equal(t, InvalidLayers, f.TargetLayers()) - changed, layers := f.SetMaxSpatialLayer(1) + changed, maxLayers, currentLayers := f.SetMaxSpatialLayer(1) require.False(t, changed) - require.Equal(t, InvalidLayers, layers) + require.Equal(t, InvalidLayers, maxLayers) + require.Equal(t, InvalidLayers, currentLayers) - changed, layers = f.SetMaxTemporalLayer(1) + changed, maxLayers, currentLayers = f.SetMaxTemporalLayer(1) require.False(t, changed) - require.Equal(t, InvalidLayers, layers) + require.Equal(t, InvalidLayers, maxLayers) + require.Equal(t, InvalidLayers, currentLayers) require.Equal(t, InvalidLayers, f.MaxLayers()) } @@ -65,31 +70,46 @@ func TestForwarderLayersVideo(t *testing.T) { require.Equal(t, InvalidLayers, f.CurrentLayers()) require.Equal(t, InvalidLayers, f.TargetLayers()) - changed, layers := f.SetMaxSpatialLayer(DefaultMaxLayerSpatial) + changed, maxLayers, currentLayers := f.SetMaxSpatialLayer(DefaultMaxLayerSpatial) require.False(t, changed) - require.Equal(t, InvalidLayers, layers) + require.Equal(t, expectedLayers, maxLayers) + require.Equal(t, InvalidLayers, currentLayers) - changed, layers = f.SetMaxSpatialLayer(DefaultMaxLayerSpatial - 1) + changed, maxLayers, currentLayers = f.SetMaxSpatialLayer(DefaultMaxLayerSpatial - 1) require.True(t, changed) expectedLayers = VideoLayers{ spatial: DefaultMaxLayerSpatial - 1, temporal: DefaultMaxLayerTemporal, } - require.Equal(t, expectedLayers, layers) + require.Equal(t, expectedLayers, maxLayers) require.Equal(t, expectedLayers, f.MaxLayers()) + require.Equal(t, InvalidLayers, currentLayers) - changed, layers = f.SetMaxTemporalLayer(DefaultMaxLayerTemporal) + f.currentLayers = VideoLayers{spatial: 0, temporal: 1} + changed, maxLayers, currentLayers = f.SetMaxSpatialLayer(DefaultMaxLayerSpatial - 1) require.False(t, changed) - require.Equal(t, InvalidLayers, layers) + expectedLayers = VideoLayers{ + spatial: DefaultMaxLayerSpatial - 1, + temporal: DefaultMaxLayerTemporal, + } + require.Equal(t, expectedLayers, maxLayers) + require.Equal(t, expectedLayers, f.MaxLayers()) + require.Equal(t, VideoLayers{spatial: 0, temporal: 1}, currentLayers) - changed, layers = f.SetMaxTemporalLayer(DefaultMaxLayerTemporal - 1) + changed, maxLayers, currentLayers = f.SetMaxTemporalLayer(DefaultMaxLayerTemporal) + require.False(t, changed) + require.Equal(t, expectedLayers, maxLayers) + require.Equal(t, VideoLayers{spatial: 0, temporal: 1}, currentLayers) + + changed, maxLayers, currentLayers = f.SetMaxTemporalLayer(DefaultMaxLayerTemporal - 1) require.True(t, changed) expectedLayers = VideoLayers{ spatial: DefaultMaxLayerSpatial - 1, temporal: DefaultMaxLayerTemporal - 1, } - require.Equal(t, expectedLayers, layers) + require.Equal(t, expectedLayers, maxLayers) require.Equal(t, expectedLayers, f.MaxLayers()) + require.Equal(t, VideoLayers{spatial: 0, temporal: 1}, currentLayers) } func TestForwarderGetForwardingStatus(t *testing.T) {