Merge branch 'master' into match-header-extensions

AUTHORS.txt

@@ -8,6 +8,7 @@ Aaron Boushley <boushley@pretzelaux.com>
 Aaron France <aaron.l.france@gmail.com>
 Adam Kiss <masterada@gmail.com>
 Aditya Kumar <k.aditya00@gmail.com>
+Adrian Cable <6544927+adriancable@users.noreply.github.com>
 Adrian Cable <adrian.cable@gmail.com>
 adwpc <adwpc@hotmail.com>
 aggresss <aggresss@163.com>
@@ -200,6 +201,7 @@ Steffen Vogel <post@steffenvogel.de>
 stephanrotolante <stephanrotolante@gmail.com>
 streamer45 <cstcld91@gmail.com>
 Suhas Gaddam <suhas.g.2011@gmail.com>
+sukun <sukunrt@gmail.com>
 Suzuki Takeo <takeo@stko.info>
 sylba2050 <masataka.hisasue@optim.co.jp>
 Tarrence van As <tarrencev@users.noreply.github.com>

constants.go

@@ -31,6 +31,13 @@ const (
 	generatedCertificateOrigin = "WebRTC"
 
 	sdesRepairRTPStreamIDURI = "urn:ietf:params:rtp-hdrext:sdes:repaired-rtp-stream-id"
+
+	// AttributeRtxPayloadType is the interceptor attribute added when Read() returns an RTX packet containing the RTX stream payload type
+	AttributeRtxPayloadType = "rtx_payload_type"
+	// AttributeRtxSsrc is the interceptor attribute added when Read() returns an RTX packet containing the RTX stream SSRC
+	AttributeRtxSsrc = "rtx_ssrc"
+	// AttributeRtxSequenceNumber is the interceptor attribute added when Read() returns an RTX packet containing the RTX stream sequence number
+	AttributeRtxSequenceNumber = "rtx_sequence_number"
 )
 
 func defaultSrtpProtectionProfiles() []dtls.SRTPProtectionProfile {

go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/pion/datachannel v1.5.5
 	github.com/pion/dtls/v2 v2.2.7
 	github.com/pion/ice/v3 v3.0.1
-	github.com/pion/interceptor v0.1.19
+	github.com/pion/interceptor v0.1.22
 	github.com/pion/logging v0.2.2
 	github.com/pion/randutil v0.1.0
 	github.com/pion/rtcp v1.2.10
@@ -21,5 +21,5 @@ require (
 	github.com/pion/transport/v3 v3.0.1
 	github.com/sclevine/agouti v3.0.0+incompatible
 	github.com/stretchr/testify v1.8.4
-	golang.org/x/net v0.14.0
+	golang.org/x/net v0.16.0
 )

go.sum

@@ -46,8 +46,8 @@ github.com/pion/dtls/v2 v2.2.7 h1:cSUBsETxepsCSFSxC3mc/aDo14qQLMSL+O6IjG28yV8=
 github.com/pion/dtls/v2 v2.2.7/go.mod h1:8WiMkebSHFD0T+dIU+UeBaoV7kDhOW5oDCzZ7WZ/F9s=
 github.com/pion/ice/v3 v3.0.1 h1:dwWGgIFDlYrKrCW13LihifuFabGw375hoU0347S9wNw=
 github.com/pion/ice/v3 v3.0.1/go.mod h1:j4tfTlj4aSEQN9gP3IdliSHcUTWTu9tlOZL0c59MFXo=
-github.com/pion/interceptor v0.1.19 h1:tq0TGBzuZQqipyBhaC1mVUCfCh8XjDKUuibq9rIl5t4=
-github.com/pion/interceptor v0.1.19/go.mod h1:VANhFxdJezB8mwToMMmrmyHyP9gym6xLqIUch31xryg=
+github.com/pion/interceptor v0.1.22 h1:khhimAF0/VmGaIfeE+bA3X1jm0lD8C8HOGcU7vpWcPA=
+github.com/pion/interceptor v0.1.22/go.mod h1:wkbPYAak5zKsfpVDYMtEfWEy8D4zL+rpxCxPImLOg3Y=
 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 h1:HhicWIg7OX5PVilyBO6plhMetInbzkVJAhbdJiAeVaI=
@@ -99,8 +99,9 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 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.12.0 h1:tFM/ta59kqch6LlvYnPa0yx5a83cL2nHflFhYKvv9Yk=
 golang.org/x/crypto v0.12.0/go.mod h1:NF0Gs7EO5K4qLn+Ylc+fih8BSTeIjAP05siRnAh98yw=
+golang.org/x/crypto v0.14.0 h1:wBqGXzWJW6m1XrIKlAH0Hs1JJ7+9KBwnIO8v66Q9cHc=
+golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
 golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
@@ -117,8 +118,9 @@ golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns=
 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.14.0 h1:BONx9s002vGdD9umnlX1Po8vOZmrgH34qlHcD1MfK14=
 golang.org/x/net v0.14.0/go.mod h1:PpSgVXXLK0OxS0F31C1/tv6XNguvCrnXIDrFMspZIUI=
+golang.org/x/net v0.16.0 h1:7eBu7KsSvFDtSXUIDbh3aqlK4DPsZ1rByC8PFfBThos=
+golang.org/x/net v0.16.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -144,8 +146,9 @@ golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 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.11.0 h1:eG7RXZHdqOJ1i+0lgLgCpSXAp6M3LYlAo6osgSi0xOM=
 golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
+golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
@@ -154,6 +157,7 @@ golang.org/x/term v0.7.0/go.mod h1:P32HKFT3hSsZrRxla30E9HqToFYAQPCMs/zFMBUFqPY=
 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.11.0/go.mod h1:zC9APTIj3jG3FdV/Ons+XE1riIZXG4aZ4GTHiPZJPIU=
+golang.org/x/term v0.13.0/go.mod h1:LTmsnFJwVN6bCy1rVCoS+qHT1HhALEFxKncY3WNNh4U=
 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=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
@@ -162,8 +166,9 @@ golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 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.12.0 h1:k+n5B8goJNdU7hSvEtMUz3d1Q6D/XW4COJSJR6fN0mc=
 golang.org/x/text v0.12.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
+golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
+golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=

rtpreceiver.go

@@ -7,6 +7,7 @@
 package webrtc
 
 import (
+	"encoding/binary"
 	"fmt"
 	"io"
 	"sync"
@@ -31,13 +32,28 @@ type trackStreams struct {
 	rtcpReadStream  *srtp.ReadStreamSRTCP
 	rtcpInterceptor interceptor.RTCPReader
 
-	repairReadStream  *srtp.ReadStreamSRTP
-	repairInterceptor interceptor.RTPReader
+	repairReadStream    *srtp.ReadStreamSRTP
+	repairInterceptor   interceptor.RTPReader
+	repairStreamChannel chan rtxPacketWithAttributes
 
 	repairRtcpReadStream  *srtp.ReadStreamSRTCP
 	repairRtcpInterceptor interceptor.RTCPReader
 }
 
+type rtxPacketWithAttributes struct {
+	pkt        []byte
+	attributes interceptor.Attributes
+	pool       *sync.Pool
+}
+
+func (p *rtxPacketWithAttributes) release() {
+	if p.pkt != nil {
+		b := p.pkt[:cap(p.pkt)]
+		p.pool.Put(b) // nolint:staticcheck
+		p.pkt = nil
+	}
+}
+
 // RTPReceiver allows an application to inspect the receipt of a TrackRemote
 type RTPReceiver struct {
 	kind      RTPCodecType
@@ -52,6 +68,8 @@ type RTPReceiver struct {
 
 	// A reference to the associated api object
 	api *API
+
+	rtxPool sync.Pool
 }
 
 // NewRTPReceiver constructs a new RTPReceiver
@@ -67,6 +85,9 @@ func (api *API) NewRTPReceiver(kind RTPCodecType, transport *DTLSTransport) (*RT
 		closed:    make(chan interface{}),
 		received:  make(chan interface{}),
 		tracks:    []trackStreams{},
+		rtxPool: sync.Pool{New: func() interface{} {
+			return make([]byte, api.settingEngine.getReceiveMTU())
+		}},
 	}
 
 	return r, nil
@@ -145,6 +166,7 @@ func (r *RTPReceiver) configureReceive(parameters RTPReceiveParameters) {
 			track: newTrackRemote(
 				r.kind,
 				parameters.Encodings[i].SSRC,
+				parameters.Encodings[i].RTX.SSRC,
 				parameters.Encodings[i].RID,
 				r,
 			),
@@ -388,8 +410,6 @@ func (r *RTPReceiver) receiveForRid(rid string, params RTPParameters, streamInfo
 }
 
 // receiveForRtx starts a routine that processes the repair stream
-// These packets aren't exposed to the user yet, but we need to process them for
-// TWCC
 func (r *RTPReceiver) receiveForRtx(ssrc SSRC, rsid string, streamInfo *interceptor.StreamInfo, rtpReadStream *srtp.ReadStreamSRTP, rtpInterceptor interceptor.RTPReader, rtcpReadStream *srtp.ReadStreamSRTCP, rtcpInterceptor interceptor.RTCPReader) error {
 	var track *trackStreams
 	if ssrc != 0 && len(r.tracks) == 1 {
@@ -411,12 +431,53 @@ func (r *RTPReceiver) receiveForRtx(ssrc SSRC, rsid string, streamInfo *intercep
 	track.repairInterceptor = rtpInterceptor
 	track.repairRtcpReadStream = rtcpReadStream
 	track.repairRtcpInterceptor = rtcpInterceptor
+	track.repairStreamChannel = make(chan rtxPacketWithAttributes)
 
 	go func() {
-		b := make([]byte, r.api.settingEngine.getReceiveMTU())
 		for {
-			if _, _, readErr := track.repairInterceptor.Read(b, nil); readErr != nil {
+			b := r.rtxPool.Get().([]byte) // nolint:forcetypeassert
+			i, attributes, err := track.repairInterceptor.Read(b, nil)
+			if err != nil {
+				r.rtxPool.Put(b) // nolint:staticcheck
+				return
+			}
+
+			// RTX packets have a different payload format. Move the OSN in the payload to the RTP header and rewrite the
+			// payload type and SSRC, so that we can return RTX packets to the caller 'transparently' i.e. in the same format
+			// as non-RTX RTP packets
+			hasExtension := b[0]&0b10000 > 0
+			hasPadding := b[0]&0b100000 > 0
+			csrcCount := b[0] & 0b1111
+			headerLength := uint16(12 + (4 * csrcCount))
+			paddingLength := 0
+			if hasExtension {
+				headerLength += 4 * (1 + binary.BigEndian.Uint16(b[headerLength+2:headerLength+4]))
+			}
+			if hasPadding {
+				paddingLength = int(b[i-1])
+			}
+
+			if i-int(headerLength)-paddingLength < 2 {
+				// BWE probe packet, ignore
+				r.rtxPool.Put(b) // nolint:staticcheck
+				continue
+			}
+
+			attributes.Set(AttributeRtxPayloadType, b[1]&0x7F)
+			attributes.Set(AttributeRtxSequenceNumber, binary.BigEndian.Uint16(b[2:4]))
+			attributes.Set(AttributeRtxSsrc, binary.BigEndian.Uint32(b[8:12]))
+
+			b[1] = (b[1] & 0x80) | uint8(track.track.PayloadType())
+			b[2] = b[headerLength]
+			b[3] = b[headerLength+1]
+			binary.BigEndian.PutUint32(b[8:12], uint32(track.track.SSRC()))
+			copy(b[headerLength:i-2], b[headerLength+2:i])
+
+			select {
+			case <-r.closed:
+				r.rtxPool.Put(b) // nolint:staticcheck
 				return
+			case track.repairStreamChannel <- rtxPacketWithAttributes{pkt: b[:i-2], attributes: attributes, pool: &r.rtxPool}:
 			}
 		}
 	}()
@@ -455,3 +516,25 @@ func (r *RTPReceiver) setRTPReadDeadline(deadline time.Time, reader *TrackRemote
 	}
 	return fmt.Errorf("%w: %d", errRTPReceiverWithSSRCTrackStreamNotFound, reader.SSRC())
 }
+
+// readRTX returns an RTX packet if one is available on the RTX track, otherwise returns nil
+func (r *RTPReceiver) readRTX(reader *TrackRemote) *rtxPacketWithAttributes {
+	if !reader.HasRTX() {
+		return nil
+	}
+
+	select {
+	case <-r.received:
+	default:
+		return nil
+	}
+
+	if t := r.streamsForTrack(reader); t != nil {
+		select {
+		case rtxPacketReceived := <-t.repairStreamChannel:
+			return &rtxPacketReceived
+		default:
+		}
+	}
+	return nil
+}

track_remote.go

@@ -24,6 +24,7 @@ type TrackRemote struct {
 	payloadType PayloadType
 	kind        RTPCodecType
 	ssrc        SSRC
+	rtxSsrc     SSRC
 	codec       RTPCodecParameters
 	params      RTPParameters
 	rid         string
@@ -33,10 +34,11 @@ type TrackRemote struct {
 	peekedAttributes interceptor.Attributes
 }
 
-func newTrackRemote(kind RTPCodecType, ssrc SSRC, rid string, receiver *RTPReceiver) *TrackRemote {
+func newTrackRemote(kind RTPCodecType, ssrc, rtxSsrc SSRC, rid string, receiver *RTPReceiver) *TrackRemote {
 	return &TrackRemote{
 		kind:     kind,
 		ssrc:     ssrc,
+		rtxSsrc:  rtxSsrc,
 		rid:      rid,
 		receiver: receiver,
 	}
@@ -125,13 +127,23 @@ func (t *TrackRemote) Read(b []byte) (n int, attributes interceptor.Attributes,
 		}
 	}
 
-	n, attributes, err = r.readRTP(b, t)
-	if err != nil {
-		return
+	// If there's a separate RTX track and an RTX packet is available, return that
+	if rtxPacketReceived := r.readRTX(t); rtxPacketReceived != nil {
+		n = copy(b, rtxPacketReceived.pkt)
+		attributes = rtxPacketReceived.attributes
+		rtxPacketReceived.release()
+		err = nil
+	} else {
+		// If there's no separate RTX track (or there's a separate RTX track but no RTX packet waiting), wait for and return
+		// a packet from the main track
+		n, attributes, err = r.readRTP(b, t)
+		if err != nil {
+			return
+		}
+		err = t.checkAndUpdateTrack(b)
 	}
 
-	err = t.checkAndUpdateTrack(b)
-	return
+	return n, attributes, err
 }
 
 // checkAndUpdateTrack checks payloadType for every incoming packet
@@ -197,3 +209,17 @@ func (t *TrackRemote) peek(b []byte) (n int, a interceptor.Attributes, err error
 func (t *TrackRemote) SetReadDeadline(deadline time.Time) error {
 	return t.receiver.setRTPReadDeadline(deadline, t)
 }
+
+// RtxSSRC returns the RTX SSRC for a track, or 0 if track does not have a separate RTX stream
+func (t *TrackRemote) RtxSSRC() SSRC {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+	return t.rtxSsrc
+}
+
+// HasRTX returns true if the track has a separate RTX stream
+func (t *TrackRemote) HasRTX() bool {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+	return t.rtxSsrc != 0
+}