diff --git a/go.mod b/go.mod index 08fb3f734..b75305fc9 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index a333ca31b..917cf90d7 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/pkg/rtc/config.go b/pkg/rtc/config.go index 820d41857..239b08608 100644 --- a/pkg/rtc/config.go +++ b/pkg/rtc/config.go @@ -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{ diff --git a/pkg/rtc/mediaengine.go b/pkg/rtc/mediaengine.go index 8bbb3db20..55836b472 100644 --- a/pkg/rtc/mediaengine.go +++ b/pkg/rtc/mediaengine.go @@ -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 diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index e2978d82d..73aa631a5 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -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), diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 22bc2ac99..0a64d74c1 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -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) } diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index c0e804d65..164d47926 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -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 +} diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 401fea8e0..9dfbcce5f 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -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, diff --git a/pkg/rtc/unhandlesimulcast.go b/pkg/rtc/unhandlesimulcast.go index 568c7dc1b..0c4292633 100644 --- a/pkg/rtc/unhandlesimulcast.go +++ b/pkg/rtc/unhandlesimulcast.go @@ -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 } diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 7e250c23a..bf73cf27a 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -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 diff --git a/pkg/sfu/buffer/factory.go b/pkg/sfu/buffer/factory.go index d0a9979f8..d59038d28 100644 --- a/pkg/sfu/buffer/factory.go +++ b/pkg/sfu/buffer/factory.go @@ -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) + } +} diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 470e11eed..54a948f7a 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -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 diff --git a/pkg/sfu/interceptor/rtx.go b/pkg/sfu/interceptor/rtx.go new file mode 100644 index 000000000..624ede827 --- /dev/null +++ b/pkg/sfu/interceptor/rtx.go @@ -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 +} diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index b05b2b23c..bdb8853ea 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -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, diff --git a/pkg/sfu/helpers.go b/pkg/sfu/utils/helpers.go similarity index 76% rename from pkg/sfu/helpers.go rename to pkg/sfu/utils/helpers.go index 1f4101910..476050416 100644 --- a/pkg/sfu/helpers.go +++ b/pkg/sfu/utils/helpers.go @@ -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 +}