dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-24 13:59:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Support rtx for publisher (#2452)

Browse Source 
* Support rtx for publisher

* remote log

* solve comment
This commit is contained in:
cnderrauber
2024-02-06 21:30:37 +08:00
committed by GitHub
parent 716844c383
commit be87a1b6f0
15 changed files with 453 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+4 -4
View File
@@ -17,7 +17,7 @@ require (
github.com/hashicorp/golang-lru/v2 v2.0.7
github.com/jxskiss/base62 v1.1.0
github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1
github.com/livekit/mediatransportutil v0.0.0-20231213075826-cccbf2b93d3f
github.com/livekit/mediatransportutil v0.0.0-20240206022904-8e203f9db6af
github.com/livekit/protocol v1.9.8
github.com/livekit/psrpc v0.5.3-0.20240129223932-473b29cda289
github.com/mackerelio/go-osstat v0.2.4
@@ -34,7 +34,7 @@ require (
github.com/pion/sdp/v3 v3.0.6
github.com/pion/transport/v2 v2.2.4
github.com/pion/turn/v2 v2.1.4
github.com/pion/webrtc/v3 v3.2.24
github.com/pion/webrtc/v3 v3.2.25
github.com/pkg/errors v0.9.1
github.com/prometheus/client_golang v1.18.0
github.com/redis/go-redis/v9 v9.4.0
@@ -65,7 +65,7 @@ require (
github.com/golang/protobuf v1.5.3 // indirect
github.com/google/go-cmp v0.6.0 // indirect
github.com/google/subcommands v1.2.0 // indirect
github.com/google/uuid v1.5.0 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
github.com/hashicorp/go-retryablehttp v0.7.5 // indirect
github.com/hashicorp/golang-lru v0.5.4 // indirect
@@ -82,7 +82,7 @@ require (
github.com/nats-io/nuid v1.0.1 // indirect
github.com/pion/datachannel v1.5.5 // indirect
github.com/pion/logging v0.2.2 // indirect
github.com/pion/mdns v0.0.9 // indirect
github.com/pion/mdns v0.0.10 // indirect
github.com/pion/randutil v0.1.0 // indirect
github.com/pion/srtp/v2 v2.0.18 // indirect
github.com/pion/stun v0.6.1 // indirect
go.sum
+8 -13
View File
@@ -73,8 +73,8 @@ github.com/google/subcommands v1.2.0 h1:vWQspBTo2nEqTUFita5/KeEWlUL8kQObDFbub/EN
github.com/google/subcommands v1.2.0/go.mod h1:ZjhPrFU+Olkh9WazFPsl27BQ4UPiG37m3yTrtFlrHVk=
github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/wire v0.5.0 h1:I7ELFeVBr3yfPIcc8+MWvrjk+3VjbcSzoXm3JVa+jD8=
github.com/google/wire v0.5.0/go.mod h1:ngWDr9Qvq3yZA10YrxfyGELY/AFWGVpy9c1LTRi1EoU=
github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY=
@@ -124,8 +124,8 @@ github.com/lithammer/shortuuid/v4 v4.0.0 h1:QRbbVkfgNippHOS8PXDkti4NaWeyYfcBTHtw
github.com/lithammer/shortuuid/v4 v4.0.0/go.mod h1:Zs8puNcrvf2rV9rTH51ZLLcj7ZXqQI3lv67aw4KiB1Y=
github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkDaKb5iXdynYrzB84ErPPO4LbRASk58=
github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
github.com/livekit/mediatransportutil v0.0.0-20231213075826-cccbf2b93d3f h1:XHrwGwLNGQB3ZqolH1YdMH/22hgXKr4vm+2M7JKMMGg=
github.com/livekit/mediatransportutil v0.0.0-20231213075826-cccbf2b93d3f/go.mod h1:GBzn9xL+mivI1pW+tyExcKgbc0VOc29I9yJsNcAVaAc=
github.com/livekit/mediatransportutil v0.0.0-20240206022904-8e203f9db6af h1:7VBmJFAqCRCCJxHG7c8z5LGEbwUGqzzw1weZgAewmLA=
github.com/livekit/mediatransportutil v0.0.0-20240206022904-8e203f9db6af/go.mod h1:GBzn9xL+mivI1pW+tyExcKgbc0VOc29I9yJsNcAVaAc=
github.com/livekit/protocol v1.9.8 h1:CyK+oQ2TgSkRGgkJfDlJd5FRP42cahEFCOyxg2Sa/ts=
github.com/livekit/protocol v1.9.8/go.mod h1:lSJlMeTJfQBEv8/D2p3zdCo+i+jTmTtn24ysL4ePK28=
github.com/livekit/psrpc v0.5.3-0.20240129223932-473b29cda289 h1:oTgNH7v9TXsBgoltKk5mnWjv4qqcPF2iV+WtEVQ6ROM=
@@ -192,8 +192,8 @@ github.com/pion/interceptor v0.1.25/go.mod h1:wkbPYAak5zKsfpVDYMtEfWEy8D4zL+rpxC
github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY=
github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms=
github.com/pion/mdns v0.0.8/go.mod h1:hYE72WX8WDveIhg7fmXgMKivD3Puklk0Ymzog0lSyaI=
github.com/pion/mdns v0.0.9 h1:7Ue5KZsqq8EuqStnpPWV33vYYEH0+skdDN5L7EiEsI4=
github.com/pion/mdns v0.0.9/go.mod h1:2JA5exfxwzXiCihmxpTKgFUpiQws2MnipoPK09vecIc=
github.com/pion/mdns v0.0.10 h1:u9/12WL2NNgtGT2nNPXT6+A+xeOF0PkawM/S/wPMWQA=
github.com/pion/mdns v0.0.10/go.mod h1:Y1scL/8TT8KQ172UfxrE4j0c04NOY71bJS1aE1zvyGY=
github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA=
github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8=
github.com/pion/rtcp v1.2.10/go.mod h1:ztfEwXZNLGyF1oQDttz/ZKIBaeeg/oWbRYqzBM9TL1I=
@@ -225,8 +225,8 @@ github.com/pion/transport/v3 v3.0.1/go.mod h1:UY7kiITrlMv7/IKgd5eTUcaahZx5oUN3l9
github.com/pion/turn/v2 v2.1.3/go.mod h1:huEpByKKHix2/b9kmTAM3YoX6MKP+/D//0ClgUYR2fY=
github.com/pion/turn/v2 v2.1.4 h1:2xn8rduI5W6sCZQkEnIUDAkrBQNl2eYIBCHMZ3QMmP8=
github.com/pion/turn/v2 v2.1.4/go.mod h1:huEpByKKHix2/b9kmTAM3YoX6MKP+/D//0ClgUYR2fY=
github.com/pion/webrtc/v3 v3.2.24 h1:MiFL5DMo2bDaaIFWr0DDpwiV/L4EGbLZb+xoRvfEo1Y=
github.com/pion/webrtc/v3 v3.2.24/go.mod h1:1CaT2fcZzZ6VZA+O1i9yK2DU4EOcXVvSbWG9pr5jefs=
github.com/pion/webrtc/v3 v3.2.25 h1:hmewKZtY60TOMOnAfEfHI+D78VpDe3OdnwcRpE8kLEc=
github.com/pion/webrtc/v3 v3.2.25/go.mod h1:1CaT2fcZzZ6VZA+O1i9yK2DU4EOcXVvSbWG9pr5jefs=
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -295,7 +295,6 @@ golang.org/x/crypto v0.8.0/go.mod h1:mRqEX+O9/h5TFCrQhkgjo2yKi0yYA+9ecGkdQoHrywE
golang.org/x/crypto v0.10.0/go.mod h1:o4eNf7Ede1fv+hwOwZsTHl9EsPFO6q6ZvYR8vYfY45I=
golang.org/x/crypto v0.11.0/go.mod h1:xgJhtzW8F9jGdVFWZESrid1U1bjeNy4zgy5cRr/CIio=
golang.org/x/crypto v0.12.0/go.mod h1:NF0Gs7EO5K4qLn+Ylc+fih8BSTeIjAP05siRnAh98yw=
golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliYc=
golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
golang.org/x/exp v0.0.0-20240119083558-1b970713d09a h1:Q8/wZp0KX97QFTc2ywcOE0YRjZPVIx+MXInMzdvQqcA=
@@ -332,7 +331,6 @@ golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
golang.org/x/net v0.11.0/go.mod h1:2L/ixqYpgIVXmeoSA/4Lu7BzTG4KIyPIryS4IsOd1oQ=
golang.org/x/net v0.13.0/go.mod h1:zEVYFnQC7m/vmpQFELhcD1EWkZlX69l4oqgmer6hfKA=
golang.org/x/net v0.14.0/go.mod h1:PpSgVXXLK0OxS0F31C1/tv6XNguvCrnXIDrFMspZIUI=
golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk=
golang.org/x/net v0.20.0 h1:aCL9BSgETF1k+blQaYUBx9hJ9LOGP3gAVemcZlf1Kpo=
golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY=
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -383,7 +381,6 @@ golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.9.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
@@ -395,7 +392,6 @@ golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
golang.org/x/term v0.9.0/go.mod h1:M6DEAAIenWoTxdKrOltXcmDY3rSplQUkrvaDU5FcQyo=
golang.org/x/term v0.10.0/go.mod h1:lpqdcUyK/oCiQxvxVrppt5ggO2KCZ5QblwqPnfZ6d5o=
golang.org/x/term v0.11.0/go.mod h1:zC9APTIj3jG3FdV/Ons+XE1riIZXG4aZ4GTHiPZJPIU=
golang.org/x/term v0.12.0/go.mod h1:owVbMEjm3cBLCHdkQu9b1opXd4ETQWc3BhuQGKgXgvU=
golang.org/x/term v0.16.0/go.mod h1:yn7UURbUtPyrVJPGPq404EukNFxcm/foM+bV/bfcDsY=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
@@ -407,7 +403,6 @@ golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
golang.org/x/text v0.10.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
golang.org/x/text v0.11.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
golang.org/x/text v0.12.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
pkg/rtc/config.go
+3 -1
View File
@@ -25,7 +25,8 @@ import (
)
const (
frameMarking = "urn:ietf:params:rtp-hdrext:framemarking"
frameMarking = "urn:ietf:params:rtp-hdrext:framemarking"
repairedRTPStreamID = "urn:ietf:params:rtp-hdrext:sdes:repaired-rtp-stream-id"
)
type WebRTCConfig struct {
@@ -87,6 +88,7 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) {
sdp.TransportCCURI,
frameMarking,
dd.ExtensionURI,
repairedRTPStreamID,
},
},
RTCPFeedback: RTCPFeedbackConfig{
pkg/rtc/mediaengine.go
+23
View File
@@ -15,6 +15,7 @@
package rtc
import (
"fmt"
"strings"
"github.com/pion/webrtc/v3"
@@ -24,8 +25,13 @@ import (
"github.com/livekit/protocol/livekit"
)
const (
videoRTXMimeType = "video/rtx"
)
var opusCodecCapability = webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus, ClockRate: 48000, Channels: 2, SDPFmtpLine: "minptime=10;useinbandfec=1"}
var redCodecCapability = webrtc.RTPCodecCapability{MimeType: sfu.MimeTypeAudioRed, ClockRate: 48000, Channels: 2, SDPFmtpLine: "111/111"}
var videoRTX = webrtc.RTPCodecCapability{MimeType: videoRTXMimeType, ClockRate: 90000}
func registerCodecs(me *webrtc.MediaEngine, codecs []*livekit.Codec, rtcpFeedback RTCPFeedbackConfig, filterOutH264HighProfile bool) error {
opusCodec := opusCodecCapability
@@ -50,6 +56,8 @@ func registerCodecs(me *webrtc.MediaEngine, codecs []*livekit.Codec, rtcpFeedbac
}
}
rtxEnabled := IsCodecEnabled(codecs, videoRTX)
h264HighProfileFmtp := "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=640032"
for _, codec := range []webrtc.RTPCodecParameters{
{
@@ -84,10 +92,25 @@ func registerCodecs(me *webrtc.MediaEngine, codecs []*livekit.Codec, rtcpFeedbac
if filterOutH264HighProfile && codec.RTPCodecCapability.SDPFmtpLine == h264HighProfileFmtp {
continue
}
if codec.MimeType == videoRTXMimeType {
continue
}
if IsCodecEnabled(codecs, codec.RTPCodecCapability) {
if err := me.RegisterCodec(codec, webrtc.RTPCodecTypeVideo); err != nil {
return err
}
if rtxEnabled {
if err := me.RegisterCodec(webrtc.RTPCodecParameters{
RTPCodecCapability: webrtc.RTPCodecCapability{
MimeType: videoRTXMimeType,
ClockRate: 90000,
SDPFmtpLine: fmt.Sprintf("apt=%d", codec.PayloadType),
},
PayloadType: codec.PayloadType + 1,
}, webrtc.RTPCodecTypeVideo); err != nil {
return err
}
}
}
}
return nil
pkg/rtc/mediatrack.go
+1 -3
View File
@@ -23,7 +23,6 @@ import (
"github.com/pion/webrtc/v3"
"go.uber.org/atomic"
"github.com/livekit/mediatransportutil/pkg/twcc"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
@@ -182,7 +181,7 @@ func (t *MediaTrack) UpdateCodecCid(codecs []*livekit.SimulcastCodec) {
}
// AddReceiver adds a new RTP receiver to the track, returns true when receiver represents a new codec
func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.TrackRemote, twcc *twcc.Responder, mid string) bool {
func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.TrackRemote, mid string) bool {
var newCodec bool
buff, rtcpReader := t.params.BufferFactory.GetBufferPair(uint32(track.SSRC()))
if buff == nil || rtcpReader == nil {
@@ -250,7 +249,6 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
track,
ti,
LoggerWithCodecMime(t.params.Logger, mime),
twcc,
t.params.OnRTCP,
t.params.VideoConfig.StreamTracker,
sfu.WithPliThrottleConfig(t.params.PLIThrottleConfig),
pkg/rtc/participant.go
+6 -8
View File
@@ -1205,6 +1205,10 @@ func (h PrimaryTransportHandler) OnFullyEstablished() {
}
func (p *ParticipantImpl) setupTransportManager() error {
p.twcc = twcc.NewTransportWideCCResponder()
p.twcc.OnFeedback(func(pkts []rtcp.Packet) {
p.postRtcp(pkts)
})
ath := AnyTransportHandler{p: p}
var pth transport.Handler = PublisherTransportHandler{ath}
var sth transport.Handler = SubscriberTransportHandler{ath}
@@ -1223,6 +1227,7 @@ func (p *ParticipantImpl) setupTransportManager() error {
// after the participant has joined
SubscriberAsPrimary: subscriberAsPrimary,
Config: p.params.Config,
Twcc: p.twcc,
ProtocolVersion: p.params.ProtocolVersion,
CongestionControlConfig: p.params.CongestionControlConfig,
EnabledPublishCodecs: p.enabledPublishCodecs,
@@ -1913,16 +1918,9 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei
p.dirty.Store(true)
}
ssrc := uint32(track.SSRC())
if p.twcc == nil {
p.twcc = twcc.NewTransportWideCCResponder(ssrc)
p.twcc.OnFeedback(func(pkts []rtcp.Packet) {
p.postRtcp(pkts)
})
}
p.pendingTracksLock.Unlock()
if mt.AddReceiver(rtpReceiver, track, p.twcc, mid) {
if mt.AddReceiver(rtpReceiver, track, mid) {
p.removePendingMigratedTrack(mt)
}
pkg/rtc/transport.go
+84 -6
View File
@@ -17,6 +17,7 @@ package rtc
import (
"fmt"
"net"
"strconv"
"strings"
"sync"
"time"
@@ -37,11 +38,14 @@ import (
"github.com/livekit/livekit-server/pkg/config"
"github.com/livekit/livekit-server/pkg/rtc/transport"
"github.com/livekit/livekit-server/pkg/rtc/types"
lkinterceptor "github.com/livekit/livekit-server/pkg/sfu/interceptor"
"github.com/livekit/livekit-server/pkg/sfu/pacer"
"github.com/livekit/livekit-server/pkg/sfu/rtpextension"
"github.com/livekit/livekit-server/pkg/sfu/streamallocator"
sfuutils "github.com/livekit/livekit-server/pkg/sfu/utils"
"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
sutils "github.com/livekit/livekit-server/pkg/utils"
lktwcc "github.com/livekit/mediatransportutil/pkg/twcc"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/logger/pionlogger"
@@ -199,6 +203,7 @@ type TransportParams struct {
ParticipantIdentity livekit.ParticipantIdentity
ProtocolVersion types.ProtocolVersion
Config *WebRTCConfig
Twcc *lktwcc.Responder
DirectionConfig DirectionConfig
CongestionControlConfig config.CongestionControlConfig
EnabledCodecs []*livekit.Codec
@@ -326,6 +331,40 @@ func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimat
ir.Add(f)
}
}
setTWCCForVideo := func(info *interceptor.StreamInfo) {
if !strings.HasPrefix(info.MimeType, "video") {
return
}
// rtx stream don't have rtcp feedback, always set twcc for rtx stream
twccFb := strings.HasSuffix(info.MimeType, "rtx")
if !twccFb {
for _, fb := range info.RTCPFeedback {
if fb.Type == webrtc.TypeRTCPFBTransportCC {
twccFb = true
break
}
}
}
if !twccFb {
return
}
twccExtID := sfuutils.GetHeaderExtensionID(info.RTPHeaderExtensions, webrtc.RTPHeaderExtensionCapability{URI: sdp.TransportCCURI})
if twccExtID != 0 {
if buffer := params.Config.BufferFactory.GetBuffer(info.SSRC); buffer != nil {
params.Logger.Debugw("set rtx twcc and ext id", "ssrc", info.SSRC, "twccExtID", twccExtID)
buffer.SetTWCCAndExtID(params.Twcc, uint8(twccExtID))
} else {
params.Logger.Errorw("failed to get buffer for rtx stream", nil, "ssrc", info.SSRC)
}
}
}
// put rtx interceptor behind unhandle simulcast interceptor so it can get the correct mid & rid
ir.Add(lkinterceptor.NewRTXInfoExtractorFactory(setTWCCForVideo, func(repair, base uint32) {
params.Logger.Debugw("rtx pair found from extension", "repair", repair, "base", base)
params.Config.BufferFactory.SetRTXPair(repair, base)
}, params.Logger))
api := webrtc.NewAPI(
webrtc.WithMediaEngine(me),
webrtc.WithSettingEngine(se),
@@ -1532,11 +1571,7 @@ func (t *PCTransport) handleRemoteDescriptionReceived(e *event) error {
}
}
func (t *PCTransport) isRemoteOfferRestartICE(sd *webrtc.SessionDescription) (string, bool, error) {
parsed, err := sd.Unmarshal()
if err != nil {
return "", false, err
}
func (t *PCTransport) isRemoteOfferRestartICE(parsed *sdp.SessionDescription) (string, bool, error) {
user, pwd, err := lksdp.ExtractICECredential(parsed)
if err != nil {
return "", false, err
@@ -1633,7 +1668,11 @@ func (t *PCTransport) createAndSendAnswer() error {
}
func (t *PCTransport) handleRemoteOfferReceived(sd *webrtc.SessionDescription) error {
iceCredential, offerRestartICE, err := t.isRemoteOfferRestartICE(sd)
parsed, err := sd.Unmarshal()
if err != nil {
return nil
}
iceCredential, offerRestartICE, err := t.isRemoteOfferRestartICE(parsed)
if err != nil {
return errors.Wrap(err, "check remote offer restart ice failed")
}
@@ -1655,6 +1694,13 @@ func (t *PCTransport) handleRemoteOfferReceived(sd *webrtc.SessionDescription) e
if err := t.setRemoteDescription(*sd); err != nil {
return err
}
rtxRepairs := rtxRepairsFromSDP(parsed, t.params.Logger)
if len(rtxRepairs) > 0 {
t.params.Logger.Debugw("rtx pairs found from sdp", "ssrcs", rtxRepairs)
for repair, base := range rtxRepairs {
t.params.Config.BufferFactory.SetRTXPair(repair, base)
}
}
if t.currentOfferIceCredential == "" || offerRestartICE {
t.currentOfferIceCredential = iceCredential
@@ -1778,3 +1824,35 @@ func configureAudioTransceiver(tr *webrtc.RTPTransceiver, stereo bool, nack bool
tr.SetCodecPreferences(configCodecs)
}
func rtxRepairsFromSDP(s *sdp.SessionDescription, logger logger.Logger) map[uint32]uint32 {
rtxRepairFlows := map[uint32]uint32{}
for _, media := range s.MediaDescriptions {
for _, attr := range media.Attributes {
switch attr.Key {
case sdp.AttrKeySSRCGroup:
split := strings.Split(attr.Value, " ")
if split[0] == sdp.SemanticTokenFlowIdentification {
// Essentially lines like `a=ssrc-group:FID 2231627014 632943048` are processed by this section
// as this declares that the second SSRC (632943048) is a rtx repair flow (RFC4588) for the first
// (2231627014) as specified in RFC5576
if len(split) == 3 {
baseSsrc, err := strconv.ParseUint(split[1], 10, 32)
if err != nil {
logger.Warnw("Failed to parse SSRC", err, "ssrc", split[1])
continue
}
rtxRepairFlow, err := strconv.ParseUint(split[2], 10, 32)
if err != nil {
logger.Warnw("Failed to parse SSRC", err, "ssrc", split[2])
continue
}
rtxRepairFlows[uint32(rtxRepairFlow)] = uint32(baseSsrc)
}
}
}
}
}
return rtxRepairFlows
}
pkg/rtc/transportmanager.go
+3
View File
@@ -33,6 +33,7 @@ import (
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/pkg/sfu"
"github.com/livekit/livekit-server/pkg/sfu/pacer"
"github.com/livekit/mediatransportutil/pkg/twcc"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
)
@@ -71,6 +72,7 @@ type TransportManagerParams struct {
SID livekit.ParticipantID
SubscriberAsPrimary bool
Config *WebRTCConfig
Twcc *twcc.Responder
ProtocolVersion types.ProtocolVersion
CongestionControlConfig config.CongestionControlConfig
EnabledSubscribeCodecs []*livekit.Codec
@@ -131,6 +133,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
ParticipantIdentity: params.Identity,
ProtocolVersion: params.ProtocolVersion,
Config: params.Config,
Twcc: params.Twcc,
DirectionConfig: params.Config.Publisher,
CongestionControlConfig: params.CongestionControlConfig,
EnabledCodecs: params.EnabledPublishCodecs,
pkg/rtc/unhandlesimulcast.go
+4 -11
View File
@@ -19,6 +19,8 @@ import (
"github.com/pion/rtp"
"github.com/pion/sdp/v3"
"github.com/pion/webrtc/v3"
"github.com/livekit/livekit-server/pkg/sfu/utils"
)
const (
@@ -109,20 +111,11 @@ type UnhandleSimulcastInterceptor struct {
simTracks map[uint32]SimulcastTrackInfo
}
func getHeaderExtensionID(extensions []interceptor.RTPHeaderExtension, extension webrtc.RTPHeaderExtensionCapability) int {
for _, h := range extensions {
if extension.URI == h.URI {
return h.ID
}
}
return 0
}
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{URI: sdp.SDESMidURI})
streamIDExtensionID := getHeaderExtensionID(info.RTPHeaderExtensions, webrtc.RTPHeaderExtensionCapability{URI: sdp.SDESRTPStreamIDURI})
midExtensionID := utils.GetHeaderExtensionID(info.RTPHeaderExtensions, webrtc.RTPHeaderExtensionCapability{URI: sdp.SDESMidURI})
streamIDExtensionID := utils.GetHeaderExtensionID(info.RTPHeaderExtensions, webrtc.RTPHeaderExtensionCapability{URI: sdp.SDESRTPStreamIDURI})
if midExtensionID == 0 || streamIDExtensionID == 0 {
return reader
}
pkg/sfu/buffer/buffer.go
+105 -33
View File
@@ -71,6 +71,7 @@ type Buffer struct {
videoPool *sync.Pool
audioPool *sync.Pool
codecType webrtc.RTPCodecType
payloadType uint8
extPackets deque.Deque[*ExtPacket]
pPackets []pendingPacket
closeOnce sync.Once
@@ -122,6 +123,8 @@ type Buffer struct {
packetNotFoundCount atomic.Uint32
packetTooOldCount atomic.Uint32
primaryBufferForRTX *Buffer
}
// NewBuffer constructs a new Buffer
@@ -156,11 +159,12 @@ func (b *Buffer) SetPaused(paused bool) {
b.paused = paused
}
func (b *Buffer) SetTWCC(twcc *twcc.Responder) {
func (b *Buffer) SetTWCCAndExtID(twcc *twcc.Responder, extID uint8) {
b.Lock()
defer b.Unlock()
b.twcc = twcc
b.twccExt = extID
}
func (b *Buffer) SetAudioLevelParams(audioLevelParams audio.AudioLevelParams) {
@@ -187,6 +191,17 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
b.clockRate = codec.ClockRate
b.lastReport = time.Now()
b.mime = strings.ToLower(codec.MimeType)
for _, codecParameter := range params.Codecs {
if strings.EqualFold(codecParameter.MimeType, codec.MimeType) {
b.payloadType = uint8(codecParameter.PayloadType)
break
}
}
if b.payloadType == 0 {
b.logger.Warnw("could not find payload type for codec", nil, "codec", codec.MimeType, "parameters", params)
b.payloadType = uint8(params.Codecs[0].PayloadType)
}
for _, ext := range params.HeaderExtensions {
switch ext.URI {
@@ -235,16 +250,6 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
case webrtc.TypeRTCPFBGoogREMB:
b.logger.Debugw("Setting feedback", "type", webrtc.TypeRTCPFBGoogREMB)
b.logger.Debugw("REMB not supported, RTCP feedback will not be generated")
case webrtc.TypeRTCPFBTransportCC:
if b.codecType == webrtc.RTPCodecTypeVideo {
b.logger.Debugw("Setting feedback", "type", webrtc.TypeRTCPFBTransportCC)
for _, ext := range params.HeaderExtensions {
if ext.URI == sdp.TransportCCURI {
b.twccExt = uint8(ext.ID)
break
}
}
}
case webrtc.TypeRTCPFBNACK:
// pion use a single mediaengine to manage negotiated codecs of peerconnection, that means we can't have different
// codec settings at track level for same codec type, so enable nack for all audio receivers but don't create nack queue
@@ -258,7 +263,7 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
}
for _, pp := range b.pPackets {
b.calc(pp.packet, pp.arrivalTime)
b.calc(pp.packet, nil, pp.arrivalTime, false)
}
b.pPackets = nil
b.bound = true
@@ -266,14 +271,38 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
// Write adds an RTP Packet, ordering is not guaranteed, newer packets may arrive later
func (b *Buffer) Write(pkt []byte) (n int, err error) {
b.Lock()
defer b.Unlock()
var rtpPacket rtp.Packet
err = rtpPacket.Unmarshal(pkt)
if err != nil {
return
}
b.Lock()
if b.closed.Load() {
b.Unlock()
err = io.EOF
return
}
if b.twcc != nil && b.twccExt != 0 && !b.closed.Load() {
if ext := rtpPacket.GetExtension(b.twccExt); ext != nil {
b.twcc.Push(rtpPacket.SSRC, binary.BigEndian.Uint16(ext[0:2]), time.Now().UnixNano(), rtpPacket.Marker)
}
}
// handle RTX packet
if pb := b.primaryBufferForRTX; pb != nil {
b.Unlock()
// skip padding only packets
if rtpPacket.Padding && len(rtpPacket.Payload) == 0 {
return
}
pb.writeRTX(&rtpPacket)
return
}
if !b.bound {
packet := make([]byte, len(pkt))
copy(packet, pkt)
@@ -281,10 +310,58 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) {
packet: packet,
arrivalTime: time.Now(),
})
b.Unlock()
return
}
b.calc(pkt, time.Now())
b.payloadType = rtpPacket.PayloadType
b.calc(pkt, &rtpPacket, time.Now(), false)
b.Unlock()
return
}
func (b *Buffer) SetPrimaryBufferForRTX(primaryBuffer *Buffer) {
b.Lock()
b.primaryBufferForRTX = primaryBuffer
pkts := b.pPackets
b.pPackets = nil
b.Unlock()
for _, pp := range pkts {
var rtpPacket rtp.Packet
err := rtpPacket.Unmarshal(pp.packet)
if err != nil {
continue
}
if rtpPacket.Padding && len(rtpPacket.Payload) == 0 {
continue
}
primaryBuffer.writeRTX(&rtpPacket)
}
}
func (b *Buffer) writeRTX(rtxPkt *rtp.Packet) (n int, err error) {
b.Lock()
defer b.Unlock()
if !b.bound {
return
}
videoPktPtr := b.videoPool.Get().(*[]byte)
defer b.videoPool.Put(videoPktPtr)
videoPkt := *rtxPkt
videoPkt.PayloadType = b.payloadType
videoPkt.SequenceNumber = binary.BigEndian.Uint16(rtxPkt.Payload[:2])
videoPkt.SSRC = b.mediaSSRC
videoPkt.Payload = rtxPkt.Payload[2:]
n, err = videoPkt.MarshalTo((*videoPktPtr))
if err != nil {
b.logger.Errorw("could not marshal repaired packet", err, "ssrc", b.mediaSSRC, "sn", videoPkt.SequenceNumber)
return
}
b.calc((*videoPktPtr)[:n], &videoPkt, time.Now(), true)
return
}
@@ -414,23 +491,25 @@ func (b *Buffer) SetRTT(rtt uint32) {
}
}
func (b *Buffer) calc(pkt []byte, arrivalTime time.Time) {
func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime time.Time, isRTX bool) {
defer func() {
b.doNACKs()
b.doReports(arrivalTime)
}()
var rtpPacket rtp.Packet
if err := rtpPacket.Unmarshal(pkt); err != nil {
b.logger.Errorw("could not unmarshal RTP packet", err)
return
if rtpPacket == nil {
rtpPacket = &rtp.Packet{}
if err := rtpPacket.Unmarshal(rawPkt); err != nil {
b.logger.Errorw("could not unmarshal RTP packet", err)
return
}
}
// process header extensions always as padding packets could be used for probing
b.processHeaderExtensions(&rtpPacket, arrivalTime)
b.processHeaderExtensions(rtpPacket, arrivalTime, isRTX)
flowState := b.updateStreamState(&rtpPacket, arrivalTime)
flowState := b.updateStreamState(rtpPacket, arrivalTime)
if flowState.IsNotHandled {
return
}
@@ -472,7 +551,7 @@ func (b *Buffer) calc(pkt []byte, arrivalTime time.Time) {
}
flowState.ExtSequenceNumber -= snAdjustment
rtpPacket.Header.SequenceNumber = uint16(flowState.ExtSequenceNumber)
_, err = b.bucket.AddPacketWithSequenceNumber(pkt, rtpPacket.Header.SequenceNumber)
_, err = b.bucket.AddPacketWithSequenceNumber(rawPkt, rtpPacket.Header.SequenceNumber)
if err != nil {
if errors.Is(err, bucket.ErrPacketTooOld) {
packetTooOldCount := b.packetTooOldCount.Inc()
@@ -485,7 +564,7 @@ func (b *Buffer) calc(pkt []byte, arrivalTime time.Time) {
return
}
ep := b.getExtPacket(&rtpPacket, arrivalTime, flowState)
ep := b.getExtPacket(rtpPacket, arrivalTime, flowState)
if ep == nil {
return
}
@@ -570,16 +649,9 @@ func (b *Buffer) updateStreamState(p *rtp.Packet, arrivalTime time.Time) RTPFlow
return flowState
}
func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime time.Time) {
// submit to TWCC even if it is a padding only packet. Clients use padding only packets as probes
// for bandwidth estimation
if b.twcc != nil && b.twccExt != 0 {
if ext := p.GetExtension(b.twccExt); ext != nil {
b.twcc.Push(binary.BigEndian.Uint16(ext[0:2]), arrivalTime.UnixNano(), p.Marker)
}
}
func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime time.Time, isRTX bool) {
if b.audioLevelExt != 0 {
if b.audioLevelExt != 0 && !isRTX {
if !b.latestTSForAudioLevelInitialized {
b.latestTSForAudioLevelInitialized = true
b.latestTSForAudioLevel = p.Timestamp
pkg/sfu/buffer/factory.go
+30
View File
@@ -51,6 +51,7 @@ func (f *FactoryOfBufferFactory) CreateBufferFactory() *Factory {
audioPool: f.audioPool,
rtpBuffers: make(map[uint32]*Buffer),
rtcpReaders: make(map[uint32]*RTCPReader),
rtxPair: make(map[uint32]uint32),
}
}
@@ -60,6 +61,7 @@ type Factory struct {
audioPool *sync.Pool
rtpBuffers map[uint32]*Buffer
rtcpReaders map[uint32]*RTCPReader
rtxPair map[uint32]uint32 // repair -> base
}
func (f *Factory) GetOrNew(packetType packetio.BufferPacketType, ssrc uint32) io.ReadWriteCloser {
@@ -84,9 +86,25 @@ func (f *Factory) GetOrNew(packetType packetio.BufferPacketType, ssrc uint32) io
}
buffer := NewBuffer(ssrc, f.videoPool, f.audioPool)
f.rtpBuffers[ssrc] = buffer
for repair, base := range f.rtxPair {
if repair == ssrc {
baseBuffer, ok := f.rtpBuffers[base]
if ok {
buffer.SetPrimaryBufferForRTX(baseBuffer)
}
break
} else if base == ssrc {
repairBuffer, ok := f.rtpBuffers[repair]
if ok {
repairBuffer.SetPrimaryBufferForRTX(buffer)
}
break
}
}
buffer.OnClose(func() {
f.Lock()
delete(f.rtpBuffers, ssrc)
delete(f.rtxPair, ssrc)
f.Unlock()
})
return buffer
@@ -111,3 +129,15 @@ func (f *Factory) GetRTCPReader(ssrc uint32) *RTCPReader {
defer f.RUnlock()
return f.rtcpReaders[ssrc]
}
func (f *Factory) SetRTXPair(repair, base uint32) {
f.Lock()
repairBuffer, baseBuffer := f.rtpBuffers[repair], f.rtpBuffers[base]
if repairBuffer == nil || baseBuffer == nil {
f.rtxPair[repair] = base
}
f.Unlock()
if repairBuffer != nil && baseBuffer != nil {
repairBuffer.SetPrimaryBufferForRTX(baseBuffer)
}
}
pkg/sfu/downtrack.go
+2 -1
View File
@@ -38,6 +38,7 @@ import (
dd "github.com/livekit/livekit-server/pkg/sfu/dependencydescriptor"
"github.com/livekit/livekit-server/pkg/sfu/pacer"
"github.com/livekit/livekit-server/pkg/sfu/rtpextension"
"github.com/livekit/livekit-server/pkg/sfu/utils"
)
// TrackSender defines an interface send media to remote peer
@@ -357,7 +358,7 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters,
}
var codec webrtc.RTPCodecParameters
for _, c := range d.upstreamCodecs {
matchCodec, err := codecParametersFuzzySearch(c, t.CodecParameters())
matchCodec, err := utils.CodecParametersFuzzySearch(c, t.CodecParameters())
if err == nil {
codec = matchCodec
break
pkg/sfu/interceptor/rtx.go
+168
View File
@@ -0,0 +1,168 @@
package interceptor
import (
"sync"
"github.com/pion/interceptor"
"github.com/pion/sdp/v3"
"github.com/pion/webrtc/v3"
"github.com/livekit/livekit-server/pkg/sfu/utils"
"github.com/livekit/protocol/logger"
)
const (
SDESRepairRTPStreamIDURI = "urn:ietf:params:rtp-hdrext:sdes:repaired-rtp-stream-id"
rtxProbeCount = 10
)
type streamInfo struct {
mid string
rid string
rsid string
}
type RTXInfoExtractorFactory struct {
onStreamFound func(*interceptor.StreamInfo)
onRTXPairFound func(repair, base uint32)
lock sync.Mutex
streams map[uint32]streamInfo
logger logger.Logger
}
func NewRTXInfoExtractorFactory(onStreamFound func(*interceptor.StreamInfo), onRTXPairFound func(repair, base uint32), logger logger.Logger) *RTXInfoExtractorFactory {
return &RTXInfoExtractorFactory{
onStreamFound: onStreamFound,
onRTXPairFound: onRTXPairFound,
streams: make(map[uint32]streamInfo),
logger: logger,
}
}
func (f *RTXInfoExtractorFactory) NewInterceptor(id string) (interceptor.Interceptor, error) {
return &RTXInfoExtractor{
factory: f,
logger: f.logger,
}, nil
}
func (f *RTXInfoExtractorFactory) setStreamInfo(ssrc uint32, mid, rid, rsid string) {
var repairSsrc, baseSsrc uint32
f.lock.Lock()
if rsid != "" {
// repair stream found, find base stream
for base, info := range f.streams {
if info.mid == mid && info.rid == rsid {
repairSsrc = ssrc
baseSsrc = base
delete(f.streams, base)
break
}
}
} else {
// base stream found, find repair stream
for repair, info := range f.streams {
if info.mid == mid && info.rsid == rid {
repairSsrc = repair
baseSsrc = ssrc
delete(f.streams, repair)
break
}
}
}
// no rtx pair found, save it for later
if repairSsrc == 0 || baseSsrc == 0 {
f.streams[ssrc] = streamInfo{
mid: mid,
rid: rid,
rsid: rsid,
}
}
f.lock.Unlock()
if repairSsrc != 0 && baseSsrc != 0 {
f.onRTXPairFound(repairSsrc, baseSsrc)
}
}
type RTXInfoExtractor struct {
interceptor.NoOp
factory *RTXInfoExtractorFactory
logger logger.Logger
}
func (u *RTXInfoExtractor) BindRemoteStream(info *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader {
u.factory.onStreamFound(info)
midExtensionID := utils.GetHeaderExtensionID(info.RTPHeaderExtensions, webrtc.RTPHeaderExtensionCapability{URI: sdp.SDESMidURI})
streamIDExtensionID := utils.GetHeaderExtensionID(info.RTPHeaderExtensions, webrtc.RTPHeaderExtensionCapability{URI: sdp.SDESRTPStreamIDURI})
repairStreamIDExtensionID := utils.GetHeaderExtensionID(info.RTPHeaderExtensions, webrtc.RTPHeaderExtensionCapability{URI: SDESRepairRTPStreamIDURI})
if midExtensionID == 0 || streamIDExtensionID == 0 || repairStreamIDExtensionID == 0 {
return reader
}
return &rtxInfoReader{
tryTimes: rtxProbeCount,
reader: reader,
midExtID: uint8(midExtensionID),
ridExtID: uint8(streamIDExtensionID),
rsidExtID: uint8(repairStreamIDExtensionID),
factory: u.factory,
logger: u.logger,
}
}
type rtxInfoReader struct {
tryTimes int
reader interceptor.RTPReader
midExtID uint8
ridExtID uint8
rsidExtID uint8
factory *RTXInfoExtractorFactory
logger logger.Logger
}
func (r *rtxInfoReader) Read(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) {
n, a, err := r.reader.Read(b, a)
if r.tryTimes < 0 || err != nil {
return n, a, err
}
if a == nil {
a = make(interceptor.Attributes)
}
header, err := a.GetRTPHeader(b[:n])
if err != nil {
return n, a, nil
}
var mid, rid, rsid string
if payload := header.GetExtension(r.midExtID); payload != nil {
mid = string(payload)
}
if payload := header.GetExtension(r.ridExtID); payload != nil {
rid = string(payload)
}
if payload := header.GetExtension(r.rsidExtID); payload != nil {
rsid = string(payload)
}
if mid != "" && (rid != "" || rsid != "") {
r.logger.Debugw("stream found", "mid", mid, "rid", rid, "rsid", rsid, "ssrc", header.SSRC)
r.tryTimes = -1
go r.factory.setStreamInfo(header.SSRC, mid, rid, rsid)
} else {
// ignore padding only packet for probe count
if !(header.Padding && n-header.MarshalSize()-int(b[n-1]) == 0) {
r.tryTimes--
}
}
return n, a, nil
}
pkg/sfu/receiver.go
-3
View File
@@ -197,7 +197,6 @@ func NewWebRTCReceiver(
track *webrtc.TrackRemote,
trackInfo *livekit.TrackInfo,
logger logger.Logger,
twcc *twcc.Responder,
onRTCP func([]rtcp.Packet),
trackersConfig config.StreamTrackersConfig,
opts ...ReceiverOpts,
@@ -209,7 +208,6 @@ func NewWebRTCReceiver(
streamID: track.StreamID(),
codec: track.Codec(),
kind: track.Kind(),
twcc: twcc,
onRTCP: onRTCP,
isSVC: IsSvcCodec(track.Codec().MimeType),
isRED: IsRedCodec(track.Codec().MimeType),
@@ -340,7 +338,6 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff
layer = buffer.RidToSpatialLayer(track.RID(), w.trackInfo.Load())
}
buff.SetLogger(w.logger.WithValues("layer", layer))
buff.SetTWCC(w.twcc)
buff.SetAudioLevelParams(audio.AudioLevelParams{
ActiveLevel: w.audioConfig.ActiveLevel,
MinPercentile: w.audioConfig.MinPercentile,
pkg/sfu/helpers.go → pkg/sfu/utils/helpers.go
+12 -3
View File
@@ -12,17 +12,18 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package sfu
package utils
import (
"strings"
"github.com/pion/interceptor"
"github.com/pion/webrtc/v3"
)
// Do a fuzzy find for a codec in the list of codecs
// Used for lookup up a codec in an existing list to find a match
func codecParametersFuzzySearch(needle webrtc.RTPCodecParameters, haystack []webrtc.RTPCodecParameters) (webrtc.RTPCodecParameters, error) {
func CodecParametersFuzzySearch(needle webrtc.RTPCodecParameters, haystack []webrtc.RTPCodecParameters) (webrtc.RTPCodecParameters, error) {
// First attempt to match on MimeType + SDPFmtpLine
for _, c := range haystack {
if strings.EqualFold(c.RTPCodecCapability.MimeType, needle.RTPCodecCapability.MimeType) &&
@@ -41,4 +42,12 @@ func codecParametersFuzzySearch(needle webrtc.RTPCodecParameters, haystack []web
return webrtc.RTPCodecParameters{}, webrtc.ErrCodecNotFound
}
// -----------------------------------------------
// GetHeaderExtensionID returns the ID of a header extension, or 0 if not found
func GetHeaderExtensionID(extensions []interceptor.RTPHeaderExtension, extension webrtc.RTPHeaderExtensionCapability) int {
for _, h := range extensions {
if extension.URI == h.URI {
return h.ID
}
}
return 0
}