Handle non-Simulcast Repair Streams

Same issue with TWCC enabled as 11b887. We need to process the
RTX packets so that we can emit proper reports.

peerconnection.go

@@ -1171,6 +1171,8 @@ func (pc *PeerConnection) startReceiver(incoming trackDetails, receiver *RTPRece
 		if len(incoming.ssrcs) > i {
 			encodings[i].SSRC = incoming.ssrcs[i]
 		}
+
+		encodings[i].RTX.SSRC = incoming.repairSsrc
 	}
 
 	if err := receiver.Receive(RTPReceiveParameters{Encodings: encodings}); err != nil {
@@ -1451,7 +1453,9 @@ func (pc *PeerConnection) handleIncomingSSRC(rtpStream io.Reader, ssrc SSRC) err
 			}
 
 			if rsid != "" {
-				return receiver.receiveForRsid(rsid, streamInfo, readStream, interceptor, rtcpReadStream, rtcpInterceptor)
+				receiver.mu.Lock()
+				defer receiver.mu.Unlock()
+				return receiver.receiveForRtx(SSRC(0), rsid, streamInfo, readStream, interceptor, rtcpReadStream, rtcpInterceptor)
 			}
 
 			track, err := receiver.receiveForRid(rid, params, streamInfo, readStream, interceptor, rtcpReadStream, rtcpInterceptor)

rtpcodingparameters.go

@@ -1,10 +1,17 @@
 package webrtc
 
+// RTPRtxParameters dictionary contains information relating to retransmission (RTX) settings.
+// https://draft.ortc.org/#dom-rtcrtprtxparameters
+type RTPRtxParameters struct {
+	SSRC SSRC `json:"ssrc"`
+}
+
 // RTPCodingParameters provides information relating to both encoding and decoding.
 // This is a subset of the RFC since Pion WebRTC doesn't implement encoding/decoding itself
 // http://draft.ortc.org/#dom-rtcrtpcodingparameters
 type RTPCodingParameters struct {
-	RID         string      `json:"rid"`
-	SSRC        SSRC        `json:"ssrc"`
-	PayloadType PayloadType `json:"payloadType"`
+	RID         string           `json:"rid"`
+	SSRC        SSRC             `json:"ssrc"`
+	PayloadType PayloadType      `json:"payloadType"`
+	RTX         RTPRtxParameters `json:"rtx"`
 }

rtpreceiver.go

@@ -158,6 +158,18 @@ func (r *RTPReceiver) Receive(parameters RTPReceiveParameters) error {
 		}
 
 		r.tracks = append(r.tracks, t)
+
+		if rtxSsrc := parameters.Encodings[i].RTX.SSRC; rtxSsrc != 0 {
+			streamInfo := createStreamInfo("", rtxSsrc, 0, codec, globalParams.HeaderExtensions)
+			rtpReadStream, rtpInterceptor, rtcpReadStream, rtcpInterceptor, err := r.transport.streamsForSSRC(rtxSsrc, *streamInfo)
+			if err != nil {
+				return err
+			}
+
+			if err = r.receiveForRtx(rtxSsrc, "", streamInfo, rtpReadStream, rtpInterceptor, rtcpReadStream, rtcpInterceptor); err != nil {
+				return err
+			}
+		}
 	}
 
 	return nil
@@ -323,37 +335,40 @@ func (r *RTPReceiver) receiveForRid(rid string, params RTPParameters, streamInfo
 	return nil, fmt.Errorf("%w: %s", errRTPReceiverForRIDTrackStreamNotFound, rid)
 }
 
-// receiveForRsid starts a routine that processes the repair stream for a RID
+// 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) receiveForRsid(rsid string, streamInfo *interceptor.StreamInfo, rtpReadStream *srtp.ReadStreamSRTP, rtpInterceptor interceptor.RTPReader, rtcpReadStream *srtp.ReadStreamSRTCP, rtcpInterceptor interceptor.RTCPReader) error {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-
-	for i := range r.tracks {
-		if r.tracks[i].track.RID() == rsid {
-			var err error
-
-			r.tracks[i].repairStreamInfo = streamInfo
-			r.tracks[i].repairReadStream = rtpReadStream
-			r.tracks[i].repairInterceptor = rtpInterceptor
-			r.tracks[i].repairRtcpReadStream = rtcpReadStream
-			r.tracks[i].repairRtcpInterceptor = rtcpInterceptor
-
-			go func() {
-				b := make([]byte, r.api.settingEngine.getReceiveMTU())
-				for {
-					if _, _, readErr := r.tracks[i].repairInterceptor.Read(b, nil); readErr != nil {
-						return
-					}
-				}
-			}()
-
-			return err
+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 {
+		track = &r.tracks[0]
+	} else {
+		for i := range r.tracks {
+			if r.tracks[i].track.RID() == rsid {
+				track = &r.tracks[i]
+			}
 		}
 	}
 
-	return fmt.Errorf("%w: %s", errRTPReceiverForRIDTrackStreamNotFound, rsid)
+	if track == nil {
+		return fmt.Errorf("%w: ssrc(%d) rsid(%s)", errRTPReceiverForRIDTrackStreamNotFound, ssrc, rsid)
+	}
+
+	track.repairStreamInfo = streamInfo
+	track.repairReadStream = rtpReadStream
+	track.repairInterceptor = rtpInterceptor
+	track.repairRtcpReadStream = rtcpReadStream
+	track.repairRtcpInterceptor = rtcpInterceptor
+
+	go func() {
+		b := make([]byte, r.api.settingEngine.getReceiveMTU())
+		for {
+			if _, _, readErr := track.repairInterceptor.Read(b, nil); readErr != nil {
+				return
+			}
+		}
+	}()
+	return nil
 }
 
 // SetReadDeadline sets the max amount of time the RTCP stream will block before returning. 0 is forever.

sdp.go

@@ -18,12 +18,13 @@ import (
 // trackDetails represents any media source that can be represented in a SDP
 // This isn't keyed by SSRC because it also needs to support rid based sources
 type trackDetails struct {
-	mid      string
-	kind     RTPCodecType
-	streamID string
-	id       string
-	ssrcs    []SSRC
-	rids     []string
+	mid        string
+	kind       RTPCodecType
+	streamID   string
+	id         string
+	ssrcs      []SSRC
+	repairSsrc SSRC
+	rids       []string
 }
 
 func trackDetailsForSSRC(trackDetails []trackDetails, ssrc SSRC) *trackDetails {
@@ -73,7 +74,7 @@ func filterTrackWithSSRC(incomingTracks []trackDetails, ssrc SSRC) []trackDetail
 func trackDetailsFromSDP(log logging.LeveledLogger, s *sdp.SessionDescription) (incomingTracks []trackDetails) { // nolint:gocognit
 	for _, media := range s.MediaDescriptions {
 		tracksInMediaSection := []trackDetails{}
-		rtxRepairFlows := map[uint32]bool{}
+		rtxRepairFlows := map[uint64]uint64{}
 
 		// Plan B can have multiple tracks in a signle media section
 		streamID := ""
@@ -106,7 +107,7 @@ func trackDetailsFromSDP(log logging.LeveledLogger, s *sdp.SessionDescription) (
 					// 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 {
-						_, err := strconv.ParseUint(split[1], 10, 32)
+						baseSsrc, err := strconv.ParseUint(split[1], 10, 32)
 						if err != nil {
 							log.Warnf("Failed to parse SSRC: %v", err)
 							continue
@@ -116,7 +117,7 @@ func trackDetailsFromSDP(log logging.LeveledLogger, s *sdp.SessionDescription) (
 							log.Warnf("Failed to parse SSRC: %v", err)
 							continue
 						}
-						rtxRepairFlows[uint32(rtxRepairFlow)] = true
+						rtxRepairFlows[rtxRepairFlow] = baseSsrc
 						tracksInMediaSection = filterTrackWithSSRC(tracksInMediaSection, SSRC(rtxRepairFlow)) // Remove if rtx was added as track before
 					}
 				}
@@ -139,7 +140,7 @@ func trackDetailsFromSDP(log logging.LeveledLogger, s *sdp.SessionDescription) (
 					continue
 				}
 
-				if rtxRepairFlow := rtxRepairFlows[uint32(ssrc)]; rtxRepairFlow {
+				if _, ok := rtxRepairFlows[ssrc]; ok {
 					continue // This ssrc is a RTX repair flow, ignore
 				}
 
@@ -165,6 +166,12 @@ func trackDetailsFromSDP(log logging.LeveledLogger, s *sdp.SessionDescription) (
 				trackDetails.id = trackID
 				trackDetails.ssrcs = []SSRC{SSRC(ssrc)}
 
+				for repairSsrc, baseSsrc := range rtxRepairFlows {
+					if baseSsrc == ssrc {
+						trackDetails.repairSsrc = SSRC(repairSsrc)
+					}
+				}
+
 				if isNewTrack {
 					tracksInMediaSection = append(tracksInMediaSection, *trackDetails)
 				}