diff --git a/go.mod b/go.mod index d6e1962b2..afee33447 100644 --- a/go.mod +++ b/go.mod @@ -13,7 +13,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.13.3-0.20220525093853-ff07c68fb2a6 + github.com/livekit/protocol v0.13.3-0.20220526091938-2ecd4805e7e2 github.com/mackerelio/go-osstat v0.2.1 github.com/magefile/mage v1.11.0 github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0 diff --git a/go.sum b/go.sum index f6c9d7efb..4029948a6 100644 --- a/go.sum +++ b/go.sum @@ -131,8 +131,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.13.3-0.20220525093853-ff07c68fb2a6 h1:9utQz3izM/EdeS6Iqp98cpM8C07/vsSP8ONzSWUWFYQ= -github.com/livekit/protocol v0.13.3-0.20220525093853-ff07c68fb2a6/go.mod h1:BLtSeVmn2rLP37xjzw7gHgaAmkWl3L/L9bPvgSbaOfo= +github.com/livekit/protocol v0.13.3-0.20220526091938-2ecd4805e7e2 h1:ufUi4uR5OgfnrweGTWXkplxudLeZw9/gtkgws2oszAo= +github.com/livekit/protocol v0.13.3-0.20220526091938-2ecd4805e7e2/go.mod h1:BLtSeVmn2rLP37xjzw7gHgaAmkWl3L/L9bPvgSbaOfo= github.com/mackerelio/go-osstat v0.2.1 h1:5AeAcBEutEErAOlDz6WCkEvm6AKYgHTUQrfwm5RbeQc= github.com/mackerelio/go-osstat v0.2.1/go.mod h1:UzRL8dMCCTqG5WdRtsxbuljMpZt9PCAGXqxPst5QtaY= github.com/magefile/mage v1.11.0 h1:C/55Ywp9BpgVVclD3lRnSYCwXTYxmSppIgLeDYlNuls= diff --git a/pkg/config/config.go b/pkg/config/config.go index c3efcc4a0..551328aa6 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -222,8 +222,8 @@ func NewConfig(confString string, c *cli.Context) (*Config, error) { {Mime: webrtc.MimeTypeOpus}, {Mime: webrtc.MimeTypeVP8}, {Mime: webrtc.MimeTypeH264}, - // {Mime: webrtc.MimeTypeAV1}, - // {Mime: webrtc.MimeTypeVP9}, + {Mime: webrtc.MimeTypeAV1}, + {Mime: webrtc.MimeTypeVP9}, }, EmptyTimeout: 5 * 60, }, diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 20617079f..ebb3bd7ac 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -2,12 +2,12 @@ package rtc import ( "context" + "strings" "sync" "github.com/pion/rtcp" "github.com/pion/webrtc/v3" "go.uber.org/atomic" - "google.golang.org/protobuf/proto" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" @@ -26,8 +26,6 @@ type MediaTrack struct { numUpTracks atomic.Uint32 buffer *buffer.Buffer - layerSSRCs [livekit.VideoQuality_HIGH + 1]uint32 - *MediaTrackReceiver lock sync.RWMutex @@ -49,6 +47,7 @@ type MediaTrackParams struct { VideoConfig config.VideoConfig Telemetry telemetry.TelemetryService Logger logger.Logger + SimTracks map[uint32]SimulcastTrackInfo } func NewMediaTrack(params MediaTrackParams) *MediaTrack { @@ -92,33 +91,35 @@ func (t *MediaTrack) SignalCid() string { return t.params.SignalCid } -func (t *MediaTrack) SdpCid() string { - return t.params.SdpCid +func (t *MediaTrack) HasSdpCid(cid string) bool { + if t.params.SdpCid == cid { + return true + } + + info := t.MediaTrackReceiver.TrackInfo(false) + t.params.Logger.Debugw("MediaTrack.HasSdpCid", "cid", cid, "trackInfo", info.String()) + for _, c := range info.Codecs { + if c.Cid == cid { + return true + } + } + return false } func (t *MediaTrack) ToProto() *livekit.TrackInfo { - info := proto.Clone(t.MediaTrackReceiver.TrackInfo()).(*livekit.TrackInfo) + info := t.MediaTrackReceiver.TrackInfo(true) info.Muted = t.IsMuted() info.Simulcast = t.IsSimulcast() - layers := t.MediaTrackReceiver.GetVideoLayers() - t.lock.RLock() - for _, layer := range layers { - if int(layer.Quality) < len(t.layerSSRCs) { - layer.Ssrc = t.layerSSRCs[layer.Quality] - } - } - t.lock.RUnlock() - info.Layers = layers - return info } -// AddReceiver adds a new RTP receiver to the track -func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.TrackRemote, twcc *twcc.Responder) { +// AddReceiver adds a new RTP receiver to the track, returns true when receiver represents a new codec +func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.TrackRemote, twcc *twcc.Responder, mid string) bool { + var newCodec bool buff, rtcpReader := t.params.BufferFactory.GetBufferPair(uint32(track.SSRC())) if buff == nil || rtcpReader == nil { t.params.Logger.Errorw("could not retrieve buffer pair", nil) - return + return newCodec } rtcpReader.OnPacket(func(bytes []byte) { @@ -138,86 +139,88 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra } }) - isNew := false t.lock.Lock() - if t.Receiver() == nil { - isNew = true - - wr := sfu.NewWebRTCReceiver( + mime := strings.ToLower(track.Codec().MimeType) + t.params.Logger.Debugw("AddReceiver", "mime", track.Codec().MimeType) + wr := t.MediaTrackReceiver.Receiver(mime) + if wr == nil { + var priority int + for idx, c := range t.params.TrackInfo.Codecs { + if strings.HasSuffix(mime, c.MimeType) { + priority = idx + break + } + } + newWR := sfu.NewWebRTCReceiver( receiver, track, t.PublisherID(), t.params.TrackInfo.Source, - t.params.Logger, + LoggerWithCodecMime(t.params.Logger, mime), twcc, sfu.WithPliThrottleConfig(t.params.PLIThrottleConfig), sfu.WithAudioConfig(t.params.AudioConfig), sfu.WithLoadBalanceThreshold(20), sfu.WithStreamTrackers(), ) - wr.SetRTCPCh(t.params.RTCPChan) - wr.OnCloseHandler(func() { + newWR.SetRTCPCh(t.params.RTCPChan) + newWR.OnCloseHandler(func() { t.RemoveAllSubscribers() - t.MediaTrackReceiver.Close() - t.MediaTrackReceiver.ClearReceiver() - t.params.Telemetry.TrackUnpublished( + t.MediaTrackReceiver.ClearReceiver(mime) + if t.MediaTrackReceiver.TryClose() { + t.params.Telemetry.TrackUnpublished( + context.Background(), + t.PublisherID(), + t.PublisherIdentity(), + t.ToProto(), + uint32(track.SSRC()), + ) + } + }) + newWR.OnStatsUpdate(func(_ *sfu.WebRTCReceiver, stat *livekit.AnalyticsStat) { + t.params.Telemetry.TrackStats(livekit.StreamType_UPSTREAM, t.PublisherID(), t.ID(), stat) + }) + if t.PrimaryReceiver() == nil { + t.params.Telemetry.TrackPublished( context.Background(), t.PublisherID(), t.PublisherIdentity(), t.ToProto(), - uint32(track.SSRC()), ) - }) - wr.OnStatsUpdate(func(_ *sfu.WebRTCReceiver, stat *livekit.AnalyticsStat) { - t.params.Telemetry.TrackStats(livekit.StreamType_UPSTREAM, t.PublisherID(), t.ID(), stat) - }) + } t.buffer = buff - t.MediaTrackReceiver.SetupReceiver(wr) + t.MediaTrackReceiver.SetupReceiver(newWR, priority, mid) + + for ssrc, info := range t.params.SimTracks { + if info.Mid == mid { + t.MediaTrackReceiver.SetLayerSsrc(mime, info.Rid, ssrc) + } + } + wr = newWR + newCodec = true } t.lock.Unlock() - t.Receiver().(*sfu.WebRTCReceiver).AddUpTrack(track, buff) + wr.(*sfu.WebRTCReceiver).AddUpTrack(track, buff) // LK-TODO: can remove this completely when VideoLayers protocol becomes the default as it has info from client or if we decide to use TrackInfo.Simulcast if t.numUpTracks.Inc() > 1 || track.RID() != "" { // cannot only rely on numUpTracks since we fire metadata events immediately after the first layer - t.MediaTrackReceiver.SetSimulcast(true) + t.SetSimulcast(true) } if t.IsSimulcast() { - layer := sfu.RidToLayer(track.RID()) - t.lock.Lock() - if int(layer) < len(t.layerSSRCs) { - t.layerSSRCs[layer] = uint32(track.SSRC()) - } - t.lock.Unlock() - } - - if isNew { - t.params.Telemetry.TrackPublished( - context.Background(), - t.PublisherID(), - t.PublisherIdentity(), - t.ToProto(), - ) + t.MediaTrackReceiver.SetLayerSsrc(mime, track.RID(), uint32(track.SSRC())) } buff.Bind(receiver.GetParameters(), track.Codec().RTPCodecCapability) -} - -func (t *MediaTrack) TrySetSimulcastSSRC(layer uint8, ssrc uint32) { - t.lock.Lock() - defer t.lock.Unlock() - - if int(layer) < len(t.layerSSRCs) && t.layerSSRCs[layer] == 0 { - t.layerSSRCs[layer] = ssrc - } + return newCodec } func (t *MediaTrack) GetConnectionScore() float32 { - receiver := t.Receiver() + receiver := t.PrimaryReceiver() if receiver == nil { return 0.0 } @@ -226,10 +229,9 @@ func (t *MediaTrack) GetConnectionScore() float32 { } func (t *MediaTrack) SetRTT(rtt uint32) { - receiver := t.Receiver() - if receiver == nil { - return - } - - receiver.(*sfu.WebRTCReceiver).SetRTT(rtt) + t.MediaTrackReceiver.SetRTT(rtt) +} + +func (t *MediaTrack) HasPendingCodec() bool { + return len(t.params.TrackInfo.Codecs) > len(t.Receivers()) } diff --git a/pkg/rtc/mediatrack_test.go b/pkg/rtc/mediatrack_test.go index 8e5906c35..eda544909 100644 --- a/pkg/rtc/mediatrack_test.go +++ b/pkg/rtc/mediatrack_test.go @@ -1,15 +1,18 @@ package rtc import ( + "sort" "sync" "testing" "time" + "github.com/pion/webrtc/v3" "github.com/stretchr/testify/require" "github.com/livekit/protocol/livekit" "github.com/livekit/livekit-server/pkg/config" + "github.com/livekit/livekit-server/pkg/rtc/types" ) func TestTrackInfo(t *testing.T) { @@ -105,51 +108,83 @@ func TestGetQualityForDimension(t *testing.T) { } func TestSubscribedMaxQuality(t *testing.T) { + subscribedCodecsAsString := func(c1 []*livekit.SubscribedCodec) string { + sort.Slice(c1, func(i, j int) bool { return c1[i].Codec < c1[j].Codec }) + var s1 string + for _, c := range c1 { + s1 += c.String() + } + return s1 + } t.Run("subscribers muted", func(t *testing.T) { - mt := NewMediaTrack(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 := NewMediaTrack(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.notifySubscriberMaxQuality("s1", livekit.VideoQuality_HIGH) - + }) + var lock sync.Mutex actualTrackID := livekit.TrackID("") - actualSubscribedQualities := make([]*livekit.SubscribedQuality, 0) - mt.OnSubscribedMaxQualityChange(func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, _maxSubscribedQuality livekit.VideoQuality) error { + actualSubscribedQualities := make([]*livekit.SubscribedCodec, 0) + mt.OnSubscribedMaxQualityChange(func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedCodec, _maxSubscribedQualities []types.SubscribedCodecQuality) error { + lock.Lock() actualTrackID = trackID actualSubscribedQualities = subscribedQualities + lock.Unlock() return nil }) - // mute all subscribers - mt.notifySubscriberMaxQuality("s1", livekit.VideoQuality_OFF) + mt.AddCodec(webrtc.MimeTypeVP8) + mt.AddCodec(webrtc.MimeTypeAV1) - expectedSubscribedQualities := []*livekit.SubscribedQuality{ - {Quality: livekit.VideoQuality_LOW, Enabled: false}, - {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, - {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + mt.notifySubscriberMaxQuality("s1", webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8}, livekit.VideoQuality_HIGH) + mt.notifySubscriberMaxQuality("s2", webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeAV1}, livekit.VideoQuality_HIGH) + + // mute all subscribers of vp8 + mt.notifySubscriberMaxQuality("s1", webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8}, livekit.VideoQuality_OFF) + + expectedSubscribedQualities := []*livekit.SubscribedCodec{ + { + Codec: webrtc.MimeTypeVP8, + Qualities: []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: false}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, + {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + }, + }, + { + Codec: webrtc.MimeTypeAV1, + Qualities: []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: true}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: true}, + {Quality: livekit.VideoQuality_HIGH, Enabled: true}, + }, + }, } + time.Sleep(10 * time.Millisecond) + lock.Lock() require.Equal(t, livekit.TrackID("v1"), actualTrackID) - require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) + require.EqualValues(t, subscribedCodecsAsString(expectedSubscribedQualities), subscribedCodecsAsString(actualSubscribedQualities)) + lock.Unlock() }) t.Run("subscribers max quality", func(t *testing.T) { @@ -182,12 +217,15 @@ func TestSubscribedMaxQuality(t *testing.T) { }, }) + mt.AddCodec(webrtc.MimeTypeVP8) + mt.AddCodec(webrtc.MimeTypeAV1) + lock := sync.RWMutex{} lock.Lock() actualTrackID := livekit.TrackID("") - actualSubscribedQualities := make([]*livekit.SubscribedQuality, 0) + actualSubscribedQualities := make([]*livekit.SubscribedCodec, 0) lock.Unlock() - mt.OnSubscribedMaxQualityChange(func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, _maxSubscribedQuality livekit.VideoQuality) error { + mt.OnSubscribedMaxQualityChange(func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedCodec, _maxSubscribedQualities []types.SubscribedCodecQuality) error { lock.Lock() actualTrackID = trackID actualSubscribedQualities = subscribedQualities @@ -195,105 +233,187 @@ func TestSubscribedMaxQuality(t *testing.T) { return nil }) - mt.maxSubscribedQuality = livekit.VideoQuality_LOW - mt.notifySubscriberMaxQuality("s1", livekit.VideoQuality_HIGH) - mt.notifySubscriberMaxQuality("s2", livekit.VideoQuality_MEDIUM) + mt.maxSubscribedQuality = map[string]livekit.VideoQuality{ + webrtc.MimeTypeVP8: livekit.VideoQuality_LOW, + webrtc.MimeTypeAV1: livekit.VideoQuality_LOW, + } + mt.notifySubscriberMaxQuality("s1", webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8}, livekit.VideoQuality_HIGH) + mt.notifySubscriberMaxQuality("s2", webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8}, livekit.VideoQuality_MEDIUM) + mt.notifySubscriberMaxQuality("s3", webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeAV1}, livekit.VideoQuality_MEDIUM) - expectedSubscribedQualities := []*livekit.SubscribedQuality{ - {Quality: livekit.VideoQuality_LOW, Enabled: true}, - {Quality: livekit.VideoQuality_MEDIUM, Enabled: true}, - {Quality: livekit.VideoQuality_HIGH, Enabled: true}, + expectedSubscribedQualities := []*livekit.SubscribedCodec{ + { + Codec: webrtc.MimeTypeVP8, + Qualities: []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: true}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: true}, + {Quality: livekit.VideoQuality_HIGH, Enabled: true}, + }, + }, + { + Codec: webrtc.MimeTypeAV1, + Qualities: []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: true}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: true}, + {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + }, + }, } lock.RLock() require.Equal(t, livekit.TrackID("v1"), actualTrackID) - require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) + require.EqualValues(t, subscribedCodecsAsString(expectedSubscribedQualities), subscribedCodecsAsString(actualSubscribedQualities)) lock.RUnlock() // "s1" dropping to MEDIUM should disable HIGH layer - mt.notifySubscriberMaxQuality("s1", livekit.VideoQuality_MEDIUM) + mt.notifySubscriberMaxQuality("s1", webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8}, livekit.VideoQuality_MEDIUM) // wait for throttle to kick in time.Sleep(110 * time.Millisecond) - expectedSubscribedQualities = []*livekit.SubscribedQuality{ - {Quality: livekit.VideoQuality_LOW, Enabled: true}, - {Quality: livekit.VideoQuality_MEDIUM, Enabled: true}, - {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + expectedSubscribedQualities = []*livekit.SubscribedCodec{ + { + Codec: webrtc.MimeTypeVP8, + Qualities: []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: true}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: true}, + {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + }, + }, + { + Codec: webrtc.MimeTypeAV1, + Qualities: []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: true}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: true}, + {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + }, + }, } lock.RLock() require.Equal(t, livekit.TrackID("v1"), actualTrackID) - require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) + require.EqualValues(t, subscribedCodecsAsString(expectedSubscribedQualities), subscribedCodecsAsString(actualSubscribedQualities)) lock.RUnlock() - // "s1" and "s1" dropping to LOW should disable HIGH & MEDIUM - mt.notifySubscriberMaxQuality("s1", livekit.VideoQuality_LOW) - mt.notifySubscriberMaxQuality("s2", livekit.VideoQuality_LOW) + // "s1" , "s2" , "s3" dropping to LOW should disable HIGH & MEDIUM + mt.notifySubscriberMaxQuality("s1", webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8}, livekit.VideoQuality_LOW) + mt.notifySubscriberMaxQuality("s2", webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8}, livekit.VideoQuality_LOW) + mt.notifySubscriberMaxQuality("s3", webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeAV1}, livekit.VideoQuality_LOW) // wait for throttle to kick in time.Sleep(110 * time.Millisecond) - expectedSubscribedQualities = []*livekit.SubscribedQuality{ - {Quality: livekit.VideoQuality_LOW, Enabled: true}, - {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, - {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + expectedSubscribedQualities = []*livekit.SubscribedCodec{ + { + Codec: webrtc.MimeTypeVP8, + Qualities: []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: true}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, + {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + }, + }, + { + Codec: webrtc.MimeTypeAV1, + Qualities: []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: true}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, + {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + }, + }, } lock.RLock() require.Equal(t, livekit.TrackID("v1"), actualTrackID) - require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) + require.EqualValues(t, subscribedCodecsAsString(expectedSubscribedQualities), subscribedCodecsAsString(actualSubscribedQualities)) lock.RUnlock() - // muting "s2" only should not disable all qualities - mt.notifySubscriberMaxQuality("s2", livekit.VideoQuality_OFF) + // muting "s2" only should not disable all qualities of vp8, no change of expected qualities + mt.notifySubscriberMaxQuality("s2", webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8}, livekit.VideoQuality_OFF) - expectedSubscribedQualities = []*livekit.SubscribedQuality{ - {Quality: livekit.VideoQuality_LOW, Enabled: true}, - {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, - {Quality: livekit.VideoQuality_HIGH, Enabled: false}, - } lock.RLock() require.Equal(t, livekit.TrackID("v1"), actualTrackID) - require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) + require.EqualValues(t, subscribedCodecsAsString(expectedSubscribedQualities), subscribedCodecsAsString(actualSubscribedQualities)) lock.RUnlock() - // muting "s1" also should disable all qualities - mt.notifySubscriberMaxQuality("s1", livekit.VideoQuality_OFF) + // muting "s1" and s3 also should disable all qualities + mt.notifySubscriberMaxQuality("s1", webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8}, livekit.VideoQuality_OFF) + mt.notifySubscriberMaxQuality("s3", webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeAV1}, livekit.VideoQuality_OFF) time.Sleep(110 * time.Millisecond) - expectedSubscribedQualities = []*livekit.SubscribedQuality{ - {Quality: livekit.VideoQuality_LOW, Enabled: false}, - {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, - {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + expectedSubscribedQualities = []*livekit.SubscribedCodec{ + { + Codec: webrtc.MimeTypeVP8, + Qualities: []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: false}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, + {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + }, + }, + { + Codec: webrtc.MimeTypeAV1, + Qualities: []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: false}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, + {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + }, + }, } lock.RLock() require.Equal(t, livekit.TrackID("v1"), actualTrackID) - require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) + require.EqualValues(t, subscribedCodecsAsString(expectedSubscribedQualities), subscribedCodecsAsString(actualSubscribedQualities)) lock.RUnlock() - // unmuting "s1" should enable previously set max quality - mt.notifySubscriberMaxQuality("s1", livekit.VideoQuality_LOW) + // unmuting "s1" should enable vp8 previously set max quality + mt.notifySubscriberMaxQuality("s1", webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8}, livekit.VideoQuality_LOW) time.Sleep(110 * time.Millisecond) - expectedSubscribedQualities = []*livekit.SubscribedQuality{ - {Quality: livekit.VideoQuality_LOW, Enabled: true}, - {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, - {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + expectedSubscribedQualities = []*livekit.SubscribedCodec{ + { + Codec: webrtc.MimeTypeVP8, + Qualities: []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: true}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, + {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + }, + }, + { + Codec: webrtc.MimeTypeAV1, + Qualities: []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: false}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, + {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + }, + }, } lock.RLock() require.Equal(t, livekit.TrackID("v1"), actualTrackID) - require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) + require.EqualValues(t, subscribedCodecsAsString(expectedSubscribedQualities), subscribedCodecsAsString(actualSubscribedQualities)) lock.RUnlock() // a higher quality from a different node should trigger that quality - mt.NotifySubscriberNodeMaxQuality("n1", livekit.VideoQuality_HIGH) + mt.NotifySubscriberNodeMaxQuality("n1", []types.SubscribedCodecQuality{ + {CodecMime: webrtc.MimeTypeVP8, Quality: livekit.VideoQuality_HIGH}, + {CodecMime: webrtc.MimeTypeAV1, Quality: livekit.VideoQuality_MEDIUM}, + }) - expectedSubscribedQualities = []*livekit.SubscribedQuality{ - {Quality: livekit.VideoQuality_LOW, Enabled: true}, - {Quality: livekit.VideoQuality_MEDIUM, Enabled: true}, - {Quality: livekit.VideoQuality_HIGH, Enabled: true}, + expectedSubscribedQualities = []*livekit.SubscribedCodec{ + { + Codec: webrtc.MimeTypeVP8, + Qualities: []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: true}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: true}, + {Quality: livekit.VideoQuality_HIGH, Enabled: true}, + }, + }, + { + Codec: webrtc.MimeTypeAV1, + Qualities: []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: true}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: true}, + {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + }, + }, } lock.RLock() require.Equal(t, livekit.TrackID("v1"), actualTrackID) - require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities) + require.EqualValues(t, subscribedCodecsAsString(expectedSubscribedQualities), subscribedCodecsAsString(actualSubscribedQualities)) lock.RUnlock() }) } diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index a1156f24d..fa3ccd52d 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -3,6 +3,7 @@ package rtc import ( "errors" "sort" + "strings" "sync" "time" @@ -25,14 +26,24 @@ const ( layerSelectionTolerance = 0.9 ) +type simulcastReceiver struct { + sfu.TrackReceiver + priority int + layerSSRCs [livekit.VideoQuality_HIGH + 1]uint32 +} + +func (r *simulcastReceiver) Priority() int { + return r.priority +} + type MediaTrackReceiver struct { params MediaTrackReceiverParams muted atomic.Bool simulcasted atomic.Bool lock sync.RWMutex + receivers []*simulcastReceiver trackInfo *livekit.TrackInfo - receiver sfu.TrackReceiver layerDimensions map[livekit.VideoQuality]*livekit.VideoLayer // track audio fraction lost @@ -91,28 +102,83 @@ func NewMediaTrackReceiver(params MediaTrackReceiverParams) *MediaTrackReceiver func (t *MediaTrackReceiver) Restart() { t.lock.Lock() - receiver := t.receiver + receivers := t.receivers t.lock.Unlock() - if receiver != nil { + for _, receiver := range receivers { receiver.SetMaxExpectedSpatialLayer(SpatialLayerForQuality(livekit.VideoQuality_HIGH)) - t.MediaTrackSubscriptions.Restart() } + + t.MediaTrackSubscriptions.Restart() } -func (t *MediaTrackReceiver) SetupReceiver(receiver sfu.TrackReceiver) { +func (t *MediaTrackReceiver) SetupReceiver(receiver sfu.TrackReceiver, priority int, mid string) { t.lock.Lock() - t.receiver = receiver + t.receivers = append(t.receivers, &simulcastReceiver{TrackReceiver: receiver, priority: priority}) + sort.Slice(t.receivers, func(i, j int) bool { + return t.receivers[i].Priority() < t.receivers[j].Priority() + }) + + if mid != "" { + if priority == 0 { + t.trackInfo.MimeType = receiver.Codec().MimeType + t.trackInfo.Mid = mid + + // for clients don't have simulcast codecs (old version or single codec), add the primary codec + if len(t.trackInfo.Codecs) == 0 && t.trackInfo.Type == livekit.TrackType_VIDEO { + t.trackInfo.Codecs = append(t.trackInfo.Codecs, &livekit.SimulcastCodecInfo{}) + } + } + + for i, ci := range t.trackInfo.Codecs { + if i == priority { + ci.Mid = mid + ci.MimeType = receiver.Codec().MimeType + } + } + } + t.lock.Unlock() + t.params.Logger.Debugw("setup receiver", "mime", receiver.Codec().MimeType, "priority", priority, "receivers", t.receivers) + t.MediaTrackSubscriptions.AddCodec(receiver.Codec().MimeType) t.MediaTrackSubscriptions.Start() } -func (t *MediaTrackReceiver) ClearReceiver() { +func (t *MediaTrackReceiver) SetLayerSsrc(mime string, rid string, ssrc uint32) { t.lock.Lock() - t.receiver = nil + defer t.lock.Unlock() + + layer := sfu.RidToLayer(rid) + for _, receiver := range t.receivers { + if strings.EqualFold(receiver.Codec().MimeType, mime) && int(layer) < len(receiver.layerSSRCs) { + receiver.layerSSRCs[layer] = ssrc + return + } + } +} + +func (t *MediaTrackReceiver) ClearReceiver(mime string) { + t.lock.Lock() + for idx, receiver := range t.receivers { + if strings.EqualFold(receiver.Codec().MimeType, mime) { + t.receivers[idx] = t.receivers[len(t.receivers)-1] + t.receivers = t.receivers[:len(t.receivers)-1] + break + } + } + closeSubscription := len(t.receivers) == 0 t.lock.Unlock() + if closeSubscription { + t.MediaTrackSubscriptions.Close() + } +} + +func (t *MediaTrackReceiver) ClearAllReceivers() { + t.lock.Lock() + t.receivers = t.receivers[:0] + t.lock.Unlock() t.MediaTrackSubscriptions.Close() } @@ -124,8 +190,12 @@ func (t *MediaTrackReceiver) OnVideoLayerUpdate(f func(layers []*livekit.VideoLa t.onVideoLayerUpdate = f } -func (t *MediaTrackReceiver) Close() { +func (t *MediaTrackReceiver) TryClose() bool { t.lock.Lock() + if len(t.receivers) > 0 { + t.lock.Unlock() + return false + } onclose := t.onClose t.lock.Unlock() @@ -134,6 +204,7 @@ func (t *MediaTrackReceiver) Close() { for _, f := range onclose { f() } + return true } func (t *MediaTrackReceiver) ID() livekit.TrackID { @@ -187,10 +258,11 @@ func (t *MediaTrackReceiver) IsMuted() bool { func (t *MediaTrackReceiver) SetMuted(muted bool) { t.muted.Store(muted) - receiver := t.Receiver() - if receiver != nil { + t.lock.RLock() + for _, receiver := range t.receivers { receiver.SetUpTrackPaused(muted) } + t.lock.RUnlock() t.MediaTrackSubscriptions.SetMuted(muted) } @@ -207,8 +279,11 @@ func (t *MediaTrackReceiver) AddOnClose(f func()) { // AddSubscriber subscribes sub to current mediaTrack func (t *MediaTrackReceiver) AddSubscriber(sub types.LocalParticipant) error { - receiver := t.Receiver() - if receiver == nil { + t.lock.RLock() + receivers := t.receivers + t.lock.RUnlock() + + if len(receivers) == 0 { // cannot add, no receiver return errors.New("cannot subscribe without a receiver in place") } @@ -221,7 +296,7 @@ func (t *MediaTrackReceiver) AddSubscriber(sub types.LocalParticipant) error { streamId = PackStreamID(t.PublisherID(), t.ID()) } - downTrack, err := t.MediaTrackSubscriptions.AddSubscriber(sub, receiver.Codec(), NewWrappedReceiver(receiver, t.ID(), streamId)) + downTrack, err := t.MediaTrackSubscriptions.AddSubscriber(sub, NewWrappedReceiver(receivers, t.ID(), streamId)) if err != nil { return err } @@ -231,9 +306,6 @@ func (t *MediaTrackReceiver) AddSubscriber(sub types.LocalParticipant) error { downTrack.AddReceiverReportListener(t.handleMaxLossFeedback) } - if err = receiver.AddDownTrack(downTrack); err != nil { - logger.Errorw("could not add down track", err, "participant", sub.Identity(), "pID", sub.ID()) - } } return nil } @@ -241,6 +313,7 @@ func (t *MediaTrackReceiver) AddSubscriber(sub types.LocalParticipant) error { func (t *MediaTrackReceiver) UpdateTrackInfo(ti *livekit.TrackInfo) { t.lock.Lock() t.params.TrackInfo = ti + t.trackInfo = proto.Clone(ti).(*livekit.TrackInfo) t.lock.Unlock() if ti != nil && t.Kind() == livekit.TrackType_VIDEO { @@ -248,11 +321,58 @@ func (t *MediaTrackReceiver) UpdateTrackInfo(ti *livekit.TrackInfo) { } } -func (t *MediaTrackReceiver) TrackInfo() *livekit.TrackInfo { +func (t *MediaTrackReceiver) TrackInfo(generateLayer bool) *livekit.TrackInfo { t.lock.RLock() - defer t.lock.RUnlock() + ti := proto.Clone(t.trackInfo).(*livekit.TrackInfo) + t.lock.RUnlock() + if !generateLayer { + return ti + } + layers := t.GetVideoLayers() - return proto.Clone(t.params.TrackInfo).(*livekit.TrackInfo) + // set video layer ssrc info + for i, ci := range ti.Codecs { + for _, receiver := range t.receivers { + if receiver.priority == i { + originLayers := ci.Layers + ci.Layers = []*livekit.VideoLayer{} + for layerIdx, layer := range layers { + ci.Layers = append(ci.Layers, proto.Clone(layer).(*livekit.VideoLayer)) + + // if origin layer has ssrc, don't override it + if layerIdx < len(originLayers) && originLayers[layerIdx].Ssrc != 0 { + ci.Layers[layerIdx].Ssrc = originLayers[layerIdx].Ssrc + } else if int(layer.Quality) < len(receiver.layerSSRCs) { + ci.Layers[layerIdx].Ssrc = receiver.layerSSRCs[layer.Quality] + } + } + + if i == 0 { + ti.Layers = ci.Layers + } + break + } + } + } + + // for client don't use simulcast codecs (old client version or single codec) + if len(ti.Codecs) == 0 && len(t.receivers) > 0 { + receiver := t.receivers[0] + originLayers := ti.Layers + ti.Layers = []*livekit.VideoLayer{} + for layerIdx, layer := range layers { + ti.Layers = append(ti.Layers, proto.Clone(layer).(*livekit.VideoLayer)) + + // if origin layer has ssrc, don't override it + if layerIdx < len(originLayers) && originLayers[layerIdx].Ssrc != 0 { + ti.Layers[layerIdx].Ssrc = originLayers[layerIdx].Ssrc + } else if int(layer.Quality) < len(receiver.layerSSRCs) { + ti.Layers[layerIdx].Ssrc = receiver.layerSSRCs[layer.Quality] + } + } + } + + return ti } func (t *MediaTrackReceiver) UpdateVideoLayers(layers []*livekit.VideoLayer) { @@ -331,7 +451,7 @@ func (t *MediaTrackReceiver) GetQualityForDimension(width, height uint32) liveki } func (t *MediaTrackReceiver) GetAudioLevel() (float64, bool) { - receiver := t.Receiver() + receiver := t.PrimaryReceiver() if receiver == nil { return 0, false } @@ -394,31 +514,66 @@ func (t *MediaTrackReceiver) DebugInfo() map[string]interface{} { info["DownTracks"] = t.MediaTrackSubscriptions.DebugInfo() - receiver := t.Receiver() - if receiver != nil { - receiverInfo := receiver.DebugInfo() - for k, v := range receiverInfo { - info[k] = v - } + t.lock.RLock() + for _, receiver := range t.receivers { + info[receiver.Codec().MimeType] = receiver.DebugInfo() } + t.lock.RUnlock() return info } -func (t *MediaTrackReceiver) Receiver() sfu.TrackReceiver { +func (t *MediaTrackReceiver) PrimaryReceiver() sfu.TrackReceiver { t.lock.RLock() defer t.lock.RUnlock() - return t.receiver + if len(t.receivers) == 0 { + return nil + } + return t.receivers[0].TrackReceiver } -func (t *MediaTrackReceiver) OnSubscribedMaxQualityChange(f func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxSubscribedQuality livekit.VideoQuality) error) { - t.MediaTrackSubscriptions.OnSubscribedMaxQualityChange(func(subscribedQualities []*livekit.SubscribedQuality, maxSubscribedQuality livekit.VideoQuality) { - if f != nil && !t.IsMuted() { - _ = f(t.ID(), subscribedQualities, maxSubscribedQuality) +func (t *MediaTrackReceiver) Receiver(mime string) sfu.TrackReceiver { + t.lock.RLock() + defer t.lock.RUnlock() + + for _, r := range t.receivers { + if strings.EqualFold(r.Codec().MimeType, mime) { + return r.TrackReceiver } - receiver := t.Receiver() - if receiver != nil { - receiver.SetMaxExpectedSpatialLayer(SpatialLayerForQuality(maxSubscribedQuality)) + } + return nil +} + +func (t *MediaTrackReceiver) Receivers() []sfu.TrackReceiver { + t.lock.RLock() + defer t.lock.RUnlock() + + receivers := make([]sfu.TrackReceiver, 0, len(t.receivers)) + for _, r := range t.receivers { + receivers = append(receivers, r.TrackReceiver) + } + return receivers +} + +func (t *MediaTrackReceiver) SetRTT(rtt uint32) { + t.lock.Lock() + defer t.lock.Unlock() + + for _, r := range t.receivers { + r.TrackReceiver.(*sfu.WebRTCReceiver).SetRTT(rtt) + } +} + +func (t *MediaTrackReceiver) OnSubscribedMaxQualityChange(f func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedCodec, maxSubscribedQualities []types.SubscribedCodecQuality) error) { + t.MediaTrackSubscriptions.OnSubscribedMaxQualityChange(func(subscribedQualities []*livekit.SubscribedCodec, maxSubscribedQualities []types.SubscribedCodecQuality) { + if f != nil && !t.IsMuted() { + _ = f(t.ID(), subscribedQualities, maxSubscribedQualities) + } + for _, q := range maxSubscribedQualities { + receiver := t.Receiver(q.CodecMime) + if receiver != nil { + receiver.SetMaxExpectedSpatialLayer(SpatialLayerForQuality(q.Quality)) + } } }) } diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index 2db9438fb..f82eb3f3f 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -38,11 +38,11 @@ type MediaTrackSubscriptions struct { // quality level enable/disable maxQualityLock sync.RWMutex - maxSubscriberQuality map[livekit.ParticipantID]livekit.VideoQuality - maxSubscriberNodeQuality map[livekit.NodeID]livekit.VideoQuality - maxSubscribedQuality livekit.VideoQuality + maxSubscriberQuality map[livekit.ParticipantID]*types.SubscribedCodecQuality + maxSubscriberNodeQuality map[livekit.NodeID][]types.SubscribedCodecQuality + maxSubscribedQuality map[string]livekit.VideoQuality // codec mime -> quality maxSubscribedQualityDebounce func(func()) - onSubscribedMaxQualityChange func(subscribedQualities []*livekit.SubscribedQuality, maxSubscribedQuality livekit.VideoQuality) + onSubscribedMaxQualityChange func(subscribedQualities []*livekit.SubscribedCodec, maxSubscribedQualities []types.SubscribedCodecQuality) maxQualityTimer *time.Timer } @@ -64,9 +64,9 @@ func NewMediaTrackSubscriptions(params MediaTrackSubscriptionsParams) *MediaTrac params: params, subscribedTracks: make(map[livekit.ParticipantID]types.SubscribedTrack), pendingClose: make(map[livekit.ParticipantID]types.SubscribedTrack), - maxSubscriberQuality: make(map[livekit.ParticipantID]livekit.VideoQuality), - maxSubscriberNodeQuality: make(map[livekit.NodeID]livekit.VideoQuality), - maxSubscribedQuality: livekit.VideoQuality_HIGH, + maxSubscriberQuality: make(map[livekit.ParticipantID]*types.SubscribedCodecQuality), + maxSubscriberNodeQuality: make(map[livekit.NodeID][]types.SubscribedCodecQuality), + maxSubscribedQuality: make(map[string]livekit.VideoQuality), maxSubscribedQualityDebounce: debounce.New(params.VideoConfig.DynacastPauseDelay), } @@ -104,8 +104,14 @@ func (t *MediaTrackSubscriptions) IsSubscriber(subID livekit.ParticipantID) bool return ok } +func (t *MediaTrackSubscriptions) AddCodec(mime string) { + t.subscribedTracksMu.Lock() + t.maxSubscribedQuality[mime] = livekit.VideoQuality_HIGH + t.subscribedTracksMu.Unlock() +} + // AddSubscriber subscribes sub to current mediaTrack -func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, codec webrtc.RTPCodecCapability, wr WrappedReceiver) (*sfu.DownTrack, error) { +func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr *WrappedReceiver) (*sfu.DownTrack, error) { trackID := t.params.MediaTrack.ID() subscriberID := sub.ID() @@ -130,14 +136,12 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, code case livekit.TrackType_VIDEO: rtcpFeedback = t.params.SubscriberConfig.RTCPFeedback.Video } + codecs := wr.Codecs() + for _, c := range codecs { + c.RTCPFeedback = rtcpFeedback + } downTrack, err := sfu.NewDownTrack( - webrtc.RTPCodecCapability{ - MimeType: codec.MimeType, - ClockRate: codec.ClockRate, - Channels: codec.Channels, - SDPFmtpLine: codec.SDPFmtpLine, - RTCPFeedback: rtcpFeedback, - }, + codecs, wr, t.params.BufferFactory, subscriberID, @@ -204,8 +208,15 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, code // when out track is bound, start loop to send reports downTrack.OnBind(func() { + wr.DetermineReceiver(downTrack.Codec()) + if err = wr.AddDownTrack(downTrack); err != nil { + logger.Errorw("could not add down track", err, "participant", sub.Identity(), "pID", sub.ID()) + } go subTrack.Bound() go t.sendDownTrackBindingReports(sub) + // initialize to default layer + t.notifySubscriberMaxQuality(subscriberID, downTrack.Codec(), livekit.VideoQuality_HIGH) + subTrack.SetPublisherMuted(t.params.MediaTrack.IsMuted()) }) downTrack.OnStatsUpdate(func(_ *sfu.DownTrack, stat *livekit.AnalyticsStat) { @@ -213,7 +224,10 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, code }) downTrack.OnMaxLayerChanged(func(dt *sfu.DownTrack, layer int32) { - go t.notifySubscriberMaxQuality(subscriberID, QualityForSpatialLayer(layer)) + if dt.Codec().MimeType == "" { + t.params.Logger.Errorw("codec mime type is empty", nil) + } + go t.notifySubscriberMaxQuality(subscriberID, dt.Codec(), QualityForSpatialLayer(layer)) }) downTrack.OnRttUpdate(func(_ *sfu.DownTrack, rtt uint32) { @@ -227,7 +241,6 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, code t.subscribedTracksMu.Lock() t.subscribedTracks[subscriberID] = subTrack t.subscribedTracksMu.Unlock() - subTrack.SetPublisherMuted(t.params.MediaTrack.IsMuted()) // since sub will lock, run it in a goroutine to avoid deadlocks go func() { @@ -235,8 +248,6 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, code sub.Negotiate() }() - // initialize to default layer - t.notifySubscriberMaxQuality(subscriberID, livekit.VideoQuality_HIGH) t.params.Telemetry.TrackSubscribed(context.Background(), subscriberID, t.params.MediaTrack.ToProto(), &livekit.ParticipantInfo{Sid: string(t.params.MediaTrack.PublisherID()), Identity: string(t.params.MediaTrack.PublisherIdentity())}) return downTrack, nil @@ -397,15 +408,20 @@ func (t *MediaTrackSubscriptions) DebugInfo() []map[string]interface{} { return subscribedTrackInfo } -func (t *MediaTrackSubscriptions) OnSubscribedMaxQualityChange(f func(subscribedQualities []*livekit.SubscribedQuality, maxSubscribedQuality livekit.VideoQuality)) { +func (t *MediaTrackSubscriptions) OnSubscribedMaxQualityChange(f func(subscribedQualities []*livekit.SubscribedCodec, maxSubscribedQualities []types.SubscribedCodecQuality)) { t.onSubscribedMaxQualityChange = f } -func (t *MediaTrackSubscriptions) notifySubscriberMaxQuality(subscriberID livekit.ParticipantID, quality livekit.VideoQuality) { +func (t *MediaTrackSubscriptions) notifySubscriberMaxQuality(subscriberID livekit.ParticipantID, codec webrtc.RTPCodecCapability, quality livekit.VideoQuality) { + t.params.Logger.Debugw("notifying subscriber max quality", "subscriberID", subscriberID, "codec", codec, "quality", quality) if t.params.MediaTrack.Kind() != livekit.TrackType_VIDEO { return } + if codec.MimeType == "" { + t.params.Logger.Errorw("codec mime type is empty", nil) + } + t.maxQualityLock.Lock() if quality == livekit.VideoQuality_OFF { _, ok := t.maxSubscriberQuality[subscriberID] @@ -417,40 +433,64 @@ func (t *MediaTrackSubscriptions) notifySubscriberMaxQuality(subscriberID liveki delete(t.maxSubscriberQuality, subscriberID) } else { maxQuality, ok := t.maxSubscriberQuality[subscriberID] - if ok && maxQuality == quality { - t.maxQualityLock.Unlock() - return + if ok { + if maxQuality.Quality == quality { + t.maxQualityLock.Unlock() + return + } + maxQuality.Quality = quality + } else { + t.maxSubscriberQuality[subscriberID] = &types.SubscribedCodecQuality{ + Quality: quality, + CodecMime: codec.MimeType, + } } - - t.maxSubscriberQuality[subscriberID] = quality } t.maxQualityLock.Unlock() t.UpdateQualityChange(false) } -func (t *MediaTrackSubscriptions) NotifySubscriberNodeMaxQuality(nodeID livekit.NodeID, quality livekit.VideoQuality) { +func (t *MediaTrackSubscriptions) NotifySubscriberNodeMaxQuality(nodeID livekit.NodeID, qualities []types.SubscribedCodecQuality) { if t.params.MediaTrack.Kind() != livekit.TrackType_VIDEO { return } + if len(qualities) == 1 && qualities[0].CodecMime == "" { + // for old version msg don't have codec mime, use first mime type + t.maxQualityLock.RLock() + for mime := range t.maxSubscribedQuality { + qualities[0].CodecMime = mime + break + } + t.maxQualityLock.RUnlock() + } + t.maxQualityLock.Lock() - if quality == livekit.VideoQuality_OFF { - _, ok := t.maxSubscriberNodeQuality[nodeID] - if !ok { + if len(qualities) == 0 { + if _, ok := t.maxSubscriberNodeQuality[nodeID]; !ok { t.maxQualityLock.Unlock() return } - delete(t.maxSubscriberNodeQuality, nodeID) } else { - maxQuality, ok := t.maxSubscriberNodeQuality[nodeID] - if ok && maxQuality == quality { - t.maxQualityLock.Unlock() - return - } + if maxQualities, ok := t.maxSubscriberNodeQuality[nodeID]; ok { + var matchCounter int + for _, quality := range qualities { + for _, maxQuality := range maxQualities { + if quality == maxQuality { + matchCounter++ + break + } + } + } - t.maxSubscriberNodeQuality[nodeID] = quality + if matchCounter == len(qualities) && matchCounter == len(maxQualities) { + t.maxQualityLock.Unlock() + return + } + } + t.maxSubscriberNodeQuality[nodeID] = qualities } t.maxQualityLock.Unlock() @@ -462,59 +502,120 @@ func (t *MediaTrackSubscriptions) UpdateQualityChange(force bool) { return } + t.params.Logger.Debugw("updating quality change", + "force", force, + "maxSubscriberQuality", t.maxSubscriberQuality, + "maxSubscriberNodeQuality", t.maxSubscriberNodeQuality, + "maxSubscribedQuality", t.maxSubscribedQuality) + + maxSubscribedQuality := make(map[string]livekit.VideoQuality, len(t.maxSubscribedQuality)) + var changed bool t.maxQualityLock.Lock() - maxSubscribedQuality := livekit.VideoQuality_OFF + // reset maxSubscribedQuality + for mime := range t.maxSubscribedQuality { + maxSubscribedQuality[mime] = livekit.VideoQuality_OFF + } + + // every codec mime type should be in maxSubscribedQuality since we + // call AddCodec when receiving a new codec receiver for _, subQuality := range t.maxSubscriberQuality { - if maxSubscribedQuality == livekit.VideoQuality_OFF || subQuality > maxSubscribedQuality { - maxSubscribedQuality = subQuality + if q, ok := maxSubscribedQuality[subQuality.CodecMime]; ok { + if q == livekit.VideoQuality_OFF || (subQuality.Quality != livekit.VideoQuality_OFF && subQuality.Quality > q) { + maxSubscribedQuality[subQuality.CodecMime] = subQuality.Quality + } + } + } + for _, subQualities := range t.maxSubscriberNodeQuality { + for _, subQuality := range subQualities { + if q, ok := maxSubscribedQuality[subQuality.CodecMime]; ok { + if q == livekit.VideoQuality_OFF || (subQuality.Quality != livekit.VideoQuality_OFF && subQuality.Quality > q) { + maxSubscribedQuality[subQuality.CodecMime] = subQuality.Quality + } + } } } - for _, subQuality := range t.maxSubscriberNodeQuality { - if maxSubscribedQuality == livekit.VideoQuality_OFF || subQuality > maxSubscribedQuality { - maxSubscribedQuality = subQuality + qualityDowngrades := make(map[string]livekit.VideoQuality, len(t.maxSubscribedQuality)) + noChangeCount := 0 + for mime, q := range maxSubscribedQuality { + if origin := t.maxSubscribedQuality[mime]; origin != q { + if q == livekit.VideoQuality_OFF || (origin != livekit.VideoQuality_OFF && origin > q) { + // quality downgrade (or become off), delay notify to publisher + qualityDowngrades[mime] = origin + if force { + t.maxSubscribedQuality[mime] = q + } + } else { + // quality upgrade, update immediately + t.maxSubscribedQuality[mime] = q + } + changed = true + } else { + noChangeCount++ } } + t.params.Logger.Debugw("updating quality change", + "changed", changed, + "maxSubscribedQuality", maxSubscribedQuality, + "t.maxSubscribedQuality", t.maxSubscribedQuality, + "comesDownQuality", qualityDowngrades) - if maxSubscribedQuality == t.maxSubscribedQuality && !force { + if !changed && !force { t.maxQualityLock.Unlock() return } - // if quality comes down(or become OFF), delay notify to publisher - if (t.maxSubscribedQuality != livekit.VideoQuality_OFF) && - (t.maxSubscribedQuality > maxSubscribedQuality || maxSubscribedQuality == livekit.VideoQuality_OFF) && - t.params.VideoConfig.DynacastPauseDelay > 0 && !force { - - t.params.Logger.Debugw("throttle quality change", "from", t.maxSubscribedQuality, "to", maxSubscribedQuality) - t.maxQualityLock.Unlock() + // if quality downgrade (or become OFF), delay notify to publisher if needed + if len(qualityDowngrades) > 0 && !force { t.maxSubscribedQualityDebounce(func() { t.UpdateQualityChange(true) }) - return + + // no quality upgrades + if len(qualityDowngrades)+noChangeCount == len(t.maxSubscribedQuality) { + t.maxQualityLock.Unlock() + return + } } - t.maxSubscribedQuality = maxSubscribedQuality + subscribedCodec := make([]*livekit.SubscribedCodec, 0, len(t.maxSubscribedQuality)) + maxSubscribedQualities := make([]types.SubscribedCodecQuality, 0, len(t.maxSubscribedQuality)) + for mime, maxQuality := range t.maxSubscribedQuality { + maxSubscribedQualities = append(maxSubscribedQualities, types.SubscribedCodecQuality{ + CodecMime: mime, + Quality: maxQuality, + }) - var subscribedQualities []*livekit.SubscribedQuality - if t.maxSubscribedQuality == livekit.VideoQuality_OFF { - subscribedQualities = []*livekit.SubscribedQuality{ - {Quality: livekit.VideoQuality_LOW, Enabled: false}, - {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, - {Quality: livekit.VideoQuality_HIGH, Enabled: false}, - } - } else { - for q := livekit.VideoQuality_LOW; q <= livekit.VideoQuality_HIGH; q++ { - subscribedQualities = append(subscribedQualities, &livekit.SubscribedQuality{ - Quality: q, - Enabled: q <= t.maxSubscribedQuality, + if maxQuality == livekit.VideoQuality_OFF { + subscribedCodec = append(subscribedCodec, &livekit.SubscribedCodec{ + Codec: mime, + Qualities: []*livekit.SubscribedQuality{ + {Quality: livekit.VideoQuality_LOW, Enabled: false}, + {Quality: livekit.VideoQuality_MEDIUM, Enabled: false}, + {Quality: livekit.VideoQuality_HIGH, Enabled: false}, + }, + }) + } else { + var subscribedQualities []*livekit.SubscribedQuality + for q := livekit.VideoQuality_LOW; q <= livekit.VideoQuality_HIGH; q++ { + subscribedQualities = append(subscribedQualities, &livekit.SubscribedQuality{ + Quality: q, + Enabled: q <= maxQuality, + }) + } + subscribedCodec = append(subscribedCodec, &livekit.SubscribedCodec{ + Codec: mime, + Qualities: subscribedQualities, }) } } t.maxQualityLock.Unlock() if t.onSubscribedMaxQualityChange != nil { - t.onSubscribedMaxQualityChange(subscribedQualities, maxSubscribedQuality) + t.params.Logger.Debugw("subscribedMaxQualityChange", + "subscribedCodec", subscribedCodec, + "maxSubscribedQualities", maxSubscribedQualities) + t.onSubscribedMaxQualityChange(subscribedCodec, maxSubscribedQualities) } } diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 7b616db9d..bab1aa124 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -546,7 +546,8 @@ func (p *ParticipantImpl) AddTrack(req *livekit.AddTrackRequest) { func (p *ParticipantImpl) SetMigrateInfo(previousAnswer *webrtc.SessionDescription, mediaTracks []*livekit.TrackPublishedResponse, dataChannels []*livekit.DataChannelInfo) { p.pendingTracksLock.Lock() for _, t := range mediaTracks { - p.pendingTracks[t.GetCid()] = &pendingTrackInfo{t.GetTrack(), true} + pendingInfo := &pendingTrackInfo{TrackInfo: t.GetTrack(), migrated: true} + p.pendingTracks[t.GetCid()] = pendingInfo } p.pendingDataChannels = dataChannels @@ -1018,8 +1019,14 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w "trackID", publishedTrack.ID(), "rid", track.RID(), "SSRC", track.SSRC()) + } else { + p.params.Logger.Warnw("webrtc Track published but can't find MediaTrack", nil, + "kind", track.Kind().String(), + "webrtcTrackID", track.ID(), + "rid", track.RID(), + "SSRC", track.SSRC()) } - if !isNewTrack && publishedTrack != nil && p.IsReady() && p.onTrackUpdated != nil { + if !isNewTrack && publishedTrack != nil && !publishedTrack.HasPendingCodec() && p.IsReady() && p.onTrackUpdated != nil { p.onTrackUpdated(p, publishedTrack) } } @@ -1298,31 +1305,40 @@ func (p *ParticipantImpl) onStreamStateChange(update *sfu.StreamStateUpdate) err }) } -func (p *ParticipantImpl) onSubscribedMaxQualityChange(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, maxSubscribedQuality livekit.VideoQuality) error { +func (p *ParticipantImpl) onSubscribedMaxQualityChange(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedCodec, maxSubscribedQualites []types.SubscribedCodecQuality) error { if len(subscribedQualities) == 0 { return nil } - subscribedQualityUpdate := &livekit.SubscribedQualityUpdate{ - TrackSid: string(trackID), - SubscribedQualities: subscribedQualities, + // normalize the codec name + for _, subscribedQuality := range subscribedQualities { + subscribedQuality.Codec = strings.ToLower(strings.TrimLeft(subscribedQuality.Codec, "video/")) } - p.params.Telemetry.TrackMaxSubscribedVideoQuality( - context.Background(), - p.ID(), - &livekit.TrackInfo{ - Sid: string(trackID), - Type: livekit.TrackType_VIDEO, - }, - maxSubscribedQuality, - ) + subscribedQualityUpdate := &livekit.SubscribedQualityUpdate{ + TrackSid: string(trackID), + SubscribedQualities: subscribedQualities[0].Qualities, // for compatible with old client + SubscribedCodecs: subscribedQualities, + } + + for _, maxSubscribedQuality := range maxSubscribedQualites { + p.params.Telemetry.TrackMaxSubscribedVideoQuality( + context.Background(), + p.ID(), + &livekit.TrackInfo{ + Sid: string(trackID), + Type: livekit.TrackType_VIDEO, + }, + maxSubscribedQuality.CodecMime, + maxSubscribedQuality.Quality, + ) + } p.params.Logger.Debugw( "sending max subscribed quality", "trackID", trackID, "qualities", subscribedQualities, - "max", maxSubscribedQuality, + "max", maxSubscribedQualites, ) return p.writeMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_SubscribedQualityUpdate{ @@ -1372,7 +1388,22 @@ func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit Source: req.Source, Layers: req.Layers, } - p.pendingTracks[req.Cid] = &pendingTrackInfo{TrackInfo: ti} + pendingInfo := &pendingTrackInfo{TrackInfo: ti} + for _, codec := range req.SimulcastCodecs { + mime := codec.Codec + if req.Type == livekit.TrackType_VIDEO && !strings.HasPrefix(mime, "video/") { + mime = "video/" + mime + } else if req.Type == livekit.TrackType_AUDIO && !strings.HasPrefix(mime, "audio/") { + mime = "audio/" + mime + } + ti.Codecs = append(ti.Codecs, &livekit.SimulcastCodecInfo{ + MimeType: string(mime), + Cid: codec.Cid, + }) + } + + p.pendingTracks[req.Cid] = pendingInfo + p.params.Logger.Debugw("pending track added", "track", ti.String(), "request", req.String()) return ti } @@ -1446,10 +1477,18 @@ func (p *ParticipantImpl) getDTX() bool { return false } -func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) (types.MediaTrack, bool) { +func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) (*MediaTrack, bool) { p.pendingTracksLock.Lock() newTrack := false + p.params.Logger.Debugw("media track received", "track", track.ID(), "kind", track.Kind()) + var mid string + for _, tr := range p.publisher.pc.GetTransceivers() { + if tr.Receiver() == rtpReceiver { + mid = tr.Mid() + break + } + } // use existing media track to handle simulcast mt, ok := p.getPublishedTrackBySdpCid(track.ID()).(*MediaTrack) if !ok { @@ -1459,17 +1498,6 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei return nil, false } - ti.MimeType = track.Codec().MimeType - - var mid string - for _, tr := range p.publisher.pc.GetTransceivers() { - if tr.Receiver() == rtpReceiver { - mid = tr.Mid() - break - } - } - ti.Mid = mid - mt = NewMediaTrack(MediaTrackParams{ TrackInfo: ti, SignalCid: signalCid, @@ -1485,14 +1513,9 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei Logger: LoggerWithTrack(p.params.Logger, livekit.TrackID(ti.Sid)), SubscriberConfig: p.params.Config.Subscriber, PLIThrottleConfig: p.params.PLIThrottleConfig, + SimTracks: p.params.SimTracks, }) - for ssrc, info := range p.params.SimTracks { - if info.Mid == mid { - mt.TrySetSimulcastSSRC(uint8(sfu.RidToLayer(info.Rid)), ssrc) - } - } - mt.OnSubscribedMaxQualityChange(p.onSubscribedMaxQualityChange) // add to published and clean up pending @@ -1511,9 +1534,7 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei } p.pendingTracksLock.Unlock() - mt.AddReceiver(rtpReceiver, track, p.twcc) - - if newTrack { + if mt.AddReceiver(rtpReceiver, track, p.twcc, mid) && !mt.HasPendingCodec() { p.handleTrackPublished(mt) } @@ -1550,19 +1571,37 @@ func (p *ParticipantImpl) onUpTrackManagerClose() { func (p *ParticipantImpl) getPendingTrack(clientId string, kind livekit.TrackType) (string, *livekit.TrackInfo) { signalCid := clientId trackInfo := p.pendingTracks[clientId] - if trackInfo == nil { - // - // If no match on client id, find first one matching type - // as MediaStreamTrack can change client id when transceiver - // is added to peer connection. - // + track_loop: for cid, ti := range p.pendingTracks { - if ti.Type == kind { + if cid == clientId { trackInfo = ti signalCid = cid break } + + for _, c := range ti.Codecs { + if c.Cid == clientId { + trackInfo = ti + signalCid = cid + break track_loop + } + } + } + + if trackInfo == nil { + // + // If no match on client id, find first one matching type + // as MediaStreamTrack can change client id when transceiver + // is added to peer connection. + // + for cid, ti := range p.pendingTracks { + if ti.Type == kind { + trackInfo = ti + signalCid = cid + break + } + } } } @@ -1587,7 +1626,8 @@ func (p *ParticipantImpl) getPublishedTrackBySignalCid(clientId string) types.Me func (p *ParticipantImpl) getPublishedTrackBySdpCid(clientId string) types.MediaTrack { for _, publishedTrack := range p.GetPublishedTracks() { - if publishedTrack.(types.LocalMediaTrack).SdpCid() == clientId { + if publishedTrack.(types.LocalMediaTrack).HasSdpCid(clientId) { + p.params.Logger.Debugw("found track by sdp cid", "sdpCid", clientId, "trackID", publishedTrack.ID()) return publishedTrack } } diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index cf4f4732c..e95d1a110 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -136,7 +136,7 @@ func TestTrackPublishing(t *testing.T) { sink := p.params.Sink.(*routingfakes.FakeMessageSink) track := &typesfakes.FakeLocalMediaTrack{} - track.SdpCidReturns("cid") + track.HasSdpCidCalls(func(s string) bool { return s == "cid" }) // directly add to publishedTracks without lock - for testing purpose only p.UpTrackManager.publishedTracks["cid"] = track diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index 3d6b0b62c..417dcfd19 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -44,9 +44,6 @@ func NewSubscribedTrack(params SubscribedTrackParams) *SubscribedTrack { debouncer: debounce.New(subscriptionDebounceInterval), } - if !s.params.AdaptiveStream { - s.params.DownTrack.SetMaxSpatialLayer(SpatialLayerForQuality(livekit.VideoQuality_HIGH)) - } return s } @@ -55,6 +52,9 @@ func (t *SubscribedTrack) OnBind(f func()) { } func (t *SubscribedTrack) Bound() { + if !t.params.AdaptiveStream { + t.params.DownTrack.SetMaxSpatialLayer(SpatialLayerForQuality(livekit.VideoQuality_HIGH)) + } if t.onBind != nil { t.onBind() } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index ddb24ef38..0d2d54850 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -36,6 +36,11 @@ const ( MigrateStateComplete ) +type SubscribedCodecQuality struct { + CodecMime string + Quality livekit.VideoQuality +} + func (m MigrateState) String() string { switch m { case MigrateStateInit: @@ -80,7 +85,7 @@ type Participant interface { resolverBySid func(participantID livekit.ParticipantID) LocalParticipant, ) error UpdateVideoLayers(updateVideoLayers *livekit.UpdateVideoLayers) error - UpdateSubscribedQuality(nodeID livekit.NodeID, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error + UpdateSubscribedQuality(nodeID livekit.NodeID, trackID livekit.TrackID, maxQualities []SubscribedCodecQuality) error UpdateMediaLoss(nodeID livekit.NodeID, trackID livekit.TrackID, fractionalLoss uint32) error DebugInfo() map[string]interface{} @@ -198,7 +203,6 @@ type MediaTrack interface { UpdateVideoLayers(layers []*livekit.VideoLayer) IsSimulcast() bool - Receiver() sfu.TrackReceiver Restart() // callbacks @@ -215,8 +219,10 @@ type MediaTrack interface { // returns quality information that's appropriate for width & height GetQualityForDimension(width, height uint32) livekit.VideoQuality - NotifySubscriberNodeMaxQuality(nodeID livekit.NodeID, quality livekit.VideoQuality) + NotifySubscriberNodeMaxQuality(nodeID livekit.NodeID, qualites []SubscribedCodecQuality) NotifySubscriberNodeMediaLoss(nodeID livekit.NodeID, fractionalLoss uint8) + + Receivers() []sfu.TrackReceiver } //counterfeiter:generate . LocalMediaTrack @@ -224,7 +230,7 @@ type LocalMediaTrack interface { MediaTrack SignalCid() string - SdpCid() string + HasSdpCid(cid string) bool GetAudioLevel() (level float64, active bool) GetConnectionScore() float32 diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index ec04cdb08..b040eeb1a 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -70,6 +70,17 @@ type FakeLocalMediaTrack struct { getQualityForDimensionReturnsOnCall map[int]struct { result1 livekit.VideoQuality } + HasSdpCidStub func(string) bool + hasSdpCidMutex sync.RWMutex + hasSdpCidArgsForCall []struct { + arg1 string + } + hasSdpCidReturns struct { + result1 bool + } + hasSdpCidReturnsOnCall map[int]struct { + result1 bool + } IDStub func() livekit.TrackID iDMutex sync.RWMutex iDArgsForCall []struct { @@ -131,11 +142,11 @@ type FakeLocalMediaTrack struct { nameReturnsOnCall map[int]struct { result1 string } - NotifySubscriberNodeMaxQualityStub func(livekit.NodeID, livekit.VideoQuality) + NotifySubscriberNodeMaxQualityStub func(livekit.NodeID, []types.SubscribedCodecQuality) notifySubscriberNodeMaxQualityMutex sync.RWMutex notifySubscriberNodeMaxQualityArgsForCall []struct { arg1 livekit.NodeID - arg2 livekit.VideoQuality + arg2 []types.SubscribedCodecQuality } NotifySubscriberNodeMediaLossStub func(livekit.NodeID, uint8) notifySubscriberNodeMediaLossMutex sync.RWMutex @@ -163,15 +174,15 @@ type FakeLocalMediaTrack struct { publisherIdentityReturnsOnCall map[int]struct { result1 livekit.ParticipantIdentity } - ReceiverStub func() sfu.TrackReceiver - receiverMutex sync.RWMutex - receiverArgsForCall []struct { + ReceiversStub func() []sfu.TrackReceiver + receiversMutex sync.RWMutex + receiversArgsForCall []struct { } - receiverReturns struct { - result1 sfu.TrackReceiver + receiversReturns struct { + result1 []sfu.TrackReceiver } - receiverReturnsOnCall map[int]struct { - result1 sfu.TrackReceiver + receiversReturnsOnCall map[int]struct { + result1 []sfu.TrackReceiver } RemoveAllSubscribersStub func() removeAllSubscribersMutex sync.RWMutex @@ -198,16 +209,6 @@ type FakeLocalMediaTrack struct { revokeDisallowedSubscribersReturnsOnCall map[int]struct { result1 []livekit.ParticipantIdentity } - SdpCidStub func() string - sdpCidMutex sync.RWMutex - sdpCidArgsForCall []struct { - } - sdpCidReturns struct { - result1 string - } - sdpCidReturnsOnCall map[int]struct { - result1 string - } SetMutedStub func(bool) setMutedMutex sync.RWMutex setMutedArgsForCall []struct { @@ -574,6 +575,67 @@ func (fake *FakeLocalMediaTrack) GetQualityForDimensionReturnsOnCall(i int, resu }{result1} } +func (fake *FakeLocalMediaTrack) HasSdpCid(arg1 string) bool { + fake.hasSdpCidMutex.Lock() + ret, specificReturn := fake.hasSdpCidReturnsOnCall[len(fake.hasSdpCidArgsForCall)] + fake.hasSdpCidArgsForCall = append(fake.hasSdpCidArgsForCall, struct { + arg1 string + }{arg1}) + stub := fake.HasSdpCidStub + fakeReturns := fake.hasSdpCidReturns + fake.recordInvocation("HasSdpCid", []interface{}{arg1}) + fake.hasSdpCidMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalMediaTrack) HasSdpCidCallCount() int { + fake.hasSdpCidMutex.RLock() + defer fake.hasSdpCidMutex.RUnlock() + return len(fake.hasSdpCidArgsForCall) +} + +func (fake *FakeLocalMediaTrack) HasSdpCidCalls(stub func(string) bool) { + fake.hasSdpCidMutex.Lock() + defer fake.hasSdpCidMutex.Unlock() + fake.HasSdpCidStub = stub +} + +func (fake *FakeLocalMediaTrack) HasSdpCidArgsForCall(i int) string { + fake.hasSdpCidMutex.RLock() + defer fake.hasSdpCidMutex.RUnlock() + argsForCall := fake.hasSdpCidArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalMediaTrack) HasSdpCidReturns(result1 bool) { + fake.hasSdpCidMutex.Lock() + defer fake.hasSdpCidMutex.Unlock() + fake.HasSdpCidStub = nil + fake.hasSdpCidReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalMediaTrack) HasSdpCidReturnsOnCall(i int, result1 bool) { + fake.hasSdpCidMutex.Lock() + defer fake.hasSdpCidMutex.Unlock() + fake.HasSdpCidStub = nil + if fake.hasSdpCidReturnsOnCall == nil { + fake.hasSdpCidReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.hasSdpCidReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalMediaTrack) ID() livekit.TrackID { fake.iDMutex.Lock() ret, specificReturn := fake.iDReturnsOnCall[len(fake.iDArgsForCall)] @@ -900,14 +962,19 @@ func (fake *FakeLocalMediaTrack) NameReturnsOnCall(i int, result1 string) { }{result1} } -func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMaxQuality(arg1 livekit.NodeID, arg2 livekit.VideoQuality) { +func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMaxQuality(arg1 livekit.NodeID, arg2 []types.SubscribedCodecQuality) { + var arg2Copy []types.SubscribedCodecQuality + if arg2 != nil { + arg2Copy = make([]types.SubscribedCodecQuality, len(arg2)) + copy(arg2Copy, arg2) + } fake.notifySubscriberNodeMaxQualityMutex.Lock() fake.notifySubscriberNodeMaxQualityArgsForCall = append(fake.notifySubscriberNodeMaxQualityArgsForCall, struct { arg1 livekit.NodeID - arg2 livekit.VideoQuality - }{arg1, arg2}) + arg2 []types.SubscribedCodecQuality + }{arg1, arg2Copy}) stub := fake.NotifySubscriberNodeMaxQualityStub - fake.recordInvocation("NotifySubscriberNodeMaxQuality", []interface{}{arg1, arg2}) + fake.recordInvocation("NotifySubscriberNodeMaxQuality", []interface{}{arg1, arg2Copy}) fake.notifySubscriberNodeMaxQualityMutex.Unlock() if stub != nil { fake.NotifySubscriberNodeMaxQualityStub(arg1, arg2) @@ -920,13 +987,13 @@ func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMaxQualityCallCount() int { return len(fake.notifySubscriberNodeMaxQualityArgsForCall) } -func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMaxQualityCalls(stub func(livekit.NodeID, livekit.VideoQuality)) { +func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMaxQualityCalls(stub func(livekit.NodeID, []types.SubscribedCodecQuality)) { fake.notifySubscriberNodeMaxQualityMutex.Lock() defer fake.notifySubscriberNodeMaxQualityMutex.Unlock() fake.NotifySubscriberNodeMaxQualityStub = stub } -func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMaxQualityArgsForCall(i int) (livekit.NodeID, livekit.VideoQuality) { +func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMaxQualityArgsForCall(i int) (livekit.NodeID, []types.SubscribedCodecQuality) { fake.notifySubscriberNodeMaxQualityMutex.RLock() defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock() argsForCall := fake.notifySubscriberNodeMaxQualityArgsForCall[i] @@ -1072,15 +1139,15 @@ func (fake *FakeLocalMediaTrack) PublisherIdentityReturnsOnCall(i int, result1 l }{result1} } -func (fake *FakeLocalMediaTrack) Receiver() sfu.TrackReceiver { - fake.receiverMutex.Lock() - ret, specificReturn := fake.receiverReturnsOnCall[len(fake.receiverArgsForCall)] - fake.receiverArgsForCall = append(fake.receiverArgsForCall, struct { +func (fake *FakeLocalMediaTrack) Receivers() []sfu.TrackReceiver { + fake.receiversMutex.Lock() + ret, specificReturn := fake.receiversReturnsOnCall[len(fake.receiversArgsForCall)] + fake.receiversArgsForCall = append(fake.receiversArgsForCall, struct { }{}) - stub := fake.ReceiverStub - fakeReturns := fake.receiverReturns - fake.recordInvocation("Receiver", []interface{}{}) - fake.receiverMutex.Unlock() + stub := fake.ReceiversStub + fakeReturns := fake.receiversReturns + fake.recordInvocation("Receivers", []interface{}{}) + fake.receiversMutex.Unlock() if stub != nil { return stub() } @@ -1090,38 +1157,38 @@ func (fake *FakeLocalMediaTrack) Receiver() sfu.TrackReceiver { return fakeReturns.result1 } -func (fake *FakeLocalMediaTrack) ReceiverCallCount() int { - fake.receiverMutex.RLock() - defer fake.receiverMutex.RUnlock() - return len(fake.receiverArgsForCall) +func (fake *FakeLocalMediaTrack) ReceiversCallCount() int { + fake.receiversMutex.RLock() + defer fake.receiversMutex.RUnlock() + return len(fake.receiversArgsForCall) } -func (fake *FakeLocalMediaTrack) ReceiverCalls(stub func() sfu.TrackReceiver) { - fake.receiverMutex.Lock() - defer fake.receiverMutex.Unlock() - fake.ReceiverStub = stub +func (fake *FakeLocalMediaTrack) ReceiversCalls(stub func() []sfu.TrackReceiver) { + fake.receiversMutex.Lock() + defer fake.receiversMutex.Unlock() + fake.ReceiversStub = stub } -func (fake *FakeLocalMediaTrack) ReceiverReturns(result1 sfu.TrackReceiver) { - fake.receiverMutex.Lock() - defer fake.receiverMutex.Unlock() - fake.ReceiverStub = nil - fake.receiverReturns = struct { - result1 sfu.TrackReceiver +func (fake *FakeLocalMediaTrack) ReceiversReturns(result1 []sfu.TrackReceiver) { + fake.receiversMutex.Lock() + defer fake.receiversMutex.Unlock() + fake.ReceiversStub = nil + fake.receiversReturns = struct { + result1 []sfu.TrackReceiver }{result1} } -func (fake *FakeLocalMediaTrack) ReceiverReturnsOnCall(i int, result1 sfu.TrackReceiver) { - fake.receiverMutex.Lock() - defer fake.receiverMutex.Unlock() - fake.ReceiverStub = nil - if fake.receiverReturnsOnCall == nil { - fake.receiverReturnsOnCall = make(map[int]struct { - result1 sfu.TrackReceiver +func (fake *FakeLocalMediaTrack) ReceiversReturnsOnCall(i int, result1 []sfu.TrackReceiver) { + fake.receiversMutex.Lock() + defer fake.receiversMutex.Unlock() + fake.ReceiversStub = nil + if fake.receiversReturnsOnCall == nil { + fake.receiversReturnsOnCall = make(map[int]struct { + result1 []sfu.TrackReceiver }) } - fake.receiverReturnsOnCall[i] = struct { - result1 sfu.TrackReceiver + fake.receiversReturnsOnCall[i] = struct { + result1 []sfu.TrackReceiver }{result1} } @@ -1272,59 +1339,6 @@ func (fake *FakeLocalMediaTrack) RevokeDisallowedSubscribersReturnsOnCall(i int, }{result1} } -func (fake *FakeLocalMediaTrack) SdpCid() string { - fake.sdpCidMutex.Lock() - ret, specificReturn := fake.sdpCidReturnsOnCall[len(fake.sdpCidArgsForCall)] - fake.sdpCidArgsForCall = append(fake.sdpCidArgsForCall, struct { - }{}) - stub := fake.SdpCidStub - fakeReturns := fake.sdpCidReturns - fake.recordInvocation("SdpCid", []interface{}{}) - fake.sdpCidMutex.Unlock() - if stub != nil { - return stub() - } - if specificReturn { - return ret.result1 - } - return fakeReturns.result1 -} - -func (fake *FakeLocalMediaTrack) SdpCidCallCount() int { - fake.sdpCidMutex.RLock() - defer fake.sdpCidMutex.RUnlock() - return len(fake.sdpCidArgsForCall) -} - -func (fake *FakeLocalMediaTrack) SdpCidCalls(stub func() string) { - fake.sdpCidMutex.Lock() - defer fake.sdpCidMutex.Unlock() - fake.SdpCidStub = stub -} - -func (fake *FakeLocalMediaTrack) SdpCidReturns(result1 string) { - fake.sdpCidMutex.Lock() - defer fake.sdpCidMutex.Unlock() - fake.SdpCidStub = nil - fake.sdpCidReturns = struct { - result1 string - }{result1} -} - -func (fake *FakeLocalMediaTrack) SdpCidReturnsOnCall(i int, result1 string) { - fake.sdpCidMutex.Lock() - defer fake.sdpCidMutex.Unlock() - fake.SdpCidStub = nil - if fake.sdpCidReturnsOnCall == nil { - fake.sdpCidReturnsOnCall = make(map[int]struct { - result1 string - }) - } - fake.sdpCidReturnsOnCall[i] = struct { - result1 string - }{result1} -} - func (fake *FakeLocalMediaTrack) SetMuted(arg1 bool) { fake.setMutedMutex.Lock() fake.setMutedArgsForCall = append(fake.setMutedArgsForCall, struct { @@ -1600,6 +1614,8 @@ func (fake *FakeLocalMediaTrack) Invocations() map[string][][]interface{} { defer fake.getConnectionScoreMutex.RUnlock() fake.getQualityForDimensionMutex.RLock() defer fake.getQualityForDimensionMutex.RUnlock() + fake.hasSdpCidMutex.RLock() + defer fake.hasSdpCidMutex.RUnlock() fake.iDMutex.RLock() defer fake.iDMutex.RUnlock() fake.isMutedMutex.RLock() @@ -1620,8 +1636,8 @@ func (fake *FakeLocalMediaTrack) Invocations() map[string][][]interface{} { defer fake.publisherIDMutex.RUnlock() fake.publisherIdentityMutex.RLock() defer fake.publisherIdentityMutex.RUnlock() - fake.receiverMutex.RLock() - defer fake.receiverMutex.RUnlock() + fake.receiversMutex.RLock() + defer fake.receiversMutex.RUnlock() fake.removeAllSubscribersMutex.RLock() defer fake.removeAllSubscribersMutex.RUnlock() fake.removeSubscriberMutex.RLock() @@ -1630,8 +1646,6 @@ func (fake *FakeLocalMediaTrack) Invocations() map[string][][]interface{} { defer fake.restartMutex.RUnlock() fake.revokeDisallowedSubscribersMutex.RLock() defer fake.revokeDisallowedSubscribersMutex.RUnlock() - fake.sdpCidMutex.RLock() - defer fake.sdpCidMutex.RUnlock() fake.setMutedMutex.RLock() defer fake.setMutedMutex.RUnlock() fake.setRTTMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index f4081ea1c..5acdf813b 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -583,12 +583,12 @@ type FakeLocalParticipant struct { updateRTTArgsForCall []struct { arg1 uint32 } - UpdateSubscribedQualityStub func(livekit.NodeID, livekit.TrackID, livekit.VideoQuality) error + UpdateSubscribedQualityStub func(livekit.NodeID, livekit.TrackID, []types.SubscribedCodecQuality) error updateSubscribedQualityMutex sync.RWMutex updateSubscribedQualityArgsForCall []struct { arg1 livekit.NodeID arg2 livekit.TrackID - arg3 livekit.VideoQuality + arg3 []types.SubscribedCodecQuality } updateSubscribedQualityReturns struct { result1 error @@ -3745,17 +3745,22 @@ func (fake *FakeLocalParticipant) UpdateRTTArgsForCall(i int) uint32 { return argsForCall.arg1 } -func (fake *FakeLocalParticipant) UpdateSubscribedQuality(arg1 livekit.NodeID, arg2 livekit.TrackID, arg3 livekit.VideoQuality) error { +func (fake *FakeLocalParticipant) UpdateSubscribedQuality(arg1 livekit.NodeID, arg2 livekit.TrackID, arg3 []types.SubscribedCodecQuality) error { + var arg3Copy []types.SubscribedCodecQuality + if arg3 != nil { + arg3Copy = make([]types.SubscribedCodecQuality, len(arg3)) + copy(arg3Copy, arg3) + } fake.updateSubscribedQualityMutex.Lock() ret, specificReturn := fake.updateSubscribedQualityReturnsOnCall[len(fake.updateSubscribedQualityArgsForCall)] fake.updateSubscribedQualityArgsForCall = append(fake.updateSubscribedQualityArgsForCall, struct { arg1 livekit.NodeID arg2 livekit.TrackID - arg3 livekit.VideoQuality - }{arg1, arg2, arg3}) + arg3 []types.SubscribedCodecQuality + }{arg1, arg2, arg3Copy}) stub := fake.UpdateSubscribedQualityStub fakeReturns := fake.updateSubscribedQualityReturns - fake.recordInvocation("UpdateSubscribedQuality", []interface{}{arg1, arg2, arg3}) + fake.recordInvocation("UpdateSubscribedQuality", []interface{}{arg1, arg2, arg3Copy}) fake.updateSubscribedQualityMutex.Unlock() if stub != nil { return stub(arg1, arg2, arg3) @@ -3772,13 +3777,13 @@ func (fake *FakeLocalParticipant) UpdateSubscribedQualityCallCount() int { return len(fake.updateSubscribedQualityArgsForCall) } -func (fake *FakeLocalParticipant) UpdateSubscribedQualityCalls(stub func(livekit.NodeID, livekit.TrackID, livekit.VideoQuality) error) { +func (fake *FakeLocalParticipant) UpdateSubscribedQualityCalls(stub func(livekit.NodeID, livekit.TrackID, []types.SubscribedCodecQuality) error) { fake.updateSubscribedQualityMutex.Lock() defer fake.updateSubscribedQualityMutex.Unlock() fake.UpdateSubscribedQualityStub = stub } -func (fake *FakeLocalParticipant) UpdateSubscribedQualityArgsForCall(i int) (livekit.NodeID, livekit.TrackID, livekit.VideoQuality) { +func (fake *FakeLocalParticipant) UpdateSubscribedQualityArgsForCall(i int) (livekit.NodeID, livekit.TrackID, []types.SubscribedCodecQuality) { fake.updateSubscribedQualityMutex.RLock() defer fake.updateSubscribedQualityMutex.RUnlock() argsForCall := fake.updateSubscribedQualityArgsForCall[i] diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index 4417e5238..12b5ef70e 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_media_track.go @@ -109,11 +109,11 @@ type FakeMediaTrack struct { nameReturnsOnCall map[int]struct { result1 string } - NotifySubscriberNodeMaxQualityStub func(livekit.NodeID, livekit.VideoQuality) + NotifySubscriberNodeMaxQualityStub func(livekit.NodeID, []types.SubscribedCodecQuality) notifySubscriberNodeMaxQualityMutex sync.RWMutex notifySubscriberNodeMaxQualityArgsForCall []struct { arg1 livekit.NodeID - arg2 livekit.VideoQuality + arg2 []types.SubscribedCodecQuality } NotifySubscriberNodeMediaLossStub func(livekit.NodeID, uint8) notifySubscriberNodeMediaLossMutex sync.RWMutex @@ -141,15 +141,15 @@ type FakeMediaTrack struct { publisherIdentityReturnsOnCall map[int]struct { result1 livekit.ParticipantIdentity } - ReceiverStub func() sfu.TrackReceiver - receiverMutex sync.RWMutex - receiverArgsForCall []struct { + ReceiversStub func() []sfu.TrackReceiver + receiversMutex sync.RWMutex + receiversArgsForCall []struct { } - receiverReturns struct { - result1 sfu.TrackReceiver + receiversReturns struct { + result1 []sfu.TrackReceiver } - receiverReturnsOnCall map[int]struct { - result1 sfu.TrackReceiver + receiversReturnsOnCall map[int]struct { + result1 []sfu.TrackReceiver } RemoveAllSubscribersStub func() removeAllSubscribersMutex sync.RWMutex @@ -744,14 +744,19 @@ func (fake *FakeMediaTrack) NameReturnsOnCall(i int, result1 string) { }{result1} } -func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQuality(arg1 livekit.NodeID, arg2 livekit.VideoQuality) { +func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQuality(arg1 livekit.NodeID, arg2 []types.SubscribedCodecQuality) { + var arg2Copy []types.SubscribedCodecQuality + if arg2 != nil { + arg2Copy = make([]types.SubscribedCodecQuality, len(arg2)) + copy(arg2Copy, arg2) + } fake.notifySubscriberNodeMaxQualityMutex.Lock() fake.notifySubscriberNodeMaxQualityArgsForCall = append(fake.notifySubscriberNodeMaxQualityArgsForCall, struct { arg1 livekit.NodeID - arg2 livekit.VideoQuality - }{arg1, arg2}) + arg2 []types.SubscribedCodecQuality + }{arg1, arg2Copy}) stub := fake.NotifySubscriberNodeMaxQualityStub - fake.recordInvocation("NotifySubscriberNodeMaxQuality", []interface{}{arg1, arg2}) + fake.recordInvocation("NotifySubscriberNodeMaxQuality", []interface{}{arg1, arg2Copy}) fake.notifySubscriberNodeMaxQualityMutex.Unlock() if stub != nil { fake.NotifySubscriberNodeMaxQualityStub(arg1, arg2) @@ -764,13 +769,13 @@ func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityCallCount() int { return len(fake.notifySubscriberNodeMaxQualityArgsForCall) } -func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityCalls(stub func(livekit.NodeID, livekit.VideoQuality)) { +func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityCalls(stub func(livekit.NodeID, []types.SubscribedCodecQuality)) { fake.notifySubscriberNodeMaxQualityMutex.Lock() defer fake.notifySubscriberNodeMaxQualityMutex.Unlock() fake.NotifySubscriberNodeMaxQualityStub = stub } -func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityArgsForCall(i int) (livekit.NodeID, livekit.VideoQuality) { +func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityArgsForCall(i int) (livekit.NodeID, []types.SubscribedCodecQuality) { fake.notifySubscriberNodeMaxQualityMutex.RLock() defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock() argsForCall := fake.notifySubscriberNodeMaxQualityArgsForCall[i] @@ -916,15 +921,15 @@ func (fake *FakeMediaTrack) PublisherIdentityReturnsOnCall(i int, result1 liveki }{result1} } -func (fake *FakeMediaTrack) Receiver() sfu.TrackReceiver { - fake.receiverMutex.Lock() - ret, specificReturn := fake.receiverReturnsOnCall[len(fake.receiverArgsForCall)] - fake.receiverArgsForCall = append(fake.receiverArgsForCall, struct { +func (fake *FakeMediaTrack) Receivers() []sfu.TrackReceiver { + fake.receiversMutex.Lock() + ret, specificReturn := fake.receiversReturnsOnCall[len(fake.receiversArgsForCall)] + fake.receiversArgsForCall = append(fake.receiversArgsForCall, struct { }{}) - stub := fake.ReceiverStub - fakeReturns := fake.receiverReturns - fake.recordInvocation("Receiver", []interface{}{}) - fake.receiverMutex.Unlock() + stub := fake.ReceiversStub + fakeReturns := fake.receiversReturns + fake.recordInvocation("Receivers", []interface{}{}) + fake.receiversMutex.Unlock() if stub != nil { return stub() } @@ -934,38 +939,38 @@ func (fake *FakeMediaTrack) Receiver() sfu.TrackReceiver { return fakeReturns.result1 } -func (fake *FakeMediaTrack) ReceiverCallCount() int { - fake.receiverMutex.RLock() - defer fake.receiverMutex.RUnlock() - return len(fake.receiverArgsForCall) +func (fake *FakeMediaTrack) ReceiversCallCount() int { + fake.receiversMutex.RLock() + defer fake.receiversMutex.RUnlock() + return len(fake.receiversArgsForCall) } -func (fake *FakeMediaTrack) ReceiverCalls(stub func() sfu.TrackReceiver) { - fake.receiverMutex.Lock() - defer fake.receiverMutex.Unlock() - fake.ReceiverStub = stub +func (fake *FakeMediaTrack) ReceiversCalls(stub func() []sfu.TrackReceiver) { + fake.receiversMutex.Lock() + defer fake.receiversMutex.Unlock() + fake.ReceiversStub = stub } -func (fake *FakeMediaTrack) ReceiverReturns(result1 sfu.TrackReceiver) { - fake.receiverMutex.Lock() - defer fake.receiverMutex.Unlock() - fake.ReceiverStub = nil - fake.receiverReturns = struct { - result1 sfu.TrackReceiver +func (fake *FakeMediaTrack) ReceiversReturns(result1 []sfu.TrackReceiver) { + fake.receiversMutex.Lock() + defer fake.receiversMutex.Unlock() + fake.ReceiversStub = nil + fake.receiversReturns = struct { + result1 []sfu.TrackReceiver }{result1} } -func (fake *FakeMediaTrack) ReceiverReturnsOnCall(i int, result1 sfu.TrackReceiver) { - fake.receiverMutex.Lock() - defer fake.receiverMutex.Unlock() - fake.ReceiverStub = nil - if fake.receiverReturnsOnCall == nil { - fake.receiverReturnsOnCall = make(map[int]struct { - result1 sfu.TrackReceiver +func (fake *FakeMediaTrack) ReceiversReturnsOnCall(i int, result1 []sfu.TrackReceiver) { + fake.receiversMutex.Lock() + defer fake.receiversMutex.Unlock() + fake.ReceiversStub = nil + if fake.receiversReturnsOnCall == nil { + fake.receiversReturnsOnCall = make(map[int]struct { + result1 []sfu.TrackReceiver }) } - fake.receiverReturnsOnCall[i] = struct { - result1 sfu.TrackReceiver + fake.receiversReturnsOnCall[i] = struct { + result1 []sfu.TrackReceiver }{result1} } @@ -1322,8 +1327,8 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} { defer fake.publisherIDMutex.RUnlock() fake.publisherIdentityMutex.RLock() defer fake.publisherIdentityMutex.RUnlock() - fake.receiverMutex.RLock() - defer fake.receiverMutex.RUnlock() + fake.receiversMutex.RLock() + defer fake.receiversMutex.RUnlock() fake.removeAllSubscribersMutex.RLock() defer fake.removeAllSubscribersMutex.RUnlock() fake.removeSubscriberMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index 7c72b558d..2e391a39c 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/pkg/rtc/types/typesfakes/fake_participant.go @@ -154,12 +154,12 @@ type FakeParticipant struct { updateMediaLossReturnsOnCall map[int]struct { result1 error } - UpdateSubscribedQualityStub func(livekit.NodeID, livekit.TrackID, livekit.VideoQuality) error + UpdateSubscribedQualityStub func(livekit.NodeID, livekit.TrackID, []types.SubscribedCodecQuality) error updateSubscribedQualityMutex sync.RWMutex updateSubscribedQualityArgsForCall []struct { arg1 livekit.NodeID arg2 livekit.TrackID - arg3 livekit.VideoQuality + arg3 []types.SubscribedCodecQuality } updateSubscribedQualityReturns struct { result1 error @@ -959,17 +959,22 @@ func (fake *FakeParticipant) UpdateMediaLossReturnsOnCall(i int, result1 error) }{result1} } -func (fake *FakeParticipant) UpdateSubscribedQuality(arg1 livekit.NodeID, arg2 livekit.TrackID, arg3 livekit.VideoQuality) error { +func (fake *FakeParticipant) UpdateSubscribedQuality(arg1 livekit.NodeID, arg2 livekit.TrackID, arg3 []types.SubscribedCodecQuality) error { + var arg3Copy []types.SubscribedCodecQuality + if arg3 != nil { + arg3Copy = make([]types.SubscribedCodecQuality, len(arg3)) + copy(arg3Copy, arg3) + } fake.updateSubscribedQualityMutex.Lock() ret, specificReturn := fake.updateSubscribedQualityReturnsOnCall[len(fake.updateSubscribedQualityArgsForCall)] fake.updateSubscribedQualityArgsForCall = append(fake.updateSubscribedQualityArgsForCall, struct { arg1 livekit.NodeID arg2 livekit.TrackID - arg3 livekit.VideoQuality - }{arg1, arg2, arg3}) + arg3 []types.SubscribedCodecQuality + }{arg1, arg2, arg3Copy}) stub := fake.UpdateSubscribedQualityStub fakeReturns := fake.updateSubscribedQualityReturns - fake.recordInvocation("UpdateSubscribedQuality", []interface{}{arg1, arg2, arg3}) + fake.recordInvocation("UpdateSubscribedQuality", []interface{}{arg1, arg2, arg3Copy}) fake.updateSubscribedQualityMutex.Unlock() if stub != nil { return stub(arg1, arg2, arg3) @@ -986,13 +991,13 @@ func (fake *FakeParticipant) UpdateSubscribedQualityCallCount() int { return len(fake.updateSubscribedQualityArgsForCall) } -func (fake *FakeParticipant) UpdateSubscribedQualityCalls(stub func(livekit.NodeID, livekit.TrackID, livekit.VideoQuality) error) { +func (fake *FakeParticipant) UpdateSubscribedQualityCalls(stub func(livekit.NodeID, livekit.TrackID, []types.SubscribedCodecQuality) error) { fake.updateSubscribedQualityMutex.Lock() defer fake.updateSubscribedQualityMutex.Unlock() fake.UpdateSubscribedQualityStub = stub } -func (fake *FakeParticipant) UpdateSubscribedQualityArgsForCall(i int) (livekit.NodeID, livekit.TrackID, livekit.VideoQuality) { +func (fake *FakeParticipant) UpdateSubscribedQualityArgsForCall(i int) (livekit.NodeID, livekit.TrackID, []types.SubscribedCodecQuality) { fake.updateSubscribedQualityMutex.RLock() defer fake.updateSubscribedQualityMutex.RUnlock() argsForCall := fake.updateSubscribedQualityArgsForCall[i] diff --git a/pkg/rtc/uptrackmanager.go b/pkg/rtc/uptrackmanager.go index 50489e55c..8be498a44 100644 --- a/pkg/rtc/uptrackmanager.go +++ b/pkg/rtc/uptrackmanager.go @@ -240,14 +240,14 @@ func (u *UpTrackManager) UpdateVideoLayers(updateVideoLayers *livekit.UpdateVide return nil } -func (u *UpTrackManager) UpdateSubscribedQuality(nodeID livekit.NodeID, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error { +func (u *UpTrackManager) UpdateSubscribedQuality(nodeID livekit.NodeID, trackID livekit.TrackID, maxQualities []types.SubscribedCodecQuality) 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) + track.NotifySubscriberNodeMaxQuality(nodeID, maxQualities) return nil } @@ -268,6 +268,7 @@ func (u *UpTrackManager) AddPublishedTrack(track types.MediaTrack) { u.publishedTracks[track.ID()] = track } u.lock.Unlock() + u.params.Logger.Debugw("added published track", "trackID", track.ID(), "trackInfo", track.ToProto().String()) track.AddOnClose(func() { notifyClose := false diff --git a/pkg/rtc/wrappedreceiver.go b/pkg/rtc/wrappedreceiver.go index a1fd3ad98..529218454 100644 --- a/pkg/rtc/wrappedreceiver.go +++ b/pkg/rtc/wrappedreceiver.go @@ -1,6 +1,8 @@ package rtc import ( + "github.com/pion/webrtc/v3" + "github.com/livekit/protocol/livekit" "github.com/livekit/livekit-server/pkg/sfu" @@ -10,22 +12,45 @@ import ( type WrappedReceiver struct { sfu.TrackReceiver - trackID livekit.TrackID - streamId string + receivers []sfu.TrackReceiver + trackID livekit.TrackID + streamId string } -func NewWrappedReceiver(receiver sfu.TrackReceiver, trackID livekit.TrackID, streamId string) WrappedReceiver { - return WrappedReceiver{ - TrackReceiver: receiver, - trackID: trackID, - streamId: streamId, +func NewWrappedReceiver(receivers []*simulcastReceiver, trackID livekit.TrackID, streamId string) *WrappedReceiver { + sfuReceivers := make([]sfu.TrackReceiver, 0, len(receivers)) + for _, r := range receivers { + sfuReceivers = append(sfuReceivers, r.TrackReceiver) + } + + return &WrappedReceiver{ + receivers: sfuReceivers, + trackID: trackID, + streamId: streamId, } } -func (r WrappedReceiver) TrackID() livekit.TrackID { +func (r *WrappedReceiver) TrackID() livekit.TrackID { return r.trackID } -func (r WrappedReceiver) StreamID() string { +func (r *WrappedReceiver) StreamID() string { return r.streamId } + +func (r *WrappedReceiver) DetermineReceiver(codec webrtc.RTPCodecCapability) { + for _, receiver := range r.receivers { + if receiver.Codec().MimeType == codec.MimeType { + r.TrackReceiver = receiver + break + } + } +} + +func (r *WrappedReceiver) Codecs() []webrtc.RTPCodecCapability { + codecs := make([]webrtc.RTPCodecCapability, 0, len(r.receivers)) + for _, receiver := range r.receivers { + codecs = append(codecs, receiver.Codec().RTPCodecCapability) + } + return codecs +} diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 4444cd3a6..2e90ff272 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -479,12 +479,12 @@ func (b *Buffer) getExtPacket(rawPacket []byte, rtpPacket *rtp.Packet, arrivalTi } case "video/h264": ep.KeyFrame = IsH264Keyframe(rtpPacket.Payload) - case "video/vp9": ep.KeyFrame = IsVp9Keyframe(rtpPacket.Payload) case "video/av1": ep.KeyFrame = IsAV1Keyframe(rtpPacket.Payload) } + if ep.KeyFrame { if b.rtpStats != nil { b.rtpStats.UpdateKeyFrame(1) diff --git a/pkg/sfu/buffer/dependencydescriptorparser.go b/pkg/sfu/buffer/dependencydescriptorparser.go index 77cf98ef9..90e1394e0 100644 --- a/pkg/sfu/buffer/dependencydescriptorparser.go +++ b/pkg/sfu/buffer/dependencydescriptorparser.go @@ -19,7 +19,7 @@ type DependencyDescriptorParser struct { } func NewDependencyDescriptorParser(ddExt uint8, logger logger.Logger, onMaxLayerChanged func(int32, int32)) *DependencyDescriptorParser { - logger.Infow("creating video stream receiver", "ddExt", ddExt) + logger.Infow("creating dependency descriptor parse", "ddExt", ddExt) return &DependencyDescriptorParser{ ddExt: ddExt, logger: logger, @@ -37,7 +37,7 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*dd.DependencyDescr } _, err := ext.Unmarshal(ddBuf) if err != nil { - r.logger.Infow("failed to parse generic dependency descriptor", "err", err) + // r.logger.Debugw("failed to parse generic dependency descriptor", "err", err, "payload", pkt.PayloadType, "ddbufLen", len(ddBuf)) return nil, videoLayer, err } @@ -45,7 +45,7 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*dd.DependencyDescr videoLayer.Spatial, videoLayer.Temporal = int32(ddVal.FrameDependencies.SpatialId), int32(ddVal.FrameDependencies.TemporalId) } if ddVal.AttachedStructure != nil && !ddVal.FirstPacketInFrame { - r.logger.Infow("ignoring non-first packet in frame with attached structure") + // r.logger.Debugw("ignoring non-first packet in frame with attached structure") return nil, videoLayer, nil } diff --git a/pkg/sfu/connectionquality/connectionstats.go b/pkg/sfu/connectionquality/connectionstats.go index 8ea44af9b..520440d19 100644 --- a/pkg/sfu/connectionquality/connectionstats.go +++ b/pkg/sfu/connectionquality/connectionstats.go @@ -20,6 +20,7 @@ const ( type ConnectionStatsParams struct { UpdateInterval time.Duration CodecType webrtc.RTPCodecType + MimeType string GetDeltaStats func() map[uint32]*buffer.StreamStatsWithLayers GetQualityParams func() *buffer.ConnectionQualityParams GetIsReducedQuality func() bool @@ -124,6 +125,7 @@ func (cs *ConnectionStats) getStat() *livekit.AnalyticsStat { return &livekit.AnalyticsStat{ Score: score, Streams: analyticsStreams, + Mime: cs.params.MimeType, } } diff --git a/pkg/sfu/dependencydescriptor/dependencydescriptorreader.go b/pkg/sfu/dependencydescriptor/dependencydescriptorreader.go index fe69af5e4..fdfff8f89 100644 --- a/pkg/sfu/dependencydescriptor/dependencydescriptorreader.go +++ b/pkg/sfu/dependencydescriptor/dependencydescriptorreader.go @@ -27,7 +27,9 @@ func NewDependencyDescriptorReader(buf []byte, structure *FrameDependencyStructu } func (r *DependencyDescriptorReader) Parse() (int, error) { - r.readMandatoryFields() + if err := r.readMandatoryFields(); err != nil { + return 0, err + } if len(r.buffer.buf) > 3 { err := r.readExtendedFields() if err != nil { diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 387784e30..e4cdee0e3 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -32,7 +32,6 @@ type TrackSender interface { Close() // ID is the globally unique identifier for this Track. ID() string - Codec() webrtc.RTPCodecCapability PeerID() livekit.ParticipantID } @@ -118,6 +117,7 @@ type DownTrack struct { forwarder *Forwarder + upstreamCodecs []webrtc.RTPCodecCapability codec webrtc.RTPCodecCapability rtpHeaderExtensions []webrtc.RTPHeaderExtensionParameter absSendTimeID int @@ -183,7 +183,7 @@ type DownTrack struct { // NewDownTrack returns a DownTrack. func NewDownTrack( - c webrtc.RTPCodecCapability, + codecs []webrtc.RTPCodecCapability, r TrackReceiver, bf *buffer.Factory, peerID livekit.ParticipantID, @@ -192,26 +192,26 @@ func NewDownTrack( ) (*DownTrack, error) { var kind webrtc.RTPCodecType switch { - case strings.HasPrefix(c.MimeType, "audio/"): + case strings.HasPrefix(codecs[0].MimeType, "audio/"): kind = webrtc.RTPCodecTypeAudio - case strings.HasPrefix(c.MimeType, "video/"): + case strings.HasPrefix(codecs[0].MimeType, "video/"): kind = webrtc.RTPCodecTypeVideo default: kind = webrtc.RTPCodecType(0) } d := &DownTrack{ - logger: logger, - id: r.TrackID(), - peerID: peerID, - maxTrack: mt, - streamID: r.StreamID(), - bufferFactory: bf, - receiver: r, - codec: c, - kind: kind, - forwarder: NewForwarder(c, kind, logger), + logger: logger, + id: r.TrackID(), + peerID: peerID, + maxTrack: mt, + streamID: r.StreamID(), + bufferFactory: bf, + receiver: r, + upstreamCodecs: codecs, + kind: kind, } + d.forwarder = NewForwarder(d.kind, d.logger) d.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{ CodecType: kind, @@ -246,11 +246,20 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, if d.bound.Load() { return webrtc.RTPCodecParameters{}, ErrTrackAlreadyBind } - parameters := webrtc.RTPCodecParameters{RTPCodecCapability: d.codec} - codec, err := codecParametersFuzzySearch(parameters, t.CodecParameters()) - if err != nil { + var codec webrtc.RTPCodecParameters + for _, c := range d.upstreamCodecs { + parameters := webrtc.RTPCodecParameters{RTPCodecCapability: c} + matchCodec, err := codecParametersFuzzySearch(parameters, t.CodecParameters()) + if err == nil { + codec = matchCodec + break + } + } + + if codec.MimeType == "" { return webrtc.RTPCodecParameters{}, webrtc.ErrUnsupportedCodec } + d.logger.Debugw("DownTrack.Bind", "codecs", d.upstreamCodecs, "matchCodec", codec) d.ssrc = uint32(t.SSRC()) d.payloadType = uint8(codec.PayloadType) @@ -264,11 +273,12 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, if strings.HasPrefix(codec.MimeType, "video/") { d.sequencer = newSequencer(d.maxTrack, d.logger) } - - d.bound.Store(true) + d.codec = codec.RTPCodecCapability + d.forwarder.DetermineCodec(d.codec) if d.onBind != nil { d.onBind() } + d.bound.Store(true) d.connectionStats.Start() d.logger.Debugw("bound") @@ -602,7 +612,12 @@ func (d *DownTrack) Close() { // 2. in case of session migration, participant migrate from other node, video track should // be resumed with same participant, set flush=false since we don't need to flush decoder. func (d *DownTrack) CloseWithFlush(flush bool) { - d.forwarder.Mute(true) + if !d.bound.Load() { + return + } + if d.forwarder != nil { + d.forwarder.Mute(true) + } // write blank frames after disabling so that other frames do not interfere. // Idea here is to send blank key frames to flush the decoder buffer at the remote end. diff --git a/pkg/sfu/downtrackspreader.go b/pkg/sfu/downtrackspreader.go index c407e208f..d3e557727 100644 --- a/pkg/sfu/downtrackspreader.go +++ b/pkg/sfu/downtrackspreader.go @@ -6,7 +6,6 @@ import ( "go.uber.org/atomic" - "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" ) @@ -82,12 +81,12 @@ func (d *DownTrackSpreader) HasDownTrack(peerID livekit.ParticipantID) bool { return ok } -func (d *DownTrackSpreader) Broadcast(layer int32, pkt *buffer.ExtPacket) { +func (d *DownTrackSpreader) Broadcast(writer func(TrackSender)) { downTracks := d.GetDownTracks() if d.params.Threshold == 0 || (len(downTracks)) < d.params.Threshold { // serial - not enough down tracks for parallelization to outweigh overhead for _, dt := range downTracks { - d.writeRTP(layer, dt, pkt) + writer(dt) } } else { // parallel - enables much more efficient multi-core utilization @@ -110,7 +109,7 @@ func (d *DownTrackSpreader) Broadcast(layer int32, pkt *buffer.ExtPacket) { } for i := n - step; i < n && i < end; i++ { - d.writeRTP(layer, downTracks[i], pkt) + writer(downTracks[i]) } } }() @@ -119,10 +118,10 @@ func (d *DownTrackSpreader) Broadcast(layer int32, pkt *buffer.ExtPacket) { } } -func (d *DownTrackSpreader) writeRTP(layer int32, dt TrackSender, pkt *buffer.ExtPacket) { - if err := dt.WriteRTP(pkt, layer); err != nil { - d.params.Logger.Errorw("failed writing to down track", err) - } +func (d *DownTrackSpreader) DownTrackCount() int { + d.downTrackMu.RLock() + defer d.downTrackMu.RUnlock() + return len(d.downTracksShadow) } func (d *DownTrackSpreader) shadowDownTracks() { diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index ae2704f5b..b3c2da4d8 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -193,9 +193,8 @@ type Forwarder struct { ddLayerSelector *DDVideoLayerSelector } -func NewForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType, logger logger.Logger) *Forwarder { +func NewForwarder(kind webrtc.RTPCodecType, logger logger.Logger) *Forwarder { f := &Forwarder{ - codec: codec, kind: kind, logger: logger, @@ -208,17 +207,6 @@ func NewForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType, log rtpMunger: NewRTPMunger(logger), } - if strings.ToLower(codec.MimeType) == "video/vp8" { - f.isTemporalSupported = true - f.vp8Munger = NewVP8Munger(logger) - } - - // TODO : we only enable dd layer selector for av1 now, at future we can - // enable it for vp9 too - if strings.ToLower(codec.MimeType) == "video/av1" { - f.ddLayerSelector = NewDDVideoLayerSelector(f.logger) - } - if f.kind == webrtc.RTPCodecTypeVideo { f.maxLayers = VideoLayers{Spatial: InvalidLayerSpatial, Temporal: DefaultMaxLayerTemporal} } else { @@ -228,6 +216,23 @@ func NewForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType, log return f } +func (f *Forwarder) DetermineCodec(codec webrtc.RTPCodecCapability) { + if f.codec.MimeType != "" { + return + } + f.codec = codec + + switch strings.ToLower(codec.MimeType) { + case "video/vp8": + f.isTemporalSupported = true + f.vp8Munger = NewVP8Munger(f.logger) + case "video/av1": + // TODO : we only enable dd layer selector for av1 now, at future we can + // enable it for vp9 too + f.ddLayerSelector = NewDDVideoLayerSelector(f.logger) + } +} + func (f *Forwarder) Mute(val bool) (bool, VideoLayers) { f.lock.Lock() defer f.lock.Unlock() diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 2e5a60b1c..eea98e7e2 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -19,8 +19,9 @@ func disable(f *Forwarder) { } func newForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType) *Forwarder { - return NewForwarder(codec, kind, logger.GetDefaultLogger()) - + f := NewForwarder(kind, logger.GetDefaultLogger()) + f.DetermineCodec(codec) + return f } func TestForwarderMute(t *testing.T) { diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index a40ab23b0..9617203a8 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -34,7 +34,8 @@ type Bitrates [DefaultMaxLayerSpatial + 1][DefaultMaxLayerTemporal + 1]int64 type TrackReceiver interface { TrackID() livekit.TrackID StreamID() string - Codec() webrtc.RTPCodecCapability + Codec() webrtc.RTPCodecParameters + HeaderExtensions() []webrtc.RTPHeaderExtensionParameter ReadRTP(buf []byte, layer uint8, sn uint16) (int, error) GetBitrateTemporalCumulative() Bitrates @@ -106,6 +107,16 @@ func RidToLayer(rid string) int32 { } } +func IsSvcCodec(mime string) bool { + switch strings.ToLower(mime) { + case "video/av1": + fallthrough + case "video/vp9": + return true + } + return false +} + type ReceiverOpts func(w *WebRTCReceiver) *WebRTCReceiver // WithPliThrottleConfig indicates minimum time(ms) between sending PLIs @@ -166,12 +177,7 @@ func NewWebRTCReceiver( isSimulcast: len(track.RID()) > 0, twcc: twcc, streamTrackerManager: NewStreamTrackerManager(logger, source), - } - switch strings.ToLower(w.codec.MimeType) { - case "video/av1": - fallthrough - case "video/vp9": - w.isSVC = true + isSVC: IsSvcCodec(track.Codec().MimeType), } w.streamTrackerManager.OnAvailableLayersChanged(w.downTrackLayerChange) @@ -256,8 +262,12 @@ func (w *WebRTCReceiver) SSRC(layer int) uint32 { return 0 } -func (w *WebRTCReceiver) Codec() webrtc.RTPCodecCapability { - return w.codec.RTPCodecCapability +func (w *WebRTCReceiver) Codec() webrtc.RTPCodecParameters { + return w.codec +} + +func (w *WebRTCReceiver) HeaderExtensions() []webrtc.RTPHeaderExtensionParameter { + return w.receiver.GetParameters().HeaderExtensions } func (w *WebRTCReceiver) Kind() webrtc.RTPCodecType { @@ -308,6 +318,7 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff if w.Kind() == webrtc.RTPCodecTypeVideo && w.useTrackers { w.streamTrackerManager.AddTracker(layer) } + go w.forwardRTP(layer) } @@ -563,7 +574,11 @@ func (w *WebRTCReceiver) forwardRTP(layer int32) { spatialTracker.Observe(pkt.Packet.SequenceNumber, pkt.Temporal, len(pkt.RawPacket), len(pkt.Packet.Payload)) } - w.downTrackSpreader.Broadcast(spatialLayer, pkt) + w.downTrackSpreader.Broadcast(func(dt TrackSender) { + if err := dt.WriteRTP(pkt, spatialLayer); err != nil { + w.logger.Errorw("failed writing to down track", err) + } + }) } } diff --git a/pkg/sfu/streamtrackermanager.go b/pkg/sfu/streamtrackermanager.go index 604d9a26d..6d22983f2 100644 --- a/pkg/sfu/streamtrackermanager.go +++ b/pkg/sfu/streamtrackermanager.go @@ -6,6 +6,7 @@ import ( "time" "github.com/go-logr/logr" + "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" ) @@ -107,7 +108,9 @@ func (s *StreamTrackerManager) AddTracker(layer int32) *StreamTracker { } params.Logger = logger.Logger(logr.Logger(s.logger).WithValues("layer", layer)) tracker := NewStreamTracker(params) + s.logger.Debugw("StreamTrackerManager add track", "layer", layer) tracker.OnStatusChanged(func(status StreamStatus) { + s.logger.Debugw("StreamTrackerManager.OnStatusChanged", "layer", layer, "status", status) if status == StreamStatusStopped { exempt := false for _, l := range exemptLayers { diff --git a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go index ed354f374..abdbefd47 100644 --- a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go +++ b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go @@ -70,13 +70,14 @@ type FakeTelemetryService struct { arg1 context.Context arg2 *livekit.Room } - TrackMaxSubscribedVideoQualityStub func(context.Context, livekit.ParticipantID, *livekit.TrackInfo, livekit.VideoQuality) + TrackMaxSubscribedVideoQualityStub func(context.Context, livekit.ParticipantID, *livekit.TrackInfo, string, livekit.VideoQuality) trackMaxSubscribedVideoQualityMutex sync.RWMutex trackMaxSubscribedVideoQualityArgsForCall []struct { arg1 context.Context arg2 livekit.ParticipantID arg3 *livekit.TrackInfo - arg4 livekit.VideoQuality + arg4 string + arg5 livekit.VideoQuality } TrackPublishedStub func(context.Context, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo) trackPublishedMutex sync.RWMutex @@ -432,19 +433,20 @@ func (fake *FakeTelemetryService) RoomStartedArgsForCall(i int) (context.Context return argsForCall.arg1, argsForCall.arg2 } -func (fake *FakeTelemetryService) TrackMaxSubscribedVideoQuality(arg1 context.Context, arg2 livekit.ParticipantID, arg3 *livekit.TrackInfo, arg4 livekit.VideoQuality) { +func (fake *FakeTelemetryService) TrackMaxSubscribedVideoQuality(arg1 context.Context, arg2 livekit.ParticipantID, arg3 *livekit.TrackInfo, arg4 string, arg5 livekit.VideoQuality) { fake.trackMaxSubscribedVideoQualityMutex.Lock() fake.trackMaxSubscribedVideoQualityArgsForCall = append(fake.trackMaxSubscribedVideoQualityArgsForCall, struct { arg1 context.Context arg2 livekit.ParticipantID arg3 *livekit.TrackInfo - arg4 livekit.VideoQuality - }{arg1, arg2, arg3, arg4}) + arg4 string + arg5 livekit.VideoQuality + }{arg1, arg2, arg3, arg4, arg5}) stub := fake.TrackMaxSubscribedVideoQualityStub - fake.recordInvocation("TrackMaxSubscribedVideoQuality", []interface{}{arg1, arg2, arg3, arg4}) + fake.recordInvocation("TrackMaxSubscribedVideoQuality", []interface{}{arg1, arg2, arg3, arg4, arg5}) fake.trackMaxSubscribedVideoQualityMutex.Unlock() if stub != nil { - fake.TrackMaxSubscribedVideoQualityStub(arg1, arg2, arg3, arg4) + fake.TrackMaxSubscribedVideoQualityStub(arg1, arg2, arg3, arg4, arg5) } } @@ -454,17 +456,17 @@ func (fake *FakeTelemetryService) TrackMaxSubscribedVideoQualityCallCount() int return len(fake.trackMaxSubscribedVideoQualityArgsForCall) } -func (fake *FakeTelemetryService) TrackMaxSubscribedVideoQualityCalls(stub func(context.Context, livekit.ParticipantID, *livekit.TrackInfo, livekit.VideoQuality)) { +func (fake *FakeTelemetryService) TrackMaxSubscribedVideoQualityCalls(stub func(context.Context, livekit.ParticipantID, *livekit.TrackInfo, string, livekit.VideoQuality)) { fake.trackMaxSubscribedVideoQualityMutex.Lock() defer fake.trackMaxSubscribedVideoQualityMutex.Unlock() fake.TrackMaxSubscribedVideoQualityStub = stub } -func (fake *FakeTelemetryService) TrackMaxSubscribedVideoQualityArgsForCall(i int) (context.Context, livekit.ParticipantID, *livekit.TrackInfo, livekit.VideoQuality) { +func (fake *FakeTelemetryService) TrackMaxSubscribedVideoQualityArgsForCall(i int) (context.Context, livekit.ParticipantID, *livekit.TrackInfo, string, livekit.VideoQuality) { fake.trackMaxSubscribedVideoQualityMutex.RLock() defer fake.trackMaxSubscribedVideoQualityMutex.RUnlock() argsForCall := fake.trackMaxSubscribedVideoQualityArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5 } func (fake *FakeTelemetryService) TrackPublished(arg1 context.Context, arg2 livekit.ParticipantID, arg3 livekit.ParticipantIdentity, arg4 *livekit.TrackInfo) { diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index 50bbe31f9..3996db003 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/pkg/telemetry/telemetryservice.go @@ -26,7 +26,7 @@ type TelemetryService interface { TrackSubscribed(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, publisher *livekit.ParticipantInfo) TrackUnsubscribed(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo) TrackPublishedUpdate(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo) - TrackMaxSubscribedVideoQuality(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, maxQuality livekit.VideoQuality) + TrackMaxSubscribedVideoQuality(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, mime string, maxQuality livekit.VideoQuality) RecordingStarted(ctx context.Context, ri *livekit.RecordingInfo) RecordingEnded(ctx context.Context, ri *livekit.RecordingInfo) EgressStarted(ctx context.Context, info *livekit.EgressInfo) @@ -153,9 +153,9 @@ func (t *telemetryService) TrackPublishedUpdate(ctx context.Context, participant }) } -func (t *telemetryService) TrackMaxSubscribedVideoQuality(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, maxQuality livekit.VideoQuality) { +func (t *telemetryService) TrackMaxSubscribedVideoQuality(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, mime string, maxQuality livekit.VideoQuality) { t.enqueue(func() { - t.internalService.TrackMaxSubscribedVideoQuality(ctx, participantID, track, maxQuality) + t.internalService.TrackMaxSubscribedVideoQuality(ctx, participantID, track, mime, maxQuality) }) } diff --git a/pkg/telemetry/telemetryserviceinternalevents.go b/pkg/telemetry/telemetryserviceinternalevents.go index 1ae0f145c..f5405fe1a 100644 --- a/pkg/telemetry/telemetryserviceinternalevents.go +++ b/pkg/telemetry/telemetryserviceinternalevents.go @@ -182,7 +182,7 @@ func (t *telemetryServiceInternal) TrackPublishedUpdate(ctx context.Context, par } func (t *telemetryServiceInternal) TrackMaxSubscribedVideoQuality(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, - maxQuality livekit.VideoQuality) { + mime string, maxQuality livekit.VideoQuality) { roomID, roomName := t.getRoomDetails(participantID) t.analytics.SendEvent(ctx, &livekit.AnalyticsEvent{ @@ -193,6 +193,7 @@ func (t *telemetryServiceInternal) TrackMaxSubscribedVideoQuality(ctx context.Co Track: track, Room: &livekit.Room{Name: string(roomName)}, MaxSubscribedVideoQuality: maxQuality, + Mime: mime, }) }