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

Code cleanup (#353)

Browse Source
This commit is contained in:
David Colburn
2022-01-19 02:13:06 -08:00
committed by GitHub
parent 6400de1c10
commit 5bea9debb7
35 changed files with 223 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/server/main_test.go
+7 -5
View File
@@ -23,12 +23,14 @@ func TestGetConfigString(t *testing.T) {
{"file", "", nil, "fileContent"},
}
for _, test := range tests {
writeConfigFile(test, t)
defer os.Remove(test.configFileName)
func() {
writeConfigFile(test, t)
defer os.Remove(test.configFileName)
configBody, err := getConfigString(test.configFileName, test.configBody)
require.Equal(t, test.expectedError, err)
require.Equal(t, test.expectedConfigBody, configBody)
configBody, err := getConfigString(test.configFileName, test.configBody)
require.Equal(t, test.expectedError, err)
require.Equal(t, test.expectedConfigBody, configBody)
}()
}
}
pkg/config/ip.go
+2 -2
View File
@@ -42,8 +42,8 @@ func GetLocalIPAddresses() ([]string, error) {
if err != nil {
return nil, err
}
loopBacks := []string{}
addresses := []string{}
loopBacks := make([]string, 0)
addresses := make([]string, 0)
for _, iface := range ifaces {
addrs, err := iface.Addrs()
if err != nil {
pkg/logger/logger.go
+1 -1
View File
@@ -3,12 +3,12 @@ package serverlogger
import (
"github.com/go-logr/logr"
"github.com/go-logr/zapr"
"github.com/livekit/livekit-server/pkg/config"
"github.com/livekit/protocol/logger"
"github.com/pion/logging"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
"github.com/livekit/livekit-server/pkg/config"
"github.com/livekit/livekit-server/pkg/sfu"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
)
pkg/rtc/config.go
+1 -1
View File
@@ -4,13 +4,13 @@ import (
"errors"
"net"
logging "github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/protocol/logger"
"github.com/pion/ice/v2"
"github.com/pion/sdp/v3"
"github.com/pion/webrtc/v3"
"github.com/livekit/livekit-server/pkg/config"
logging "github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
)
pkg/rtc/mediatrack.go
+8 -8
View File
@@ -32,7 +32,7 @@ type MediaTrack struct {
numUpTracks uint32
buffer *buffer.Buffer
layerSsrcs [livekit.VideoQuality_HIGH + 1]uint32
layerSSRCs [livekit.VideoQuality_HIGH + 1]uint32
audioLevelMu sync.RWMutex
audioLevel *AudioLevel
@@ -116,8 +116,8 @@ func (t *MediaTrack) ToProto() *livekit.TrackInfo {
info.Simulcast = t.IsSimulcast()
layers := t.MediaTrackReceiver.GetVideoLayers()
for _, layer := range layers {
if int(layer.Quality) < len(t.layerSsrcs) {
layer.Ssrc = t.layerSsrcs[layer.Quality]
if int(layer.Quality) < len(t.layerSSRCs) {
layer.Ssrc = t.layerSSRCs[layer.Quality]
}
}
info.Layers = layers
@@ -228,8 +228,8 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
if t.IsSimulcast() {
layer := sfu.RidToLayer(track.RID())
if int(layer) < len(t.layerSsrcs) {
t.layerSsrcs[layer] = uint32(track.SSRC())
if int(layer) < len(t.layerSSRCs) {
t.layerSSRCs[layer] = uint32(track.SSRC())
}
}
@@ -239,8 +239,8 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
}
func (t *MediaTrack) TrySetSimulcastSSRC(layer uint8, ssrc uint32) {
if int(layer) < len(t.layerSsrcs) && t.layerSsrcs[layer] == 0 {
t.layerSsrcs[layer] = ssrc
if int(layer) < len(t.layerSSRCs) && t.layerSSRCs[layer] == 0 {
t.layerSSRCs[layer] = ssrc
}
}
@@ -262,7 +262,7 @@ func (t *MediaTrack) handlePublisherFeedback(packets []rtcp.Packet) {
var totalLost uint32
var maxSeqNum uint32
//forward to telemetry
// forward to telemetry
t.params.Telemetry.HandleRTCP(livekit.StreamType_UPSTREAM, t.params.ParticipantID, t.ID(), packets)
for _, p := range packets {
pkg/rtc/mediatrack_test.go
+20 -20
View File
@@ -129,7 +129,7 @@ func TestSubscribedMaxQuality(t *testing.T) {
mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_HIGH)
actualTrackID := livekit.TrackID("")
actualSubscribedQualities := []*livekit.SubscribedQuality{}
actualSubscribedQualities := make([]*livekit.SubscribedQuality, 0)
mt.OnSubscribedMaxQualityChange(func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, _maxSubscribedQuality livekit.VideoQuality) error {
actualTrackID = trackID
actualSubscribedQualities = subscribedQualities
@@ -140,9 +140,9 @@ func TestSubscribedMaxQuality(t *testing.T) {
mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_OFF)
expectedSubscribedQualities := []*livekit.SubscribedQuality{
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: false},
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_MEDIUM, Enabled: false},
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_HIGH, Enabled: false},
{Quality: livekit.VideoQuality_LOW, Enabled: false},
{Quality: livekit.VideoQuality_MEDIUM, Enabled: false},
{Quality: livekit.VideoQuality_HIGH, Enabled: false},
}
require.Equal(t, livekit.TrackID("v1"), actualTrackID)
require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities)
@@ -174,7 +174,7 @@ func TestSubscribedMaxQuality(t *testing.T) {
}})
actualTrackID := livekit.TrackID("")
actualSubscribedQualities := []*livekit.SubscribedQuality{}
actualSubscribedQualities := make([]*livekit.SubscribedQuality, 0)
mt.OnSubscribedMaxQualityChange(func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality, _maxSubscribedQuality livekit.VideoQuality) error {
actualTrackID = trackID
actualSubscribedQualities = subscribedQualities
@@ -185,9 +185,9 @@ func TestSubscribedMaxQuality(t *testing.T) {
mt.NotifySubscriberMaxQuality("s2", livekit.VideoQuality_MEDIUM)
expectedSubscribedQualities := []*livekit.SubscribedQuality{
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: true},
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_MEDIUM, Enabled: true},
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_HIGH, Enabled: true},
{Quality: livekit.VideoQuality_LOW, Enabled: true},
{Quality: livekit.VideoQuality_MEDIUM, Enabled: true},
{Quality: livekit.VideoQuality_HIGH, Enabled: true},
}
require.Equal(t, livekit.TrackID("v1"), actualTrackID)
require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities)
@@ -196,9 +196,9 @@ func TestSubscribedMaxQuality(t *testing.T) {
mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_MEDIUM)
expectedSubscribedQualities = []*livekit.SubscribedQuality{
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: true},
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_MEDIUM, Enabled: true},
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_HIGH, Enabled: false},
{Quality: livekit.VideoQuality_LOW, Enabled: true},
{Quality: livekit.VideoQuality_MEDIUM, Enabled: true},
{Quality: livekit.VideoQuality_HIGH, Enabled: false},
}
require.Equal(t, livekit.TrackID("v1"), actualTrackID)
require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities)
@@ -208,9 +208,9 @@ func TestSubscribedMaxQuality(t *testing.T) {
mt.NotifySubscriberMaxQuality("s2", livekit.VideoQuality_LOW)
expectedSubscribedQualities = []*livekit.SubscribedQuality{
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: true},
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_MEDIUM, Enabled: false},
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_HIGH, Enabled: false},
{Quality: livekit.VideoQuality_LOW, Enabled: true},
{Quality: livekit.VideoQuality_MEDIUM, Enabled: false},
{Quality: livekit.VideoQuality_HIGH, Enabled: false},
}
require.Equal(t, livekit.TrackID("v1"), actualTrackID)
require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities)
@@ -219,9 +219,9 @@ func TestSubscribedMaxQuality(t *testing.T) {
mt.NotifySubscriberMaxQuality("s2", livekit.VideoQuality_OFF)
expectedSubscribedQualities = []*livekit.SubscribedQuality{
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: true},
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_MEDIUM, Enabled: false},
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_HIGH, Enabled: false},
{Quality: livekit.VideoQuality_LOW, Enabled: true},
{Quality: livekit.VideoQuality_MEDIUM, Enabled: false},
{Quality: livekit.VideoQuality_HIGH, Enabled: false},
}
require.Equal(t, livekit.TrackID("v1"), actualTrackID)
require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities)
@@ -230,9 +230,9 @@ func TestSubscribedMaxQuality(t *testing.T) {
mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_OFF)
expectedSubscribedQualities = []*livekit.SubscribedQuality{
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: false},
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_MEDIUM, Enabled: false},
&livekit.SubscribedQuality{Quality: livekit.VideoQuality_HIGH, Enabled: false},
{Quality: livekit.VideoQuality_LOW, Enabled: false},
{Quality: livekit.VideoQuality_MEDIUM, Enabled: false},
{Quality: livekit.VideoQuality_HIGH, Enabled: false},
}
require.Equal(t, livekit.TrackID("v1"), actualTrackID)
require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities)
pkg/rtc/mediatrackreceiver.go
+1 -1
View File
@@ -360,7 +360,7 @@ func (t *MediaTrackReceiver) OnSubscribedMaxQualityChange(f func(trackID livekit
})
}
//---------------------------
// ---------------------------
func SpatialLayerForQuality(quality livekit.VideoQuality) int32 {
switch quality {
pkg/rtc/mediatracksubscriptions.go
+2 -2
View File
@@ -201,7 +201,7 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, code
return
}
// if the source has been terminated, we'll need to terminate all of the subscribedtracks
// if the source has been terminated, we'll need to terminate all the subscribed tracks
// however, if the dest sub has disconnected, then we can skip
if sender == nil {
return
@@ -321,7 +321,7 @@ func (t *MediaTrackSubscriptions) getSubscribedTrack(subscriberID livekit.Partic
return t.subscribedTracks[subscriberID]
}
// TODO: send for all downtracks from the source participant
// TODO: send for all down tracks from the source participant
// https://tools.ietf.org/html/rfc7941
func (t *MediaTrackSubscriptions) sendDownTrackBindingReports(sub types.LocalParticipant) {
var sd []rtcp.SourceDescriptionChunk
pkg/rtc/participant.go
+20 -20
View File
@@ -89,7 +89,7 @@ type ParticipantImpl struct {
pendingTracksLock sync.RWMutex
pendingTracks map[string]*pendingTrackInfo
*UptrackManager
*UpTrackManager
// tracks the current participant is subscribed to, map of sid => DownTrack
subscribedTracks map[livekit.TrackID]types.SubscribedTrack
@@ -204,7 +204,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
p.subscriber.OnStreamStateChange(p.onStreamStateChange)
p.setupUptrackManager()
p.setupUpTrackManager()
return p, nil
}
@@ -258,7 +258,7 @@ func (p *ParticipantImpl) ToProto() *livekit.ParticipantInfo {
Hidden: p.Hidden(),
Recorder: p.IsRecorder(),
}
info.Tracks = p.UptrackManager.ToProto()
info.Tracks = p.UpTrackManager.ToProto()
return info
}
@@ -416,7 +416,7 @@ func (p *ParticipantImpl) AddICECandidate(candidate webrtc.ICECandidateInit, tar
func (p *ParticipantImpl) Start() {
p.once.Do(func() {
p.UptrackManager.Start()
p.UpTrackManager.Start()
go p.rtcpSendWorker()
go p.downTracksRTCPWorker()
})
@@ -437,7 +437,7 @@ func (p *ParticipantImpl) Close(sendLeave bool) error {
})
}
p.UptrackManager.Close()
p.UpTrackManager.Close()
p.pendingTracksLock.Lock()
p.pendingTracks = make(map[string]*pendingTrackInfo)
@@ -449,14 +449,14 @@ func (p *ParticipantImpl) Close(sendLeave bool) error {
disallowedSubscriptions[trackID] = publisherID
}
// remove all downtracks
var downtracksToClose []*sfu.DownTrack
// remove all down tracks
var downTracksToClose []*sfu.DownTrack
for _, st := range p.subscribedTracks {
downtracksToClose = append(downtracksToClose, st.DownTrack())
downTracksToClose = append(downTracksToClose, st.DownTrack())
}
p.lock.Unlock()
for _, dt := range downtracksToClose {
for _, dt := range downTracksToClose {
dt.Close()
}
@@ -812,7 +812,7 @@ func (p *ParticipantImpl) RemoveSubscribedTrack(subTrack types.SubscribedTrack)
Message: &livekit.SignalResponse_SpeakersChanged{
SpeakersChanged: &livekit.SpeakersChanged{
Speakers: []*livekit.SpeakerInfo{
&livekit.SpeakerInfo{
{
Sid: string(subTrack.PublisherID()),
Level: 0,
Active: false,
@@ -848,13 +848,13 @@ func (p *ParticipantImpl) SubscriptionPermissionUpdate(publisherID livekit.Parti
}
}
func (p *ParticipantImpl) setupUptrackManager() {
p.UptrackManager = NewUptrackManager(UptrackManagerParams{
func (p *ParticipantImpl) setupUpTrackManager() {
p.UpTrackManager = NewUpTrackManager(UpTrackManagerParams{
SID: p.params.SID,
Logger: p.params.Logger,
})
p.UptrackManager.OnPublishedTrackUpdated(func(track types.MediaTrack, onlyIfReady bool) {
p.UpTrackManager.OnPublishedTrackUpdated(func(track types.MediaTrack, onlyIfReady bool) {
if onlyIfReady && !p.IsReady() {
return
}
@@ -864,7 +864,7 @@ func (p *ParticipantImpl) setupUptrackManager() {
}
})
p.UptrackManager.OnUptrackManagerClose(p.onUptrackManagerClose)
p.UpTrackManager.OnUpTrackManagerClose(p.onUpTrackManagerClose)
}
func (p *ParticipantImpl) sendIceCandidate(c *webrtc.ICECandidate, target livekit.SignalTarget) {
@@ -1070,7 +1070,7 @@ func (p *ParticipantImpl) downTracksRTCPWorker() {
if err == io.EOF || err == io.ErrClosedPipe {
return
}
logger.Errorw("could not send downtrack reports", err)
logger.Errorw("could not send down track reports", err)
}
}
@@ -1241,9 +1241,9 @@ func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit
}
func (p *ParticipantImpl) setTrackMuted(trackID livekit.TrackID, muted bool) {
track := p.UptrackManager.SetPublishedTrackMuted(trackID, muted)
track := p.UpTrackManager.SetPublishedTrackMuted(trackID, muted)
if track != nil {
// handled in UptrackManager for a published track, no need to update state of pending track
// handled in UpTrackManager for a published track, no need to update state of pending track
return
}
@@ -1348,7 +1348,7 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei
mt.OnSubscribedMaxQualityChange(p.onSubscribedMaxQualityChange)
// add to published and clean up pending
p.UptrackManager.AddPublishedTrack(mt)
p.UpTrackManager.AddPublishedTrack(mt)
delete(p.pendingTracks, signalCid)
newTrack = true
@@ -1398,7 +1398,7 @@ func (p *ParticipantImpl) hasPendingMigratedTrack() bool {
return false
}
func (p *ParticipantImpl) onUptrackManagerClose() {
func (p *ParticipantImpl) onUpTrackManagerClose() {
close(p.rtcpCh)
}
@@ -1505,7 +1505,7 @@ func (p *ParticipantImpl) DebugInfo() map[string]interface{} {
p.pendingTracksLock.RUnlock()
info["PendingTracks"] = pendingTrackInfo
info["UptrackManager"] = p.UptrackManager.DebugInfo()
info["UpTrackManager"] = p.UpTrackManager.DebugInfo()
subscribedTrackInfo := make(map[livekit.TrackID]interface{})
p.lock.RLock()
pkg/rtc/participant_internal_test.go
+12 -12
View File
@@ -80,15 +80,15 @@ func TestTrackPublishing(t *testing.T) {
p.OnTrackPublished(func(p types.LocalParticipant, track types.MediaTrack) {
published = true
})
p.UptrackManager.AddPublishedTrack(track)
p.UpTrackManager.AddPublishedTrack(track)
p.handleTrackPublished(track)
require.True(t, published)
require.False(t, updated)
require.Len(t, p.UptrackManager.publishedTracks, 1)
require.Len(t, p.UpTrackManager.publishedTracks, 1)
track.AddOnCloseArgsForCall(0)()
require.Len(t, p.UptrackManager.publishedTracks, 0)
require.Len(t, p.UpTrackManager.publishedTracks, 0)
require.True(t, updated)
})
@@ -137,7 +137,7 @@ func TestTrackPublishing(t *testing.T) {
track := &typesfakes.FakeLocalMediaTrack{}
track.SignalCidReturns("cid")
// directly add to publishedTracks without lock - for testing purpose only
p.UptrackManager.publishedTracks["cid"] = track
p.UpTrackManager.publishedTracks["cid"] = track
p.AddTrack(&livekit.AddTrackRequest{
Cid: "cid",
@@ -154,7 +154,7 @@ func TestTrackPublishing(t *testing.T) {
track := &typesfakes.FakeLocalMediaTrack{}
track.SdpCidReturns("cid")
// directly add to publishedTracks without lock - for testing purpose only
p.UptrackManager.publishedTracks["cid"] = track
p.UpTrackManager.publishedTracks["cid"] = track
p.AddTrack(&livekit.AddTrackRequest{
Cid: "cid",
@@ -203,7 +203,7 @@ func TestDisconnectTiming(t *testing.T) {
}
}()
track := &typesfakes.FakeMediaTrack{}
p.UptrackManager.AddPublishedTrack(track)
p.UpTrackManager.AddPublishedTrack(track)
p.handleTrackPublished(track)
// close channel and then try to Negotiate
@@ -284,30 +284,30 @@ func TestConnectionQuality(t *testing.T) {
t.Run("smooth sailing", func(t *testing.T) {
p := newParticipantForTest("test")
p.UptrackManager.publishedTracks["video"] = testPublishedVideoTrack(2, 3, 3)
p.UptrackManager.publishedTracks["audio"] = testPublishedAudioTrack(1000, 0)
p.UpTrackManager.publishedTracks["video"] = testPublishedVideoTrack(2, 3, 3)
p.UpTrackManager.publishedTracks["audio"] = testPublishedAudioTrack(1000, 0)
require.Equal(t, livekit.ConnectionQuality_EXCELLENT, p.GetConnectionQuality().GetQuality())
})
t.Run("reduced publishing", func(t *testing.T) {
p := newParticipantForTest("test")
p.UptrackManager.publishedTracks["video"] = testPublishedVideoTrack(3, 2, 3)
p.UptrackManager.publishedTracks["audio"] = testPublishedAudioTrack(1000, 100)
p.UpTrackManager.publishedTracks["video"] = testPublishedVideoTrack(3, 2, 3)
p.UpTrackManager.publishedTracks["audio"] = testPublishedAudioTrack(1000, 100)
require.Equal(t, livekit.ConnectionQuality_GOOD, p.GetConnectionQuality().GetQuality())
})
t.Run("audio smooth publishing", func(t *testing.T) {
p := newParticipantForTest("test")
p.UptrackManager.publishedTracks["audio"] = testPublishedAudioTrack(1000, 10)
p.UpTrackManager.publishedTracks["audio"] = testPublishedAudioTrack(1000, 10)
require.Equal(t, livekit.ConnectionQuality_EXCELLENT, p.GetConnectionQuality().GetQuality())
})
t.Run("audio reduced publishing", func(t *testing.T) {
p := newParticipantForTest("test")
p.UptrackManager.publishedTracks["audio"] = testPublishedAudioTrack(1000, 100)
p.UpTrackManager.publishedTracks["audio"] = testPublishedAudioTrack(1000, 100)
require.Equal(t, livekit.ConnectionQuality_GOOD, p.GetConnectionQuality().GetQuality())
})
pkg/rtc/transport.go
+5 -5
View File
@@ -8,7 +8,6 @@ import (
"github.com/bep/debounce"
"github.com/go-logr/logr"
"github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/pion/interceptor"
@@ -16,6 +15,7 @@ import (
"github.com/pion/webrtc/v3"
"github.com/livekit/livekit-server/pkg/config"
serverlogger "github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/pkg/sfu"
"github.com/livekit/livekit-server/pkg/telemetry"
@@ -352,10 +352,10 @@ func (t *PCTransport) preparePC(previousAnswer webrtc.SessionDescription) error
return err
}
// for pion generate unmatched sdp, it always append data channel to last m-lines,
// for pion generate unmatched sdp, it always appends data channel to last m-lines,
// that is not consistent with our subscribe offer which data channel is first m-lines,
// so use a dumb pc to negotiate sdp with only data channel then the data channel will
// sticky to first m-lines(subsequent sdp negotiation will keep m-lines's sequence)
// sticky to first m-lines(subsequent sdp negotiation will keep m-lines' sequence)
offer, err := t.pc.CreateOffer(nil)
if err != nil {
return err
@@ -375,7 +375,7 @@ func (t *PCTransport) preparePC(previousAnswer webrtc.SessionDescription) error
}
// replace client's fingerprint into dump pc's answer, for pion's dtls process, it will
// keep the firgerprint at first call of SetRemoteDescription, if dumb pc and client pc use
// keep the fingerprint at first call of SetRemoteDescription, if dumb pc and client pc use
// different fingerprint, that will cause pion denied dtls data after handshake with client
// complete (can't pass fingerprint change).
// in this step, we don't established connection with dump pc(no candidate swap), just use
@@ -481,7 +481,7 @@ func getMidValue(media *sdp.MediaDescription) string {
}
func extractFingerprint(desc *sdp.SessionDescription) (string, string, error) {
fingerprints := []string{}
fingerprints := make([]string, 0)
if fingerprint, haveFingerprint := desc.Attribute("fingerprint"); haveFingerprint {
fingerprints = append(fingerprints, fingerprint)
pkg/rtc/types/interfaces.go
+1 -1
View File
@@ -135,7 +135,7 @@ type LocalParticipant interface {
SetPreviousAnswer(previous *webrtc.SessionDescription)
}
// Room is a container of participants, and can provide room level actions
// Room is a container of participants, and can provide room-level actions
//counterfeiter:generate . Room
type Room interface {
Name() livekit.RoomName
pkg/rtc/unhandlesimulcast.go
+2 -2
View File
@@ -107,8 +107,8 @@ func getHeaderExtensionID(extensions []interceptor.RTPHeaderExtension, extension
func (u *UnhandleSimulcastInterceptor) BindRemoteStream(info *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader {
if t, ok := u.simTracks[info.SSRC]; ok {
// if we support fec for simulcast streams at future, should get rsid extensions
midExtensionID := getHeaderExtensionID(info.RTPHeaderExtensions, webrtc.RTPHeaderExtensionCapability{sdp.SDESMidURI})
streamIDExtensionID := getHeaderExtensionID(info.RTPHeaderExtensions, webrtc.RTPHeaderExtensionCapability{sdp.SDESRTPStreamIDURI})
midExtensionID := getHeaderExtensionID(info.RTPHeaderExtensions, webrtc.RTPHeaderExtensionCapability{URI: sdp.SDESMidURI})
streamIDExtensionID := getHeaderExtensionID(info.RTPHeaderExtensions, webrtc.RTPHeaderExtensionCapability{URI: sdp.SDESRTPStreamIDURI})
if midExtensionID == 0 || streamIDExtensionID == 0 {
return reader
}
pkg/rtc/uptrackmanager.go
+31 -31
View File
@@ -10,13 +10,13 @@ import (
"github.com/livekit/livekit-server/pkg/rtc/types"
)
type UptrackManagerParams struct {
type UpTrackManagerParams struct {
SID livekit.ParticipantID
Logger logger.Logger
}
type UptrackManager struct {
params UptrackManagerParams
type UpTrackManager struct {
params UpTrackManagerParams
closed bool
@@ -34,18 +34,18 @@ type UptrackManager struct {
onTrackUpdated func(track types.MediaTrack, onlyIfReady bool)
}
func NewUptrackManager(params UptrackManagerParams) *UptrackManager {
return &UptrackManager{
func NewUpTrackManager(params UpTrackManagerParams) *UpTrackManager {
return &UpTrackManager{
params: params,
publishedTracks: make(map[livekit.TrackID]types.MediaTrack, 0),
pendingSubscriptions: make(map[livekit.TrackID][]livekit.ParticipantID),
}
}
func (u *UptrackManager) Start() {
func (u *UpTrackManager) Start() {
}
func (u *UptrackManager) Close() {
func (u *UpTrackManager) Close() {
u.lock.Lock()
u.closed = true
@@ -62,11 +62,11 @@ func (u *UptrackManager) Close() {
}
}
func (u *UptrackManager) OnUptrackManagerClose(f func()) {
func (u *UpTrackManager) OnUpTrackManagerClose(f func()) {
u.onClose = f
}
func (u *UptrackManager) ToProto() []*livekit.TrackInfo {
func (u *UpTrackManager) ToProto() []*livekit.TrackInfo {
u.lock.RLock()
defer u.lock.RUnlock()
@@ -78,12 +78,12 @@ func (u *UptrackManager) ToProto() []*livekit.TrackInfo {
return trackInfos
}
func (u *UptrackManager) OnPublishedTrackUpdated(f func(track types.MediaTrack, onlyIfReady bool)) {
func (u *UpTrackManager) OnPublishedTrackUpdated(f func(track types.MediaTrack, onlyIfReady bool)) {
u.onTrackUpdated = f
}
// AddSubscriber subscribes op to all publishedTracks
func (u *UptrackManager) AddSubscriber(sub types.LocalParticipant, params types.AddSubscriberParams) (int, error) {
func (u *UpTrackManager) AddSubscriber(sub types.LocalParticipant, params types.AddSubscriberParams) (int, error) {
var tracks []types.MediaTrack
if params.AllTracks {
tracks = u.GetPublishedTracks()
@@ -127,7 +127,7 @@ func (u *UptrackManager) AddSubscriber(sub types.LocalParticipant, params types.
return n, nil
}
func (u *UptrackManager) RemoveSubscriber(sub types.LocalParticipant, trackID livekit.TrackID) {
func (u *UpTrackManager) RemoveSubscriber(sub types.LocalParticipant, trackID livekit.TrackID) {
track := u.GetPublishedTrack(trackID)
if track != nil {
track.RemoveSubscriber(sub.ID())
@@ -138,7 +138,7 @@ func (u *UptrackManager) RemoveSubscriber(sub types.LocalParticipant, trackID li
u.lock.Unlock()
}
func (u *UptrackManager) SetPublishedTrackMuted(trackID livekit.TrackID, muted bool) types.MediaTrack {
func (u *UpTrackManager) SetPublishedTrackMuted(trackID livekit.TrackID, muted bool) types.MediaTrack {
u.lock.RLock()
track := u.publishedTracks[trackID]
u.lock.RUnlock()
@@ -158,14 +158,14 @@ func (u *UptrackManager) SetPublishedTrackMuted(trackID livekit.TrackID, muted b
return track
}
func (u *UptrackManager) GetPublishedTrack(trackID livekit.TrackID) types.MediaTrack {
func (u *UpTrackManager) GetPublishedTrack(trackID livekit.TrackID) types.MediaTrack {
u.lock.RLock()
defer u.lock.RUnlock()
return u.getPublishedTrack(trackID)
}
func (u *UptrackManager) GetPublishedTracks() []types.MediaTrack {
func (u *UpTrackManager) GetPublishedTracks() []types.MediaTrack {
u.lock.RLock()
defer u.lock.RUnlock()
@@ -176,7 +176,7 @@ func (u *UptrackManager) GetPublishedTracks() []types.MediaTrack {
return tracks
}
func (u *UptrackManager) UpdateSubscriptionPermissions(
func (u *UpTrackManager) UpdateSubscriptionPermissions(
permissions *livekit.UpdateSubscriptionPermissions,
resolver func(participantID livekit.ParticipantID) types.LocalParticipant,
) error {
@@ -192,7 +192,7 @@ func (u *UptrackManager) UpdateSubscriptionPermissions(
return nil
}
func (u *UptrackManager) UpdateVideoLayers(updateVideoLayers *livekit.UpdateVideoLayers) error {
func (u *UpTrackManager) UpdateVideoLayers(updateVideoLayers *livekit.UpdateVideoLayers) error {
track := u.GetPublishedTrack(livekit.TrackID(updateVideoLayers.TrackSid))
if track == nil {
u.params.Logger.Warnw("could not find track", nil, "trackID", livekit.TrackID(updateVideoLayers.TrackSid))
@@ -203,7 +203,7 @@ func (u *UptrackManager) UpdateVideoLayers(updateVideoLayers *livekit.UpdateVide
return nil
}
func (u *UptrackManager) UpdateSubscribedQuality(nodeID string, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error {
func (u *UpTrackManager) UpdateSubscribedQuality(nodeID string, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error {
track := u.GetPublishedTrack(trackID)
if track == nil {
u.params.Logger.Warnw("could not find track", nil, "trackID", trackID)
@@ -214,7 +214,7 @@ func (u *UptrackManager) UpdateSubscribedQuality(nodeID string, trackID livekit.
return nil
}
func (u *UptrackManager) UpdateMediaLoss(nodeID string, trackID livekit.TrackID, fractionalLoss uint32) error {
func (u *UpTrackManager) UpdateMediaLoss(nodeID string, trackID livekit.TrackID, fractionalLoss uint32) error {
track := u.GetPublishedTrack(trackID)
if track == nil {
u.params.Logger.Warnw("could not find track", nil, "trackID", trackID)
@@ -225,7 +225,7 @@ func (u *UptrackManager) UpdateMediaLoss(nodeID string, trackID livekit.TrackID,
return nil
}
func (u *UptrackManager) AddPublishedTrack(track types.MediaTrack) {
func (u *UpTrackManager) AddPublishedTrack(track types.MediaTrack) {
u.lock.Lock()
if _, ok := u.publishedTracks[track.ID()]; !ok {
u.publishedTracks[track.ID()] = track
@@ -258,16 +258,16 @@ func (u *UptrackManager) AddPublishedTrack(track types.MediaTrack) {
})
}
func (u *UptrackManager) RemovePublishedTrack(track types.MediaTrack) {
func (u *UpTrackManager) RemovePublishedTrack(track types.MediaTrack) {
track.RemoveAllSubscribers()
}
// should be called with lock held
func (u *UptrackManager) getPublishedTrack(trackID livekit.TrackID) types.MediaTrack {
func (u *UpTrackManager) getPublishedTrack(trackID livekit.TrackID) types.MediaTrack {
return u.publishedTracks[trackID]
}
func (u *UptrackManager) updateSubscriptionPermissions(permissions *livekit.UpdateSubscriptionPermissions) {
func (u *UpTrackManager) updateSubscriptionPermissions(permissions *livekit.UpdateSubscriptionPermissions) {
// every update overrides the existing
// all_participants takes precedence
@@ -284,7 +284,7 @@ func (u *UptrackManager) updateSubscriptionPermissions(permissions *livekit.Upda
}
}
func (u *UptrackManager) hasPermission(trackID livekit.TrackID, subscriberID livekit.ParticipantID) bool {
func (u *UpTrackManager) hasPermission(trackID livekit.TrackID, subscriberID livekit.ParticipantID) bool {
if u.subscriptionPermissions == nil {
return true
}
@@ -307,12 +307,12 @@ func (u *UptrackManager) hasPermission(trackID livekit.TrackID, subscriberID liv
return false
}
func (u *UptrackManager) getAllowedSubscribers(trackID livekit.TrackID) []livekit.ParticipantID {
func (u *UpTrackManager) getAllowedSubscribers(trackID livekit.TrackID) []livekit.ParticipantID {
if u.subscriptionPermissions == nil {
return nil
}
allowed := []livekit.ParticipantID{}
allowed := make([]livekit.ParticipantID, 0)
for subscriberID, perms := range u.subscriptionPermissions {
if perms.AllTracks {
allowed = append(allowed, subscriberID)
@@ -330,7 +330,7 @@ func (u *UptrackManager) getAllowedSubscribers(trackID livekit.TrackID) []liveki
return allowed
}
func (u *UptrackManager) maybeAddPendingSubscription(trackID livekit.TrackID, sub types.LocalParticipant) {
func (u *UpTrackManager) maybeAddPendingSubscription(trackID livekit.TrackID, sub types.LocalParticipant) {
subscriberID := sub.ID()
pending := u.pendingSubscriptions[trackID]
@@ -345,7 +345,7 @@ func (u *UptrackManager) maybeAddPendingSubscription(trackID livekit.TrackID, su
go sub.SubscriptionPermissionUpdate(u.params.SID, trackID, false)
}
func (u *UptrackManager) maybeRemovePendingSubscription(trackID livekit.TrackID, sub types.LocalParticipant) {
func (u *UpTrackManager) maybeRemovePendingSubscription(trackID livekit.TrackID, sub types.LocalParticipant) {
subscriberID := sub.ID()
pending := u.pendingSubscriptions[trackID]
@@ -362,7 +362,7 @@ func (u *UptrackManager) maybeRemovePendingSubscription(trackID livekit.TrackID,
}
}
func (u *UptrackManager) processPendingSubscriptions(resolver func(participantID livekit.ParticipantID) types.LocalParticipant) {
func (u *UpTrackManager) processPendingSubscriptions(resolver func(participantID livekit.ParticipantID) types.LocalParticipant) {
updatedPendingSubscriptions := make(map[livekit.TrackID][]livekit.ParticipantID)
for trackID, pending := range u.pendingSubscriptions {
track := u.getPublishedTrack(trackID)
@@ -402,7 +402,7 @@ func (u *UptrackManager) processPendingSubscriptions(resolver func(participantID
u.pendingSubscriptions = updatedPendingSubscriptions
}
func (u *UptrackManager) maybeRevokeSubscriptions(resolver func(participantID livekit.ParticipantID) types.LocalParticipant) {
func (u *UpTrackManager) maybeRevokeSubscriptions(resolver func(participantID livekit.ParticipantID) types.LocalParticipant) {
for _, track := range u.publishedTracks {
trackID := track.ID()
allowed := u.getAllowedSubscribers(trackID)
@@ -426,7 +426,7 @@ func (u *UptrackManager) maybeRevokeSubscriptions(resolver func(participantID li
}
}
func (u *UptrackManager) DebugInfo() map[string]interface{} {
func (u *UpTrackManager) DebugInfo() map[string]interface{} {
info := map[string]interface{}{}
publishedTrackInfo := make(map[livekit.TrackID]interface{})
pkg/rtc/uptrackmanager_test.go
+4 -3
View File
@@ -3,14 +3,15 @@ package rtc
import (
"testing"
"github.com/livekit/livekit-server/pkg/rtc/types/typesfakes"
"github.com/livekit/protocol/livekit"
"github.com/stretchr/testify/require"
"github.com/livekit/livekit-server/pkg/rtc/types/typesfakes"
)
func TestUpdateSubscriptionPermissions(t *testing.T) {
t.Run("updates permissions", func(t *testing.T) {
um := NewUptrackManager(UptrackManagerParams{})
um := NewUpTrackManager(UpTrackManagerParams{})
tra := &typesfakes.FakeMediaTrack{}
tra.IDReturns("audio")
@@ -85,7 +86,7 @@ func TestUpdateSubscriptionPermissions(t *testing.T) {
func TestPermissions(t *testing.T) {
t.Run("checks permissions", func(t *testing.T) {
um := NewUptrackManager(UptrackManagerParams{})
um := NewUpTrackManager(UpTrackManagerParams{})
tra := &typesfakes.FakeMediaTrack{}
tra.IDReturns("audio")
pkg/rtc/wrappedreceiver.go
+2 -1
View File
@@ -1,8 +1,9 @@
package rtc
import (
"github.com/livekit/livekit-server/pkg/sfu"
"github.com/livekit/protocol/livekit"
"github.com/livekit/livekit-server/pkg/sfu"
)
// wrapper around WebRTC receiver, overriding its ID
pkg/service/redisroomstore_test.go
+1 -1
View File
@@ -48,7 +48,7 @@ func TestParticipantPersistence(t *testing.T) {
require.NoError(t, err)
require.Len(t, participants, 1)
// deleting participant should return back to normal
// deleting participant should return to normal
require.NoError(t, rs.DeleteParticipant(ctx, roomName, livekit.ParticipantIdentity(p.Identity)))
participants, err = rs.ListParticipants(ctx, roomName)
pkg/sfu/buffer/buffer.go
+1 -1
View File
@@ -83,7 +83,7 @@ type Buffer struct {
latestTimestamp uint32 // latest received RTP timestamp on packet
latestTimestampTime int64 // Time of the latest timestamp (in nanos since unix epoch)
lastFractionLostToReport uint8 // Last fractionlost from subscribers, should report to publisher; Audio only
lastFractionLostToReport uint8 // Last fraction lost from subscribers, should report to publisher; Audio only
// callbacks
onClose func()
pkg/sfu/connectionquality/mos.go
+1 -1
View File
@@ -40,7 +40,7 @@ func mosAudioEmodel(cur, prev *ConnectionStat) float64 {
rx := 93.2 - percentageLost
ry := 0.18*rx*rx - 27.9*rx + 1126.62
//Jitter is in MicroSecs (1/1e6) units. Convert it to MilliSecs
// Jitter is in MicroSecs (1/1e6) units. Convert it to MilliSecs
d := float64(rtt + (cur.Jitter / 1000))
h := d - 177.3
if h < 0 {
pkg/sfu/downtrack.go
+4 -5
View File
@@ -10,9 +10,7 @@ import (
"sync/atomic"
"time"
"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
"github.com/livekit/protocol/livekit"
"github.com/pion/rtcp"
"github.com/pion/rtp"
"github.com/pion/sdp/v3"
@@ -20,6 +18,7 @@ import (
"github.com/pion/webrtc/v3"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
)
const (
@@ -28,7 +27,7 @@ const (
// TrackSender defines an interface send media to remote peer
type TrackSender interface {
UptrackLayersChange(availableLayers []uint16)
UpTrackLayersChange(availableLayers []uint16)
WriteRTP(p *buffer.ExtPacket, layer int32) error
Close()
// ID is the globally unique identifier for this Track.
@@ -480,8 +479,8 @@ func (d *DownTrack) GetForwardingStatus() ForwardingStatus {
return d.forwarder.GetForwardingStatus()
}
func (d *DownTrack) UptrackLayersChange(availableLayers []uint16) {
d.forwarder.UptrackLayersChange(availableLayers)
func (d *DownTrack) UpTrackLayersChange(availableLayers []uint16) {
d.forwarder.UpTrackLayersChange(availableLayers)
if d.onAvailableLayersChanged != nil {
d.onAvailableLayersChanged(d)
pkg/sfu/forwarder.go
+2 -2
View File
@@ -293,7 +293,7 @@ func (f *Forwarder) GetForwardingStatus() ForwardingStatus {
return ForwardingStatusOptimal
}
func (f *Forwarder) UptrackLayersChange(availableLayers []uint16) {
func (f *Forwarder) UpTrackLayersChange(availableLayers []uint16) {
f.lock.Lock()
defer f.lock.Unlock()
@@ -711,7 +711,7 @@ func (f *Forwarder) ProvisionalAllocateGetBestWeightedTransition() VideoTransiti
}
}
// starting from mimimum to target, find transition which gives the best
// starting from minimum to target, find transition which gives the best
// transition taking into account bits saved vs cost of such a transition
bestLayers := InvalidLayers
bestBandwidthDelta := int64(0)
pkg/sfu/forwarder_test.go
+8 -8
View File
@@ -94,7 +94,7 @@ func TestForwarderGetForwardingStatus(t *testing.T) {
// with available layers, should be off
availableLayers := []uint16{0, 1, 2}
f.UptrackLayersChange(availableLayers)
f.UpTrackLayersChange(availableLayers)
require.Equal(t, ForwardingStatusOff, f.GetForwardingStatus())
// when muted, should be optimal
@@ -112,25 +112,25 @@ func TestForwarderGetForwardingStatus(t *testing.T) {
// when available layers are lower than max subscribed, optimal as long as target is at max available
availableLayers = []uint16{0, 1}
f.UptrackLayersChange(availableLayers)
f.UpTrackLayersChange(availableLayers)
require.Equal(t, ForwardingStatusOptimal, f.GetForwardingStatus())
}
func TestForwarderUptrackLayersChange(t *testing.T) {
func TestForwarderUpTrackLayersChange(t *testing.T) {
f := NewForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo)
require.Nil(t, f.availableLayers)
availableLayers := []uint16{0, 1, 2}
f.UptrackLayersChange(availableLayers)
f.UpTrackLayersChange(availableLayers)
require.Equal(t, availableLayers, f.availableLayers)
availableLayers = []uint16{0, 2}
f.UptrackLayersChange(availableLayers)
f.UpTrackLayersChange(availableLayers)
require.Equal(t, availableLayers, f.availableLayers)
availableLayers = []uint16{}
f.UptrackLayersChange(availableLayers)
f.UpTrackLayersChange(availableLayers)
require.Equal(t, availableLayers, f.availableLayers)
}
@@ -182,7 +182,7 @@ func TestForwarderAllocate(t *testing.T) {
// awaiting measurement, i.e. bitrates are not available, but layers available
f.lastAllocation.state = VideoAllocationStateNone
disable(f)
f.UptrackLayersChange([]uint16{0})
f.UpTrackLayersChange([]uint16{0})
expectedTargetLayers := VideoLayers{
spatial: 0,
temporal: DefaultMaxLayerTemporal,
@@ -611,7 +611,7 @@ func TestForwarderFinalizeAllocate(t *testing.T) {
// layers available, but still awaiting measurement
f.lastAllocation.state = VideoAllocationStateAwaitingMeasurement
disable(f)
f.UptrackLayersChange([]uint16{0, 1})
f.UpTrackLayersChange([]uint16{0, 1})
expectedResult = VideoAllocation{
state: VideoAllocationStateAwaitingMeasurement,
change: VideoStreamingChangeNone,
pkg/sfu/prober.go
+5 -5
View File
@@ -91,12 +91,12 @@
//
// A few things to note
// 1. When a probe cluster is added, the expected media rate is provided.
// So, the wake up interval takes that into account. For example,
// So, the wake-up interval takes that into account. For example,
// if probing at 5 Mbps for 1/2 second and if 4 Mbps of it is expected
// to be provided by media traffic, the wake up interval becomes 8 ms.
// to be provided by media traffic, the wake-up interval becomes 8 ms.
// 2. The amount of probing should actually be capped at some value to
// avoid too much self-induced congestion. It maybe something like 500 kbps.
// That will increase the wake up interval to 16 ms in the above example.
// That will increase the wake-up interval to 16 ms in the above example.
// 3. In practice, the probing interval may also be shorter. Typically,
// it can be run for 2 - 3 RTTs to get a good measurement. For
// the longest hauls, RTT could be 250 ms or so leading to the probing
@@ -356,9 +356,9 @@ func (c *Cluster) Process(p *Prober) {
// Calculate number of probe bytes that should have been sent since start.
// Overall goal is to send desired number of probe bytes in minDuration.
// However it is possible that timeElapsed is more than minDuration due
// However, it is possible that timeElapsed is more than minDuration due
// to scheduling variance. When overshooting time budget, use a capped
// short fall if there a grace period given.
// short fall if there is a grace period given.
windowDone := float64(timeElapsed) / float64(c.minDuration)
if windowDone > 1.0 {
// cluster has been running for longer than minDuration
pkg/sfu/receiver.go
+10 -10
View File
@@ -9,12 +9,12 @@ import (
"sync/atomic"
"time"
"github.com/livekit/protocol/livekit"
"github.com/pion/rtcp"
"github.com/pion/webrtc/v3"
"github.com/rs/zerolog/log"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
"github.com/livekit/protocol/livekit"
)
type AudioLevelHandle func(level uint8, duration uint32)
@@ -243,12 +243,12 @@ func (w *WebRTCReceiver) AddDownTrack(track TrackSender) {
}
if w.Kind() == webrtc.RTPCodecTypeVideo {
// notify added downtrack of available layers
// notify added down track of available layers
w.upTrackMu.RLock()
layers, ok := w.availableLayers.Load().([]uint16)
w.upTrackMu.RUnlock()
if ok && len(layers) != 0 {
track.UptrackLayersChange(layers)
track.UpTrackLayersChange(layers)
}
}
@@ -308,12 +308,12 @@ func (w *WebRTCReceiver) SetMaxExpectedSpatialLayer(layer int32) {
//
// Some higher layer is expected to start.
// If the layer was not stopped (i. e. it will still be in available layers),
// If the layer was not stopped (i.e. it will still be in available layers),
// don't need to do anything. If not, reset the stream tracker so that
// the layer is declared available on the first packet
//
// NOTE: There may be a race between checking if a layer is available and
// resetting the tracker, i. e. the track may stop just after checking.
// resetting the tracker, i.e. the track may stop just after checking.
// But, those conditions should be rare. In those cases, the restart will
// take longer.
//
@@ -339,14 +339,14 @@ func (w *WebRTCReceiver) NumAvailableSpatialLayers() int {
return len(layers)
}
func (w *WebRTCReceiver) downtrackLayerChange(layers []uint16) {
func (w *WebRTCReceiver) downTrackLayerChange(layers []uint16) {
w.downTrackMu.RLock()
downTracks := w.downTracks
w.downTrackMu.RUnlock()
for _, dt := range downTracks {
if dt != nil {
dt.UptrackLayersChange(layers)
dt.UpTrackLayersChange(layers)
}
}
}
@@ -371,7 +371,7 @@ func (w *WebRTCReceiver) addAvailableLayer(layer uint16) {
w.availableLayers.Store(layers)
w.upTrackMu.Unlock()
w.downtrackLayerChange(layers)
w.downTrackLayerChange(layers)
}
func (w *WebRTCReceiver) removeAvailableLayer(layer uint16) {
@@ -392,7 +392,7 @@ func (w *WebRTCReceiver) removeAvailableLayer(layer uint16) {
w.upTrackMu.Unlock()
// need to immediately switch off unavailable layers
w.downtrackLayerChange(newLayers)
w.downTrackLayerChange(newLayers)
}
func (w *WebRTCReceiver) GetBitrateTemporalCumulative() Bitrates {
@@ -558,7 +558,7 @@ func (w *WebRTCReceiver) forwardRTP(layer int32) {
func (w *WebRTCReceiver) writeRTP(layer int32, dt TrackSender, pkt *buffer.ExtPacket) {
if err := dt.WriteRTP(pkt, layer); err != nil {
log.Error().Err(err).Str("id", string(dt.ID())).Msg("Error writing to down track")
log.Error().Err(err).Str("id", dt.ID()).Msg("Error writing to down track")
}
}
pkg/sfu/sequencer.go
+1 -1
View File
@@ -44,7 +44,7 @@ type packetMeta struct {
lastNack uint32
// Spatial layer of packet
layer uint8
// Information that differs depending the codec
// Information that differs depending on the codec
misc uint64
}
pkg/sfu/sequencer_test.go
+1 -1
View File
@@ -114,7 +114,7 @@ func Test_packetMeta_VP8(t *testing.T) {
p.packVP8(vp8)
// booleans are not packed, so they will be `false` in unpacked.
// Also TID is only two bits, so it should be modulo 3.
// Also, TID is only two bits, so it should be modulo 3.
expectedVP8 := &buffer.VP8{
FirstByte: 25,
PictureIDPresent: 1,
pkg/sfu/streamallocator.go
+4 -3
View File
@@ -7,11 +7,12 @@ import (
"sync"
"time"
"github.com/livekit/livekit-server/pkg/config"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/pion/rtcp"
"github.com/pion/webrtc/v3"
"github.com/livekit/livekit-server/pkg/config"
)
const (
@@ -817,7 +818,7 @@ func (s *StreamAllocator) allocateAllTracks() {
availableChannelCapacity := s.committedChannelCapacity
//
// This pass is just to find out if there is any left over channel capacity.
// This pass is just to find out if there is any leftover channel capacity.
// Infinite channel capacity is given so that exempt tracks do not stall
//
for _, track := range s.exemptVideoTracksSorted {
@@ -974,7 +975,7 @@ func (s *StreamAllocator) maybeBoostLayer() {
}
func (s *StreamAllocator) isTimeToBoost() bool {
// if enough time has passed since last esitmate drop or last estimate boost,
// if enough time has passed since last estimate drop or last estimate boost,
// artificially boost estimate before allocating.
// Checking against last estimate boost prevents multiple artificial boosts
// in situations where multiple tracks become available in a short span.
pkg/sfu/streamtracker.go
+2 -2
View File
@@ -24,7 +24,7 @@ const (
StreamStatusActive StreamStatus = 1
)
// StreamTracker keeps track of packet flow and ensures a particular uptrack is consistently producing
// StreamTracker keeps track of packet flow and ensures a particular up track is consistently producing
// It runs its own goroutine for detection, and fires OnStatusChanged callback
type StreamTracker struct {
// number of samples needed per cycle
@@ -169,7 +169,7 @@ func (s *StreamTracker) Observe(sn uint16) {
s.lastSN = sn
atomic.AddUint32(&s.countSinceLast, 1)
// declare stream active and start the detect worker
// declare stream active and start the detection worker
go s.init()
return
pkg/sfu/streamtracker_test.go
+1 -1
View File
@@ -100,7 +100,7 @@ func TestStreamTracker(t *testing.T) {
require.Equal(t, StreamStatusActive, tracker.Status())
require.Equal(t, uint32(1), callbackCalled.get())
// obaerver a few more
// observe a few more
tracker.Observe(2)
tracker.Observe(3)
tracker.Observe(4)
pkg/telemetry/rtt.go
+2 -2
View File
@@ -37,8 +37,8 @@ func SaturatedUsToCompactNtp(us int64) uint32 {
}
// TimeMicrosToNtp convert us to NtpTime (from webrtc)
func TimeMicrosToNtp(time_us int64) *NtpTime {
timeNtpUs := time_us + NtpJan1970Sec*MicroSecondsInSecond
func TimeMicrosToNtp(us int64) *NtpTime {
timeNtpUs := us + NtpJan1970Sec*MicroSecondsInSecond
// Convert seconds to uint32 through uint64 for well-defined cast.
// Wrap around (will happen in 2036) is expected for ntp time.
pkg/telemetry/statsworker.go
+3 -3
View File
@@ -108,9 +108,9 @@ func (s *StatsWorker) calculateTotalBytesPackets(allBuffers []*buffer.Buffer) (t
totalBytes = 0
totalPackets = 0
for _, buffer := range allBuffers {
totalBytes += buffer.GetStats().TotalByte
totalPackets += buffer.GetStats().PacketCount
for _, buff := range allBuffers {
totalBytes += buff.GetStats().TotalByte
totalPackets += buff.GetStats().PacketCount
}
return totalBytes, totalPackets
}
pkg/telemetry/test/telemetry_service_events_test.go
+6 -6
View File
@@ -11,7 +11,7 @@ import (
func Test_OnParticipantJoin_EventIsSent(t *testing.T) {
fixture := createFixture()
//prepare
// prepare
room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
partSID := "part1"
clientInfo := &livekit.ClientInfo{
@@ -25,10 +25,10 @@ func Test_OnParticipantJoin_EventIsSent(t *testing.T) {
}
participantInfo := &livekit.ParticipantInfo{Sid: partSID}
//do
// do
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo)
//test
// test
require.Equal(t, 1, fixture.analytics.SendEventCallCount())
_, event := fixture.analytics.SendEventArgsForCall(0)
require.Equal(t, livekit.AnalyticsEventType_PARTICIPANT_JOINED, event.Type)
@@ -48,15 +48,15 @@ func Test_OnParticipantJoin_EventIsSent(t *testing.T) {
func Test_OnParticipantLeft_EventIsSent(t *testing.T) {
fixture := createFixture()
//prepare
// prepare
room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
partSID := "part1"
participantInfo := &livekit.ParticipantInfo{Sid: partSID}
//do
// do
fixture.sut.ParticipantLeft(context.Background(), room, participantInfo)
//test
// test
require.Equal(t, 1, fixture.analytics.SendEventCallCount())
_, event := fixture.analytics.SendEventArgsForCall(0)
require.Equal(t, livekit.AnalyticsEventType_PARTICIPANT_LEFT, event.Type)
pkg/telemetry/test/telemetry_service_test.go
+46 -46
View File
@@ -28,19 +28,19 @@ func createFixture() *telemetryServiceFixture {
func Test_ParticipantAndRoomDataAreSentWithAnalytics(t *testing.T) {
fixture := createFixture()
//prepare
// prepare
room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
partSID := livekit.ParticipantID("part1")
clientInfo := &livekit.ClientInfo{Sdk: 2}
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo)
//do
// do
packet := 33
fixture.sut.OnDownstreamPacket(partSID, "", packet)
fixture.sut.SendAnalytics()
//test
// test
require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
_, stats := fixture.analytics.SendStatsArgsForCall(0)
require.Equal(t, 1, len(stats))
@@ -53,14 +53,14 @@ func Test_ParticipantAndRoomDataAreSentWithAnalytics(t *testing.T) {
func Test_OnDownstreamPackets(t *testing.T) {
fixture := createFixture()
//prepare
// prepare
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
clientInfo := &livekit.ClientInfo{Sdk: 2}
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo)
//do
// do
packets := []int{33, 23}
totalBytes := packets[0] + packets[1]
totalPackets := len(packets)
@@ -70,7 +70,7 @@ func Test_OnDownstreamPackets(t *testing.T) {
}
fixture.sut.SendAnalytics()
//test
// test
require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
_, stats := fixture.analytics.SendStatsArgsForCall(0)
require.Equal(t, 1, len(stats))
@@ -83,14 +83,14 @@ func Test_OnDownstreamPackets(t *testing.T) {
func Test_OnDownstreamPackets_SeveralTracks(t *testing.T) {
fixture := createFixture()
//prepare
// prepare
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
clientInfo := &livekit.ClientInfo{Sdk: 2}
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo)
//do
// do
packet1 := 33
trackID1 := livekit.TrackID("trackID1")
packet2 := 23
@@ -99,7 +99,7 @@ func Test_OnDownstreamPackets_SeveralTracks(t *testing.T) {
fixture.sut.OnDownstreamPacket(partSID, trackID2, packet2)
fixture.sut.SendAnalytics()
//test
// test
require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
_, stats := fixture.analytics.SendStatsArgsForCall(0)
require.Equal(t, 2, len(stats))
@@ -124,13 +124,13 @@ func Test_OnDownstreamPackets_SeveralTracks(t *testing.T) {
func Test_OnDownStreamRTCP(t *testing.T) {
fixture := createFixture()
//prepare
// prepare
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil)
//do
// do
pkts := []rtcp.Packet{
&rtcp.TransportLayerNack{},
&rtcp.PictureLossIndication{},
@@ -147,7 +147,7 @@ func Test_OnDownStreamRTCP(t *testing.T) {
fixture.sut.HandleRTCP(livekit.StreamType_DOWNSTREAM, partSID, trackID, pkts)
fixture.sut.SendAnalytics()
//test
// test
require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
_, stats := fixture.analytics.SendStatsArgsForCall(0)
require.Equal(t, 1, len(stats))
@@ -164,13 +164,13 @@ func Test_OnDownStreamRTCP(t *testing.T) {
func Test_PacketLostDiffShouldBeSentToTelemetry(t *testing.T) {
fixture := createFixture()
//prepare
// prepare
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil)
//do
// do
pkts1 := []rtcp.Packet{
&rtcp.ReceiverReport{
Reports: []rtcp.ReceptionReport{
@@ -192,7 +192,7 @@ func Test_PacketLostDiffShouldBeSentToTelemetry(t *testing.T) {
fixture.sut.HandleRTCP(livekit.StreamType_DOWNSTREAM, partSID, trackID, pkts2)
fixture.sut.SendAnalytics()
//test
// test
require.Equal(t, 2, fixture.analytics.SendStatsCallCount()) // 2 calls to fixture.sut.SendAnalytics()
_, stats := fixture.analytics.SendStatsArgsForCall(0)
require.Equal(t, 1, len(stats))
@@ -208,13 +208,13 @@ func Test_PacketLostDiffShouldBeSentToTelemetry(t *testing.T) {
func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) {
fixture := createFixture()
//prepare
// prepare
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil)
//do
// do
pkts1 := []rtcp.Packet{
&rtcp.TransportLayerNack{},
}
@@ -229,7 +229,7 @@ func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) {
fixture.sut.HandleRTCP(livekit.StreamType_DOWNSTREAM, partSID, trackID2, pkts2)
fixture.sut.SendAnalytics()
//test
// test
require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
_, stats := fixture.analytics.SendStatsArgsForCall(0)
require.Equal(t, 2, len(stats))
@@ -253,13 +253,13 @@ func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) {
func Test_OnUpstreamRTCP(t *testing.T) {
fixture := createFixture()
//prepare
// prepare
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil)
//do
// do
pkts := []rtcp.Packet{
&rtcp.TransportLayerNack{},
&rtcp.PictureLossIndication{},
@@ -282,7 +282,7 @@ func Test_OnUpstreamRTCP(t *testing.T) {
fixture.sut.HandleRTCP(livekit.StreamType_UPSTREAM, partSID, trackID, pkts)
fixture.sut.SendAnalytics()
//test
// test
require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
_, stats := fixture.analytics.SendStatsArgsForCall(0)
require.Equal(t, 1, len(stats))
@@ -299,7 +299,7 @@ func Test_OnUpstreamRTCP(t *testing.T) {
func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) {
fixture := createFixture()
//prepare
// prepare
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
@@ -308,27 +308,27 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) {
// there should be bytes reported so that stats are sent
buf := &buffer.Buffer{}
totalBytes := 1
tolalPackets := 1
totalPackets := 1
buf.SetStatsTestOnly(buffer.Stats{
PacketCount: uint32(tolalPackets),
PacketCount: uint32(totalPackets),
TotalByte: uint64(totalBytes),
})
trackID1 := livekit.TrackID("trackID1")
trackID2 := livekit.TrackID("trackID2")
fixture.sut.AddUpTrack(partSID, trackID1, buf)
fixture.sut.AddUpTrack(partSID, trackID2, buf) //using same buffer is not correct but for test it is fine
fixture.sut.AddUpTrack(partSID, trackID2, buf) // using same buffer is not correct but for test it is fine
pkts1 := []rtcp.Packet{
&rtcp.TransportLayerNack{},
}
pkts2 := []rtcp.Packet{
&rtcp.FullIntraRequest{},
}
//do
// do
fixture.sut.HandleRTCP(livekit.StreamType_UPSTREAM, partSID, trackID1, pkts1)
fixture.sut.HandleRTCP(livekit.StreamType_UPSTREAM, partSID, trackID2, pkts2)
fixture.sut.SendAnalytics()
//test
// test
require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
_, stats := fixture.analytics.SendStatsArgsForCall(0)
require.Equal(t, 2, len(stats))
@@ -346,19 +346,19 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) {
require.Equal(t, 1, int(sentStat.FirCount)) // see pkts2 above
}
require.Equal(t, totalBytes, int(sentStat.TotalBytes))
require.Equal(t, tolalPackets, int(sentStat.TotalPackets))
require.Equal(t, totalPackets, int(sentStat.TotalPackets))
}
require.True(t, found1)
require.True(t, found2)
//remove 1 buffer
// remove 1 buffer
fixture.sut.TrackUnpublished(context.Background(), partSID, &livekit.TrackInfo{Sid: string(trackID2)}, 0)
fixture.sut.SendAnalytics()
require.Equal(t, 2, fixture.analytics.SendStatsCallCount())
_, stats = fixture.analytics.SendStatsArgsForCall(1)
require.Equal(t, 2, len(stats)) // still 2 tracks, next call won't contain 1 track
//now only 1 track stats remaining
// now only 1 track stats remaining
fixture.sut.SendAnalytics()
require.Equal(t, 3, fixture.analytics.SendStatsCallCount())
_, stats = fixture.analytics.SendStatsArgsForCall(2)
@@ -368,29 +368,29 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) {
func Test_AnalyticsSentWhenParticipantLeaves(t *testing.T) {
fixture := createFixture()
//prepare
// prepare
room := &livekit.Room{}
partSID := "part1"
participantInfo := &livekit.ParticipantInfo{Sid: partSID}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil)
//do
// do
fixture.sut.ParticipantLeft(context.Background(), room, participantInfo)
//test
// test
require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
}
func Test_AddUpTrack(t *testing.T) {
fixture := createFixture()
//prepare
// prepare
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil)
//do
// do
var totalBytes uint64 = 3
var totalPackets uint32 = 3
buf := &buffer.Buffer{}
@@ -403,7 +403,7 @@ func Test_AddUpTrack(t *testing.T) {
fixture.sut.AddUpTrack(partSID, trackID, buf)
fixture.sut.SendAnalytics()
//test
// test
require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
_, stats := fixture.analytics.SendStatsArgsForCall(0)
require.Equal(t, 1, len(stats))
@@ -416,21 +416,21 @@ func Test_AddUpTrack(t *testing.T) {
func Test_AddUpTrack_SeveralBuffers_Simulcast(t *testing.T) {
fixture := createFixture()
//prepare
// prepare
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil)
//do
// do
trackID := livekit.TrackID("trackID")
//buffer 1
// buffer 1
buf1 := &buffer.Buffer{}
buf1.SetStatsTestOnly(buffer.Stats{
PacketCount: 1,
TotalByte: 1,
})
fixture.sut.AddUpTrack(partSID, trackID, buf1)
//buffer 2
// buffer 2
buf2 := &buffer.Buffer{}
buf2.SetStatsTestOnly(buffer.Stats{
PacketCount: 2,
@@ -438,7 +438,7 @@ func Test_AddUpTrack_SeveralBuffers_Simulcast(t *testing.T) {
})
fixture.sut.AddUpTrack(partSID, trackID, buf2)
fixture.sut.SendAnalytics()
//test
// test
totalBytes := buf1.GetStats().TotalByte + buf2.GetStats().TotalByte
totalPackets := buf1.GetStats().PacketCount + buf2.GetStats().PacketCount
require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
@@ -453,25 +453,25 @@ func Test_AddUpTrack_SeveralBuffers_Simulcast(t *testing.T) {
func Test_BothDownstreamAndUpstreamStatsAreSentTogether(t *testing.T) {
fixture := createFixture()
//prepare
// prepare
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil)
//do
//upstream bytes
// do
// upstream bytes
buf := &buffer.Buffer{}
buf.SetStatsTestOnly(buffer.Stats{
PacketCount: 3,
TotalByte: 3,
})
fixture.sut.AddUpTrack(partSID, "trackID", buf)
//downstream bytes
// downstream bytes
fixture.sut.OnDownstreamPacket(partSID, "trackID1", 1)
fixture.sut.SendAnalytics()
//test
// test
require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
_, stats := fixture.analytics.SendStatsArgsForCall(0)
require.Equal(t, 2, len(stats))
test/client/trackwriter.go
+1 -1
View File
@@ -135,7 +135,7 @@ func (w *TrackWriter) writeOgg() {
}
func (w *TrackWriter) writeVP8() {
// Send our video file frame at a time. Pace our sending so we send it at the same speed it should be played back as.
// Send our video file frame at a time. Pace our sending such that we send it at the same speed it should be played back as.
// This isn't required since the video is timestamped, but we will such much higher loss if we send all at once.
sleepTime := time.Millisecond * time.Duration((float32(w.ivfheader.TimebaseNumerator)/float32(w.ivfheader.TimebaseDenominator))*1000)
for {
test/singlenode_test.go
+5 -4
View File
@@ -8,14 +8,15 @@ import (
"testing"
"time"
"github.com/livekit/livekit-server/pkg/rtc"
"github.com/livekit/livekit-server/pkg/testutils"
testclient "github.com/livekit/livekit-server/test/client"
"github.com/livekit/protocol/auth"
"github.com/livekit/protocol/livekit"
"github.com/pion/webrtc/v3"
"github.com/stretchr/testify/require"
"github.com/thoas/go-funk"
"github.com/livekit/livekit-server/pkg/rtc"
"github.com/livekit/livekit-server/pkg/testutils"
testclient "github.com/livekit/livekit-server/test/client"
)
func TestClientCouldConnect(t *testing.T) {
@@ -177,7 +178,7 @@ func TestSinglePublisher(t *testing.T) {
require.True(t, strings.HasPrefix(tr.ID(), "TR_"), "track should begin with TR")
}
// when c3 disconnects.. ensure subscriber is cleaned up correctly
// when c3 disconnects, ensure subscriber is cleaned up correctly
c3.Stop()
testutils.WithTimeout(t, "c3 is cleaned up as a subscriber", func() bool {