dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-27 22:34:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

simulcast codecs support (#720)

Browse Source 
simulcast codecs support 

Co-authored-by: David Zhao <dz@livekit.io>
This commit is contained in:
cnderrauber
2022-05-27 19:55:50 +08:00
committed by GitHub
co-authored by David Zhao
parent 3054f3eea3
commit f958fbcc1c
30 changed files with 1099 additions and 575 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
pkg/config/config.go
+2 -2
View File
@@ -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,
},
pkg/rtc/mediatrack.go
+71 -69
View File
@@ -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())
}
pkg/rtc/mediatrack_test.go
+205 -85
View File
@@ -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()
})
}
pkg/rtc/mediatrackreceiver.go
+191 -36
View File
@@ -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))
}
}
})
}
pkg/rtc/mediatracksubscriptions.go
+167 -66
View File
@@ -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)
}
}
pkg/rtc/participant.go
+86 -46
View File
@@ -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
}
}
pkg/rtc/participant_internal_test.go
+1 -1
View File
@@ -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
pkg/rtc/subscribedtrack.go
+3 -3
View File
@@ -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()
}
pkg/rtc/types/interfaces.go
+10 -4
View File
@@ -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
pkg/rtc/types/typesfakes/fake_local_media_track.go
+127 -113
View File
@@ -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()
pkg/rtc/types/typesfakes/fake_local_participant.go
+13 -8
View File
@@ -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]
pkg/rtc/types/typesfakes/fake_media_track.go
+53 -48
View File
@@ -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()
pkg/rtc/types/typesfakes/fake_participant.go
+13 -8
View File
@@ -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]
pkg/rtc/uptrackmanager.go
+3 -2
View File
@@ -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
pkg/rtc/wrappedreceiver.go
+34 -9
View File
@@ -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
}
pkg/sfu/buffer/buffer.go
+1 -1
View File
@@ -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)
pkg/sfu/buffer/dependencydescriptorparser.go
+3 -3
View File
@@ -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
}
pkg/sfu/connectionquality/connectionstats.go
+2
View File
@@ -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,
}
}
pkg/sfu/dependencydescriptor/dependencydescriptorreader.go
+3 -1
View File
@@ -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 {
pkg/sfu/downtrack.go
+35 -20
View File
@@ -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.
pkg/sfu/downtrackspreader.go
+7 -8
View File
@@ -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() {
pkg/sfu/forwarder.go
+18 -13
View File
@@ -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()
pkg/sfu/forwarder_test.go
+3 -2
View File
@@ -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) {
pkg/sfu/receiver.go
+25 -10
View File
@@ -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)
}
})
}
}
pkg/sfu/streamtrackermanager.go
+3
View File
@@ -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 {
pkg/telemetry/telemetryfakes/fake_telemetry_service.go
+12 -10
View File
@@ -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) {
pkg/telemetry/telemetryservice.go
+3 -3
View File
@@ -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)
})
}
pkg/telemetry/telemetryserviceinternalevents.go
+2 -1
View File
@@ -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,
})
}