dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-09-01 20:09:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into raja_1833

Browse Source
This commit is contained in:
boks1971
2024-11-05 09:52:17 +05:30
parent 828cb13d77 365e63230d
commit 0297819a31
15 changed files with 76 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -28,7 +28,7 @@ require (
github.com/olekukonko/tablewriter v0.0.5
github.com/ory/dockertest/v3 v3.11.0
github.com/pion/dtls/v2 v2.2.12
github.com/pion/ice/v2 v2.3.36
github.com/pion/ice/v2 v2.3.37
github.com/pion/interceptor v0.1.37
github.com/pion/rtcp v1.2.14
github.com/pion/rtp v1.8.9
go.sum
+2 -2
View File
@@ -233,8 +233,8 @@ github.com/pion/datachannel v1.5.9/go.mod h1:kDUuk4CU4Uxp82NH4LQZbISULkX/HtzKa4P
github.com/pion/dtls/v2 v2.2.7/go.mod h1:8WiMkebSHFD0T+dIU+UeBaoV7kDhOW5oDCzZ7WZ/F9s=
github.com/pion/dtls/v2 v2.2.12 h1:KP7H5/c1EiVAAKUmXyCzPiQe5+bCJrpOeKg/L05dunk=
github.com/pion/dtls/v2 v2.2.12/go.mod h1:d9SYc9fch0CqK90mRk1dC7AkzzpwJj6u2GU3u+9pqFE=
github.com/pion/ice/v2 v2.3.36 h1:SopeXiVbbcooUg2EIR8sq4b13RQ8gzrkkldOVg+bBsc=
github.com/pion/ice/v2 v2.3.36/go.mod h1:mBF7lnigdqgtB+YHkaY/Y6s6tsyRyo4u4rPGRuOjUBQ=
github.com/pion/ice/v2 v2.3.37 h1:ObIdaNDu1rCo7hObhs34YSBcO7fjslJMZV0ux+uZWh0=
github.com/pion/ice/v2 v2.3.37/go.mod h1:mBF7lnigdqgtB+YHkaY/Y6s6tsyRyo4u4rPGRuOjUBQ=
github.com/pion/interceptor v0.1.37 h1:aRA8Zpab/wE7/c0O3fh1PqY0AJI3fCSEM5lRWJVorwI=
github.com/pion/interceptor v0.1.37/go.mod h1:JzxbJ4umVTlZAf+/utHzNesY8tmRkM2lVmkS82TTj8Y=
github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY=
pkg/config/configtest/checkyamltag.go
+5
View File
@@ -31,6 +31,11 @@ func checkYAMLTags(t reflect.Type, seen map[reflect.Type]struct{}) error {
for i := 0; i < t.NumField(); i++ {
field := t.Field(i)
if !field.IsExported() {
// ignore unexported fields
continue
}
if field.Type.Kind() == reflect.Bool {
// ignore boolean fields
continue
pkg/routing/roommanager.go
-2
View File
@@ -24,8 +24,6 @@ import (
"github.com/livekit/psrpc/pkg/middleware"
)
//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate
//counterfeiter:generate . RoomManagerClient
type RoomManagerClient interface {
rpc.TypedRoomManagerClient
pkg/routing/signal.go
-2
View File
@@ -36,8 +36,6 @@ import (
var ErrSignalWriteFailed = errors.New("signal write failed")
var ErrSignalMessageDropped = errors.New("signal message dropped")
//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate
//counterfeiter:generate . SignalClient
type SignalClient interface {
ActiveCount() int
pkg/rtc/participant.go
+34 -8
View File
@@ -158,7 +158,6 @@ type ParticipantParams struct {
ForwardStats *sfu.ForwardStats
DisableSenderReportPassThrough bool
MetricConfig metric.MetricConfig
DropRemoteICECandidates bool
}
type ParticipantImpl struct {
@@ -964,7 +963,12 @@ func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseRea
p.clearMigrationTimer()
if sendLeave {
p.sendLeaveRequest(reason, isExpectedToResume, false, false)
p.sendLeaveRequest(
reason,
isExpectedToResume,
false, // isExpectedToReconnect
false, // sendOnlyIfSupportingLeaveRequestWithAction
)
}
if p.supervisor != nil {
@@ -1071,7 +1075,12 @@ func (p *ParticipantImpl) MaybeStartMigration(force bool, onStart func()) bool {
onStart()
}
p.sendLeaveRequest(types.ParticipantCloseReasonMigrationRequested, true, false, true)
p.sendLeaveRequest(
types.ParticipantCloseReasonMigrationRequested,
true, // isExpectedToResume
false, // isExpectedToReconnect
true, // sendOnlyIfSupportingLeaveRequestWithAction
)
p.CloseSignalConnection(types.SignallingCloseReasonMigration)
p.clearMigrationTimer()
@@ -1449,7 +1458,6 @@ func (p *ParticipantImpl) setupTransportManager() error {
PublisherHandler: pth,
SubscriberHandler: sth,
DataChannelStats: p.dataChannelStats,
DropRemoteICECandidates: p.params.DropRemoteICECandidates,
}
if p.params.SyncStreams && p.params.PlayoutDelay.GetEnabled() && p.params.ClientInfo.isFirefox() {
// we will disable playout delay for Firefox if the user is expecting
@@ -1820,8 +1828,14 @@ func (p *ParticipantImpl) setupDisconnectTimer() {
}
func (p *ParticipantImpl) onAnyTransportFailed() {
// clients support resuming of connections when websocket becomes disconnected
p.sendLeaveRequest(types.ParticipantCloseReasonPeerConnectionDisconnected, true, false, true)
p.sendLeaveRequest(
types.ParticipantCloseReasonPeerConnectionDisconnected,
true, // isExpectedToResume
false, // isExpectedToReconnect
true, // sendOnlyIfSupportingLeaveRequestWithAction
)
// clients support resuming of connections when signalling becomes disconnected
p.CloseSignalConnection(types.SignallingCloseReasonTransportFailure)
// detect when participant has actually left.
@@ -2658,7 +2672,12 @@ func (p *ParticipantImpl) GetCachedDownTrack(trackID livekit.TrackID) (*webrtc.R
}
func (p *ParticipantImpl) IssueFullReconnect(reason types.ParticipantCloseReason) {
p.sendLeaveRequest(reason, false, true, false)
p.sendLeaveRequest(
reason,
false, // isExpectedToResume
true, // isExpectedToReconnect
false, // sendOnlyIfSupportingLeaveRequestWithAction
)
scr := types.SignallingCloseReasonUnknown
switch reason {
@@ -2807,7 +2826,14 @@ func (p *ParticipantImpl) setupEnabledCodecs(publishEnabledCodecs []*livekit.Cod
}
func (p *ParticipantImpl) GetEnabledPublishCodecs() []*livekit.Codec {
return p.enabledPublishCodecs
codecs := make([]*livekit.Codec, 0, len(p.enabledPublishCodecs))
for _, c := range p.enabledPublishCodecs {
if c.Mime == "video/rtx" {
continue
}
codecs = append(codecs, c)
}
return codecs
}
func (p *ParticipantImpl) UpdateAudioTrack(update *livekit.UpdateLocalAudioTrack) error {
pkg/rtc/room.go
+4 -21
View File
@@ -18,7 +18,6 @@ import (
"context"
"fmt"
"math"
"net"
"slices"
"sort"
"strings"
@@ -55,8 +54,6 @@ const (
dataForwardLoadBalanceThreshold = 20
simulateDisconnectSignalTimeout = 5 * time.Second
minIPTruncateLen = 8
)
var (
@@ -1840,25 +1837,11 @@ func connectionDetailsFields(infos []*types.ICEConnectionInfo) []interface{} {
if c.Trickle {
cStr += "[trickle]"
}
remoteAddress := c.Remote.Address()
ipAddr := net.ParseIP(remoteAddress)
isPrivate := false
if ipAddr != nil {
isPrivate = ipAddr.IsPrivate()
}
if !isPrivate && len(remoteAddress) > minIPTruncateLen {
remoteAddress = remoteAddress[:len(remoteAddress)-3] + "..."
}
cStr += " " + fmt.Sprintf("%s %s %s:%d", c.Remote.NetworkType(), c.Remote.Type(), remoteAddress, c.Remote.Port())
cStr += " " + fmt.Sprintf("%s %s %s:%d", c.Remote.NetworkType(), c.Remote.Type(), MaybeTruncateIP(c.Remote.Address()), c.Remote.Port())
if relatedAddress := c.Remote.RelatedAddress(); relatedAddress != nil {
ipAddr = net.ParseIP(relatedAddress.Address)
if ipAddr != nil {
isPrivate = ipAddr.IsPrivate()
relatedAddressAddress := relatedAddress.Address
if !isPrivate && len(relatedAddressAddress) > minIPTruncateLen {
relatedAddressAddress = relatedAddressAddress[:len(relatedAddressAddress)-3] + "..."
}
cStr += " " + fmt.Sprintf(" related %s:%d", relatedAddressAddress, relatedAddress.Port)
relatedAddr := MaybeTruncateIP(relatedAddress.Address)
if relatedAddr != "" {
cStr += " " + fmt.Sprintf(" related %s:%d", relatedAddr, relatedAddress.Port)
}
}
candidates = append(candidates, cStr)
pkg/rtc/transport.go
+10 -37
View File
@@ -150,10 +150,10 @@ func (w wrappedICECandidatePairLogger) MarshalLogObject(e zapcore.ObjectEncoder)
if w.pair.Remote != nil {
e.AddString("remoteProtocol", w.pair.Remote.Protocol.String())
e.AddString("remoteCandidateType", w.pair.Remote.Typ.String())
e.AddString("remoteAdddress", w.pair.Remote.Address[:len(w.pair.Remote.Address)-3]+"...")
e.AddString("remoteAdddress", MaybeTruncateIP(w.pair.Remote.Address))
e.AddUint16("remotePort", w.pair.Remote.Port)
if w.pair.Remote.RelatedAddress != "" {
e.AddString("relatedAdddress", w.pair.Remote.RelatedAddress[:len(w.pair.Remote.RelatedAddress)-3]+"...")
e.AddString("relatedAdddress", MaybeTruncateIP(w.pair.Remote.RelatedAddress))
e.AddUint16("relatedPort", w.pair.Remote.RelatedPort)
}
}
@@ -254,7 +254,6 @@ type TransportParams struct {
IsSendSide bool
AllowPlayoutDelay bool
DataChannelMaxBufferedAmount uint64
DropRemoteICECandidates bool
}
func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimator cc.BandwidthEstimator)) (*webrtc.PeerConnection, *webrtc.MediaEngine, error) {
@@ -299,9 +298,12 @@ func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimat
se.DisableSRTCPReplayProtection(true)
if !params.ProtocolVersion.SupportsICELite() || !params.ClientInfo.SupportPrflxOverRelay() {
// if client don't support prflx over relay which is only Firefox, disable ICE Lite to ensure that
// dropping remote ICE candidates does not get enabled. Firefox does aggressive nomination and
// dropping remote ICE candidates means server would accept all switches and it could end up with
// the lower priority candidate. As Firefox does not support migration, ICE Lite can be disabled.
// aggressive nomination is handled properly. Firefox does aggressive nomination even if peer is
// ICE Lite (see comment as to historical reasons: https://github.com/pion/ice/pull/739#issuecomment-2452245066).
// pion/ice (as of v2.3.37) will accept all use-candidate switches when in ICE Lite mode.
// That combined with aggressive nomination from Firefox could potentially lead to the two ends
// ending up with different candidates.
// As Firefox does not support migration, ICE Lite can be disabled.
se.SetLite(false)
}
se.SetDTLSRetransmissionInterval(dtlsRetransmissionInterval)
@@ -1415,7 +1417,7 @@ func (t *PCTransport) handleRemoteICECandidate(e event) error {
c := e.data.(*webrtc.ICECandidateInit)
filtered := false
if t.preferTCP.Load() && !strings.Contains(c.Candidate, "tcp") {
if t.preferTCP.Load() && !strings.Contains(strings.ToLower(c.Candidate), "tcp") {
t.params.Logger.Debugw("filtering out remote candidate", "candidate", c.Candidate)
filtered = true
}
@@ -1435,12 +1437,6 @@ func (t *PCTransport) handleRemoteICECandidate(e event) error {
return nil
}
if t.params.DropRemoteICECandidates {
t.params.Logger.Debugw("dropping remote ICE candidate", "candidate", c.Candidate)
t.connectionDetails.AddRemoteCandidate(*c, true, true, true)
return nil
}
if err := t.pc.AddICECandidate(*c); err != nil {
t.params.Logger.Warnw("failed to add cached ICE candidate", err, "candidate", c)
return errors.Wrap(err, "add ice candidate failed")
@@ -1465,25 +1461,6 @@ func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP,
return sd
}
_, iceLite := parsed.Attribute("ice-lite")
var liteSet bool
if isLocal {
if t.localICEIsLite == nil {
t.localICEIsLite = &iceLite
liteSet = true
}
} else {
if t.remoteICEIsLite == nil {
t.remoteICEIsLite = &iceLite
liteSet = true
}
}
if liteSet && t.localICEIsLite != nil && t.remoteICEIsLite != nil {
// only drop remote candidates if local is lite and remote is not
t.params.DropRemoteICECandidates = t.params.DropRemoteICECandidates && (*t.localICEIsLite && !*t.remoteICEIsLite)
t.params.Logger.Debugw("setting DropRemoteICECandidates", "dropRemoteCandidate", t.params.DropRemoteICECandidates, "localICELite", *t.localICEIsLite, "remoteICELite", *t.remoteICEIsLite)
}
filterAttributes := func(attrs []sdp.Attribute) []sdp.Attribute {
filteredAttrs := make([]sdp.Attribute, 0, len(attrs))
for _, a := range attrs {
@@ -1494,7 +1471,7 @@ func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP,
filteredAttrs = append(filteredAttrs, a)
continue
}
excluded := (!isLocal && t.params.DropRemoteICECandidates) || (preferTCP && !c.NetworkType().IsTCP())
excluded := preferTCP && !c.NetworkType().IsTCP()
if !excluded {
if !t.params.Config.UseMDNS && types.IsICECandidateMDNS(c) {
excluded = true
@@ -1712,10 +1689,6 @@ func (t *PCTransport) setRemoteDescription(sd webrtc.SessionDescription) error {
}
for _, c := range t.pendingRemoteCandidates {
if t.params.DropRemoteICECandidates {
t.connectionDetails.AddRemoteCandidate(*c, true, true, true)
continue
}
if err := t.pc.AddICECandidate(*c); err != nil {
t.params.Logger.Warnw("failed to add cached ICE candidate", err, "candidate", c)
return errors.Wrap(err, "add ice candidate failed")
pkg/rtc/transport_test.go
-110
View File
@@ -28,7 +28,6 @@ import (
"github.com/livekit/livekit-server/pkg/rtc/transport"
"github.com/livekit/livekit-server/pkg/rtc/transport/transportfakes"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/pkg/testutils"
"github.com/livekit/protocol/livekit"
)
@@ -504,115 +503,6 @@ func TestFilteringCandidates(t *testing.T) {
transport.Close()
}
func TestDropRemoteICECandidates(t *testing.T) {
cases := []struct {
name string
remoteLite bool
localLite bool
expecteLocalDrop bool
expecteRemoteDrop bool
}{
{
name: "both not lite",
localLite: false,
remoteLite: false,
expecteLocalDrop: false,
expecteRemoteDrop: false,
},
{
name: "remote lite",
localLite: false,
remoteLite: true,
expecteLocalDrop: false,
expecteRemoteDrop: true,
},
{
name: "local lite",
localLite: true,
remoteLite: false,
expecteLocalDrop: true,
expecteRemoteDrop: false,
},
{
name: "both lite",
localLite: true,
remoteLite: true,
expecteLocalDrop: false,
expecteRemoteDrop: false,
},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
params := TransportParams{
ParticipantID: "id",
ParticipantIdentity: "identity",
Config: &WebRTCConfig{},
IsOfferer: true,
ProtocolVersion: types.CurrentProtocol,
DropRemoteICECandidates: true,
}
paramsA := params
paramsA.Config.SettingEngine.SetLite(c.localLite)
handlerA := &transportfakes.FakeHandler{}
paramsA.Handler = handlerA
transportLocal, err := NewPCTransport(paramsA)
require.NoError(t, err)
_, err = transportLocal.pc.CreateDataChannel(LossyDataChannel, nil)
require.NoError(t, err)
paramsB := params
paramsB.Config.SettingEngine.SetLite(c.remoteLite)
handlerB := &transportfakes.FakeHandler{}
paramsB.Handler = handlerB
paramsB.IsOfferer = false
transportRemote, err := NewPCTransport(paramsB)
require.NoError(t, err)
require.False(t, transportLocal.IsEstablished())
require.False(t, transportRemote.IsEstablished())
handleICEExchange(t, transportLocal, transportRemote, handlerA, handlerB)
var offer atomic.Pointer[webrtc.SessionDescription]
handlerA.OnOfferCalls(func(sd webrtc.SessionDescription) error {
parsed, err := sd.Unmarshal()
require.NoError(t, err)
_, lite := parsed.Attribute("ice-lite")
require.Equal(t, c.localLite, lite)
offer.Store(&sd)
return nil
})
transportLocal.Negotiate(true)
require.Eventually(t, func() bool {
return offer.Load() != nil
}, 100*time.Millisecond, time.Millisecond*10, "offer not received")
handlerB.OnAnswerCalls(func(sd webrtc.SessionDescription) error {
parsed, err := sd.Unmarshal()
require.NoError(t, err)
_, lite := parsed.Attribute("ice-lite")
require.Equal(t, c.remoteLite, lite, sd.SDP)
transportLocal.HandleRemoteDescription(sd)
return nil
})
transportRemote.HandleRemoteDescription(*offer.Load())
require.Eventually(t, func() bool {
return transportLocal.IsEstablished()
}, 10*time.Second, time.Millisecond*10, "transportA is not established")
require.Eventually(t, func() bool {
return transportRemote.IsEstablished()
}, 10*time.Second, time.Millisecond*10, "transportB is not established")
require.Equal(t, c.expecteLocalDrop, transportLocal.params.DropRemoteICECandidates)
require.Equal(t, c.expecteRemoteDrop, transportRemote.params.DropRemoteICECandidates)
transportLocal.Close()
transportRemote.Close()
})
}
}
func handleICEExchange(t *testing.T, a, b *PCTransport, ah, bh *transportfakes.FakeHandler) {
ah.OnICECandidateCalls(func(candidate *webrtc.ICECandidate, target livekit.SignalTarget) error {
pkg/rtc/transportmanager.go
-3
View File
@@ -103,7 +103,6 @@ type TransportManagerParams struct {
PublisherHandler transport.Handler
SubscriberHandler transport.Handler
DataChannelStats *telemetry.BytesTrackStats
DropRemoteICECandidates bool
}
type TransportManager struct {
@@ -158,7 +157,6 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
ClientInfo: params.ClientInfo,
Transport: livekit.SignalTarget_PUBLISHER,
Handler: TransportManagerPublisherTransportHandler{TransportManagerTransportHandler{params.PublisherHandler, t, lgr}},
DropRemoteICECandidates: params.DropRemoteICECandidates,
})
if err != nil {
return nil, err
@@ -182,7 +180,6 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
DataChannelMaxBufferedAmount: params.DataChannelMaxBufferedAmount,
Transport: livekit.SignalTarget_SUBSCRIBER,
Handler: TransportManagerTransportHandler{params.SubscriberHandler, t, lgr},
DropRemoteICECandidates: params.DropRemoteICECandidates,
})
if err != nil {
return nil, err
pkg/rtc/utils.go
+16
View File
@@ -18,6 +18,7 @@ import (
"encoding/json"
"errors"
"io"
"net"
"strings"
"github.com/pion/webrtc/v3"
@@ -28,6 +29,8 @@ import (
const (
trackIdSeparator = "|"
cMinIPTruncateLen = 8
)
func UnpackStreamID(packed string) (participantID livekit.ParticipantID, trackID livekit.TrackID) {
@@ -198,3 +201,16 @@ func LoggerWithCodecMime(l logger.Logger, mime string) logger.Logger {
}
return l
}
func MaybeTruncateIP(addr string) string {
ipAddr := net.ParseIP(addr)
if ipAddr == nil {
return ""
}
if ipAddr.IsPrivate() || len(addr) <= cMinIPTruncateLen {
return addr
}
return addr[:len(addr)-3] + "..."
}
pkg/service/clients.go
-2
View File
@@ -27,8 +27,6 @@ import (
"github.com/livekit/protocol/utils/guid"
)
//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate
//counterfeiter:generate . IOClient
type IOClient interface {
CreateEgress(ctx context.Context, info *livekit.EgressInfo) (*emptypb.Empty, error)
pkg/service/signal.go
-2
View File
@@ -31,8 +31,6 @@ import (
"github.com/livekit/psrpc/pkg/middleware"
)
//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate
//counterfeiter:generate . SessionHandler
type SessionHandler interface {
Logger(ctx context.Context) logger.Logger
pkg/telemetry/analyticsservice.go
+1 -1
View File
@@ -29,7 +29,7 @@ import (
"github.com/livekit/livekit-server/pkg/routing"
)
//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 . AnalyticsService
//counterfeiter:generate . AnalyticsService
type AnalyticsService interface {
SendStats(ctx context.Context, stats []*livekit.AnalyticsStat)
SendEvent(ctx context.Context, events *livekit.AnalyticsEvent)
pkg/telemetry/telemetryservice.go
+3 -1
View File
@@ -26,7 +26,9 @@ import (
"github.com/livekit/protocol/webhook"
)
//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 . TelemetryService
//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate
//counterfeiter:generate . TelemetryService
type TelemetryService interface {
// TrackStats is called periodically for each track in both directions (published/subscribed)
TrackStats(key StatsKey, stat *livekit.AnalyticsStat)