dandri/livekit
1
0 Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-10-22 15:22:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add RTX to downstream (#3247)

Browse Source 
* Add RTX to downstream

* test
This commit is contained in:
Raja Subramanian authored and GitHub committed 2024-12-13 09:57:03 +05:30
1 parent 79eda6b72b
commit 789d0484e2
6 files changed
+72 -28
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pkg/config/config.go
+2 -1
View File
@@ -333,11 +333,12 @@ var DefaultConfig = Config{
AutoCreate: true,
EnabledCodecs: []CodecSpec{
{Mime: webrtc.MimeTypeOpus},
{Mime: "audio/red"},
{Mime: sfu.MimeTypeAudioRed},
{Mime: webrtc.MimeTypeVP8},
{Mime: webrtc.MimeTypeH264},
{Mime: webrtc.MimeTypeVP9},
{Mime: webrtc.MimeTypeAV1},
{Mime: webrtc.MimeTypeRTX},
},
EmptyTimeout: 5 * 60,
DepartureTimeout: 20,
pkg/rtc/mediaengine.go
+3 -7
View File
@@ -24,10 +24,6 @@ import (
"github.com/livekit/protocol/livekit"
)
const (
videoRTXMimeType = "video/rtx"
)
var OpusCodecCapability = webrtc.RTPCodecCapability{
MimeType: webrtc.MimeTypeOpus,
ClockRate: 48000,
@@ -41,7 +37,7 @@ var RedCodecCapability = webrtc.RTPCodecCapability{
SDPFmtpLine: "111/111",
}
var videoRTX = webrtc.RTPCodecCapability{
MimeType: videoRTXMimeType,
MimeType: webrtc.MimeTypeRTX,
ClockRate: 90000,
}
@@ -137,7 +133,7 @@ func registerCodecs(me *webrtc.MediaEngine, codecs []*livekit.Codec, rtcpFeedbac
if filterOutH264HighProfile && codec.RTPCodecCapability.SDPFmtpLine == h264HighProfileFmtp {
continue
}
if codec.MimeType == videoRTXMimeType {
if strings.EqualFold(codec.MimeType, webrtc.MimeTypeRTX) {
continue
}
if IsCodecEnabled(codecs, codec.RTPCodecCapability) {
@@ -147,7 +143,7 @@ func registerCodecs(me *webrtc.MediaEngine, codecs []*livekit.Codec, rtcpFeedbac
if rtxEnabled {
if err := me.RegisterCodec(webrtc.RTPCodecParameters{
RTPCodecCapability: webrtc.RTPCodecCapability{
MimeType: videoRTXMimeType,
MimeType: webrtc.MimeTypeRTX,
ClockRate: 90000,
SDPFmtpLine: fmt.Sprintf("apt=%d", codec.PayloadType),
},
pkg/rtc/participant.go
+1 -1
View File
@@ -3017,7 +3017,7 @@ func (p *ParticipantImpl) setupEnabledCodecs(publishEnabledCodecs []*livekit.Cod
func (p *ParticipantImpl) GetEnabledPublishCodecs() []*livekit.Codec {
codecs := make([]*livekit.Codec, 0, len(p.enabledPublishCodecs))
for _, c := range p.enabledPublishCodecs {
if c.Mime == "video/rtx" {
if strings.EqualFold(c.Mime, webrtc.MimeTypeRTX) {
continue
}
codecs = append(codecs, c)
pkg/sfu/downtrack.go
+51 -19
View File
@@ -20,6 +20,7 @@ import (
"fmt"
"io"
"math"
"math/rand"
"strings"
"sync"
"time"
@@ -217,13 +218,16 @@ type DowntrackParams struct {
// - closed
// once closed, a DownTrack cannot be re-used.
type DownTrack struct {
params DowntrackParams
id livekit.TrackID
kind webrtc.RTPCodecType
mime string
ssrc uint32
payloadType uint8
sequencer *sequencer
params DowntrackParams
id livekit.TrackID
kind webrtc.RTPCodecType
mime string
ssrc uint32
ssrcRTX uint32
payloadType uint8
payloadTypeRTX uint8
sequencer *sequencer
rtxSequenceNumber atomic.Uint64
forwarder *Forwarder
@@ -244,6 +248,7 @@ type DownTrack struct {
transceiver atomic.Pointer[webrtc.RTPTransceiver]
writeStream webrtc.TrackLocalWriter
rtcpReader *buffer.RTCPReader
rtcpReaderRTX *buffer.RTCPReader
listenerLock sync.RWMutex
receiverReportListeners []ReceiverReportListener
@@ -387,6 +392,7 @@ func NewDownTrack(params DowntrackParams) (*DownTrack, error) {
}
d.params.Receiver.AddOnReady(d.handleReceiverReady)
d.rtxSequenceNumber.Store(uint64(rand.Intn(1<<14)) + uint64(1<<15)) // a random number in third quartile of sequence number space
d.params.Logger.Debugw("downtrack created", "upstreamCodecs", d.upstreamCodecs)
return d, nil
@@ -490,6 +496,7 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters,
"codecs", d.upstreamCodecs,
"matchCodec", codec,
"ssrc", t.SSRC(),
"ssrcRTX", t.SSRCRetransmission(),
"isFECEnabled", isFECEnabled,
}
if d.isRED {
@@ -504,15 +511,23 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters,
)
d.ssrc = uint32(t.SSRC())
d.ssrcRTX = uint32(t.SSRCRetransmission())
d.payloadType = uint8(codec.PayloadType)
d.payloadTypeRTX = uint8(utils.FindRTXPayloadType(codec.PayloadType, t.CodecParameters()))
d.writeStream = t.WriteStream()
d.mime = strings.ToLower(codec.MimeType)
if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, uint32(t.SSRC())).(*buffer.RTCPReader); rr != nil {
if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, d.ssrc).(*buffer.RTCPReader); rr != nil {
rr.OnPacket(func(pkt []byte) {
d.handleRTCP(pkt)
})
d.rtcpReader = rr
}
if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, d.ssrcRTX).(*buffer.RTCPReader); rr != nil {
rr.OnPacket(func(pkt []byte) {
d.handleRTCP(pkt)
})
d.rtcpReaderRTX = rr
}
d.sequencer = newSequencer(d.params.MaxTrack, d.kind == webrtc.RTPCodecTypeVideo, d.params.Logger)
@@ -1154,6 +1169,11 @@ func (d *DownTrack) CloseWithFlush(flush bool) {
d.rtcpReader.Close()
d.rtcpReader.OnPacket(nil)
}
if d.rtcpReaderRTX != nil && flush {
d.params.Logger.Debugw("downtrack close rtcp rtx reader")
d.rtcpReaderRTX.Close()
d.rtcpReaderRTX.OnPacket(nil)
}
d.bindLock.Unlock()
d.connectionStats.Close()
@@ -1761,7 +1781,7 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
}
case *rtcp.TransportLayerCC:
if p.MediaSSRC == d.ssrc {
if p.MediaSSRC == d.ssrc || p.MediaSSRC == d.ssrcRTX {
if sal := d.getStreamAllocatorListener(); sal != nil {
sal.OnTransportCCFeedback(d, p)
}
@@ -1883,17 +1903,13 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
Timestamp: epm.timestamp,
SSRC: d.ssrc,
}
rtxOffset := 0
if d.payloadTypeRTX != 0 && d.ssrcRTX != 0 {
rtxOffset = 2
poolEntity := PacketFactory.Get().(*[]byte)
payload := *poolEntity
if len(epm.codecBytesSlice) != 0 {
n := copy(payload, epm.codecBytesSlice)
m := copy(payload[n:], pkt.Payload[epm.numCodecBytesIn:])
payload = payload[:n+m]
} else {
copy(payload, epm.codecBytes[:epm.numCodecBytesOut])
copy(payload[epm.numCodecBytesOut:], pkt.Payload[epm.numCodecBytesIn:])
payload = payload[:int(epm.numCodecBytesOut)+len(pkt.Payload)-int(epm.numCodecBytesIn)]
hdr.PayloadType = d.payloadTypeRTX
hdr.SequenceNumber = uint16(d.rtxSequenceNumber.Inc())
hdr.SSRC = d.ssrcRTX
}
if d.dependencyDescriptorExtID != 0 {
@@ -1912,6 +1928,22 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
}
d.addDummyExtensions(hdr)
poolEntity := PacketFactory.Get().(*[]byte)
payload := *poolEntity
if rtxOffset != 0 {
// write OSN (Original Sequence Number)
binary.BigEndian.PutUint16(payload[0:2], epm.targetSeqNo)
}
if len(epm.codecBytesSlice) != 0 {
n := copy(payload[rtxOffset:], epm.codecBytesSlice)
m := copy(payload[rtxOffset+n:], pkt.Payload[epm.numCodecBytesIn:])
payload = payload[:rtxOffset+n+m]
} else {
copy(payload[rtxOffset:], epm.codecBytes[:epm.numCodecBytesOut])
copy(payload[rtxOffset+int(epm.numCodecBytesOut):], pkt.Payload[epm.numCodecBytesIn:])
payload = payload[:rtxOffset+int(epm.numCodecBytesOut)+len(pkt.Payload)-int(epm.numCodecBytesIn)]
}
headerSize := hdr.MarshalSize()
d.rtpStats.Update(
mono.UnixNano(),
pkg/sfu/utils/helpers.go
+12
View File
@@ -45,6 +45,18 @@ func CodecParametersFuzzySearch(needle webrtc.RTPCodecParameters, haystack []web
return webrtc.RTPCodecParameters{}, webrtc.ErrCodecNotFound
}
// Given a CodecParameters find the RTX CodecParameters if one exists
func FindRTXPayloadType(needle webrtc.PayloadType, haystack []webrtc.RTPCodecParameters) webrtc.PayloadType {
aptStr := fmt.Sprintf("apt=%d", needle)
for _, c := range haystack {
if aptStr == c.SDPFmtpLine {
return c.PayloadType
}
}
return webrtc.PayloadType(0)
}
// 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 {
test/client/client.go
+3
View File
@@ -42,6 +42,7 @@ import (
"github.com/livekit/livekit-server/pkg/rtc"
"github.com/livekit/livekit-server/pkg/rtc/transport/transportfakes"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
)
type SignalRequestHandler func(msg *livekit.SignalRequest) error
@@ -172,6 +173,8 @@ func NewRTCClient(conn *websocket.Conn, opts *Options) (*RTCClient, error) {
}
conf.SettingEngine.SetLite(false)
conf.SettingEngine.SetAnsweringDTLSRole(webrtc.DTLSRoleClient)
ff := buffer.NewFactoryOfBufferFactory(500, 200)
conf.SetBufferFactory(ff.CreateBufferFactory())
var codecs []*livekit.Codec
for _, codec := range []*livekit.Codec{
{