From 5bea9debb7a52d6b45755888002e7cef60cd6b9f Mon Sep 17 00:00:00 2001 From: David Colburn Date: Wed, 19 Jan 2022 02:13:06 -0800 Subject: [PATCH] Code cleanup (#353) --- cmd/server/main_test.go | 12 ++- pkg/config/ip.go | 4 +- pkg/logger/logger.go | 2 +- pkg/rtc/config.go | 2 +- pkg/rtc/mediatrack.go | 16 ++-- pkg/rtc/mediatrack_test.go | 40 ++++---- pkg/rtc/mediatrackreceiver.go | 2 +- pkg/rtc/mediatracksubscriptions.go | 4 +- pkg/rtc/participant.go | 40 ++++---- pkg/rtc/participant_internal_test.go | 24 ++--- pkg/rtc/transport.go | 10 +- pkg/rtc/types/interfaces.go | 2 +- pkg/rtc/unhandlesimulcast.go | 4 +- pkg/rtc/uptrackmanager.go | 62 ++++++------- pkg/rtc/uptrackmanager_test.go | 7 +- pkg/rtc/wrappedreceiver.go | 3 +- pkg/service/redisroomstore_test.go | 2 +- pkg/sfu/buffer/buffer.go | 2 +- pkg/sfu/connectionquality/mos.go | 2 +- pkg/sfu/downtrack.go | 9 +- pkg/sfu/forwarder.go | 4 +- pkg/sfu/forwarder_test.go | 16 ++-- pkg/sfu/prober.go | 10 +- pkg/sfu/receiver.go | 20 ++-- pkg/sfu/sequencer.go | 2 +- pkg/sfu/sequencer_test.go | 2 +- pkg/sfu/streamallocator.go | 7 +- pkg/sfu/streamtracker.go | 4 +- pkg/sfu/streamtracker_test.go | 2 +- pkg/telemetry/rtt.go | 4 +- pkg/telemetry/statsworker.go | 6 +- .../test/telemetry_service_events_test.go | 12 +-- pkg/telemetry/test/telemetry_service_test.go | 92 +++++++++---------- test/client/trackwriter.go | 2 +- test/singlenode_test.go | 9 +- 35 files changed, 223 insertions(+), 218 deletions(-) diff --git a/cmd/server/main_test.go b/cmd/server/main_test.go index 3ec968db8..97a4b8cf0 100644 --- a/cmd/server/main_test.go +++ b/cmd/server/main_test.go @@ -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) + }() } } diff --git a/pkg/config/ip.go b/pkg/config/ip.go index ec41211c2..a33216f2c 100644 --- a/pkg/config/ip.go +++ b/pkg/config/ip.go @@ -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 { diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go index 248a76747..4514320b5 100644 --- a/pkg/logger/logger.go +++ b/pkg/logger/logger.go @@ -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" ) diff --git a/pkg/rtc/config.go b/pkg/rtc/config.go index ab989c7ea..5597349e6 100644 --- a/pkg/rtc/config.go +++ b/pkg/rtc/config.go @@ -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" ) diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index f2361cf61..00f1198d8 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -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 { diff --git a/pkg/rtc/mediatrack_test.go b/pkg/rtc/mediatrack_test.go index d22330f5c..3cfc441f0 100644 --- a/pkg/rtc/mediatrack_test.go +++ b/pkg/rtc/mediatrack_test.go @@ -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) diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index 0a75d6942..5b44f1d04 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -360,7 +360,7 @@ func (t *MediaTrackReceiver) OnSubscribedMaxQualityChange(f func(trackID livekit }) } -//--------------------------- +// --------------------------- func SpatialLayerForQuality(quality livekit.VideoQuality) int32 { switch quality { diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index d8c9133b9..ba72d5b5d 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -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 diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index d45cf6343..a0b35015e 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -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() diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 1fe5144bc..aa1f91198 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -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()) }) diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index eb553a443..56c410039 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -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) diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 384e7acdd..126d5d413 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -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 diff --git a/pkg/rtc/unhandlesimulcast.go b/pkg/rtc/unhandlesimulcast.go index 877f57b32..0fd611443 100644 --- a/pkg/rtc/unhandlesimulcast.go +++ b/pkg/rtc/unhandlesimulcast.go @@ -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 } diff --git a/pkg/rtc/uptrackmanager.go b/pkg/rtc/uptrackmanager.go index 61ca2c55d..37d68f34c 100644 --- a/pkg/rtc/uptrackmanager.go +++ b/pkg/rtc/uptrackmanager.go @@ -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{}) diff --git a/pkg/rtc/uptrackmanager_test.go b/pkg/rtc/uptrackmanager_test.go index f40af8281..e903c3dfd 100644 --- a/pkg/rtc/uptrackmanager_test.go +++ b/pkg/rtc/uptrackmanager_test.go @@ -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") diff --git a/pkg/rtc/wrappedreceiver.go b/pkg/rtc/wrappedreceiver.go index b5ef59adc..a1fd3ad98 100644 --- a/pkg/rtc/wrappedreceiver.go +++ b/pkg/rtc/wrappedreceiver.go @@ -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 diff --git a/pkg/service/redisroomstore_test.go b/pkg/service/redisroomstore_test.go index a3138ab12..c21be1abf 100644 --- a/pkg/service/redisroomstore_test.go +++ b/pkg/service/redisroomstore_test.go @@ -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) diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 9aaef37fb..714132bbc 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -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() diff --git a/pkg/sfu/connectionquality/mos.go b/pkg/sfu/connectionquality/mos.go index 8c5fb0f01..d7a35b701 100644 --- a/pkg/sfu/connectionquality/mos.go +++ b/pkg/sfu/connectionquality/mos.go @@ -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 { diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index a0614b486..51bbfc4bd 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -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) diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index e2d283384..804f1ab1b 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -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) diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 1b5751376..2c3393db6 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -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, diff --git a/pkg/sfu/prober.go b/pkg/sfu/prober.go index cfc4f553d..63d39cce0 100644 --- a/pkg/sfu/prober.go +++ b/pkg/sfu/prober.go @@ -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 diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index c16a42b7e..c138af68d 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -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") } } diff --git a/pkg/sfu/sequencer.go b/pkg/sfu/sequencer.go index c8b9907cc..4f26923ad 100644 --- a/pkg/sfu/sequencer.go +++ b/pkg/sfu/sequencer.go @@ -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 } diff --git a/pkg/sfu/sequencer_test.go b/pkg/sfu/sequencer_test.go index 9896108e0..01a5b6213 100644 --- a/pkg/sfu/sequencer_test.go +++ b/pkg/sfu/sequencer_test.go @@ -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, diff --git a/pkg/sfu/streamallocator.go b/pkg/sfu/streamallocator.go index 58a79a265..017df97b3 100644 --- a/pkg/sfu/streamallocator.go +++ b/pkg/sfu/streamallocator.go @@ -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. diff --git a/pkg/sfu/streamtracker.go b/pkg/sfu/streamtracker.go index 7ca54b88b..10ec76b94 100644 --- a/pkg/sfu/streamtracker.go +++ b/pkg/sfu/streamtracker.go @@ -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 diff --git a/pkg/sfu/streamtracker_test.go b/pkg/sfu/streamtracker_test.go index 6ad4372ef..3b6f45e75 100644 --- a/pkg/sfu/streamtracker_test.go +++ b/pkg/sfu/streamtracker_test.go @@ -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) diff --git a/pkg/telemetry/rtt.go b/pkg/telemetry/rtt.go index f385e13e2..387f88be1 100644 --- a/pkg/telemetry/rtt.go +++ b/pkg/telemetry/rtt.go @@ -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. diff --git a/pkg/telemetry/statsworker.go b/pkg/telemetry/statsworker.go index e43d412c2..4d8b477f3 100644 --- a/pkg/telemetry/statsworker.go +++ b/pkg/telemetry/statsworker.go @@ -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 } diff --git a/pkg/telemetry/test/telemetry_service_events_test.go b/pkg/telemetry/test/telemetry_service_events_test.go index 48cf297a7..2f064145f 100644 --- a/pkg/telemetry/test/telemetry_service_events_test.go +++ b/pkg/telemetry/test/telemetry_service_events_test.go @@ -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) diff --git a/pkg/telemetry/test/telemetry_service_test.go b/pkg/telemetry/test/telemetry_service_test.go index c2de65d05..2ad50ee46 100644 --- a/pkg/telemetry/test/telemetry_service_test.go +++ b/pkg/telemetry/test/telemetry_service_test.go @@ -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)) diff --git a/test/client/trackwriter.go b/test/client/trackwriter.go index df9865779..c4939d212 100644 --- a/test/client/trackwriter.go +++ b/test/client/trackwriter.go @@ -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 { diff --git a/test/singlenode_test.go b/test/singlenode_test.go index 431ce4e39..37337b60c 100644 --- a/test/singlenode_test.go +++ b/test/singlenode_test.go @@ -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 {