Repository: pion/interceptor
Pull request #229 (2024-02-19 01:15:39)

Support TrackLocal RTX
#### Description

Support TrackLocal RTX




codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pion/interceptor/pull/229?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) Report
Attention: `41 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`5574fda`)](https://app.codecov.io/gh/pion/interceptor/commit/5574fda4dd5cc44b2c2e51fbfe885d747c37f45f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) 79.63% compared to head [(`e530b47`)](https://app.codecov.io/gh/pion/interceptor/pull/229?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) 77.02%.

| [Files](https://app.codecov.io/gh/pion/interceptor/pull/229?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | Patch % | Lines |
|---|---|---|
| [pkg/nack/responder\_interceptor.go](https://app.codecov.io/gh/pion/interceptor/pull/229?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL25hY2svcmVzcG9uZGVyX2ludGVyY2VwdG9yLmdv) | 23.07% | [36 Missing and 4 partials :warning: ](https://app.codecov.io/gh/pion/interceptor/pull/229?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) |
| [pkg/nack/nack.go](https://app.codecov.io/gh/pion/interceptor/pull/229?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL25hY2svbmFjay5nbw==) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/pion/interceptor/pull/229?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master     #229      +/-   ##
==========================================
- Coverage   79.63%   77.02%   -2.62%     
==========================================
  Files          70       70              
  Lines        3791     3830      +39     
==========================================
- Hits         3019     2950      -69     
- Misses        637      735      +98     
- Partials      135      145      +10     
```

| [Flag](https://app.codecov.io/gh/pion/interceptor/pull/229/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | Coverage Δ | |
|---|---|---|
| [go](https://app.codecov.io/gh/pion/interceptor/pull/229/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | `?` | |
| [wasm](https://app.codecov.io/gh/pion/interceptor/pull/229/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | `77.02% <22.64%> (-0.61%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#carryforward-flags-in-the-pull-request-comment) to find out more.


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/pion/interceptor/pull/229?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).


Patch:
diff --git a/pkg/nack/responder_interceptor.go b/pkg/nack/responder_interceptor.go
index 03b084d..1d5745f 100644
--- a/pkg/nack/responder_interceptor.go
+++ b/pkg/nack/responder_interceptor.go
@@ -4,6 +4,7 @@
 package nack
 
 import (
+	"encoding/binary"
 	"sync"
 
 	"github.com/pion/interceptor"
@@ -62,6 +63,11 @@ type ResponderInterceptor struct {
 type localStream struct {
 	sendBuffer *sendBuffer
 	rtpWriter  interceptor.RTPWriter
+
+	// Non-zero if Retransmissions should be sent on a distinct stream
+	rtxSsrc        uint32
+	rtxPayloadType uint8
+	rtxSequencer   rtp.Sequencer
 }
 
 // NewResponderInterceptor returns a new ResponderInterceptorFactor
@@ -108,7 +114,13 @@ func (n *ResponderInterceptor) BindLocalStream(info *interceptor.StreamInfo, wri
 	// error is already checked in NewGeneratorInterceptor
 	sendBuffer, _ := newSendBuffer(n.size)
 	n.streamsMu.Lock()
-	n.streams[info.SSRC] = &localStream{sendBuffer: sendBuffer, rtpWriter: writer}
+	n.streams[info.SSRC] = &localStream{
+		sendBuffer:     sendBuffer,
+		rtpWriter:      writer,
+		rtxSsrc:        info.SSRCRetransmission,
+		rtxPayloadType: info.PayloadTypeRetransmission,
+		rtxSequencer:   rtp.NewRandomSequencer(),
+	}
 	n.streamsMu.Unlock()
 
 	return interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
@@ -139,8 +151,43 @@ func (n *ResponderInterceptor) resendPackets(nack *rtcp.TransportLayerNack) {
 	for i := range nack.Nacks {
 		nack.Nacks[i].Range(func(seq uint16) bool {
 			if p := stream.sendBuffer.get(seq); p != nil {
-				if _, err := stream.rtpWriter.Write(p.Header(), p.Payload(), interceptor.Attributes{}); err != nil {
-					n.log.Warnf("failed resending nacked packet: %+v", err)
+				if stream.rtxSsrc != 0 {
+					// Store the original sequence number and rewrite the sequence number.
+					originalSequenceNumber := p.Header().SequenceNumber
+					p.Header().SequenceNumber = stream.rtxSequencer.NextSequenceNumber()
+
+					// Rewrite the SSRC.
+					p.Header().SSRC = stream.rtxSsrc
+					// Rewrite the payload type.
+					p.Header().PayloadType = stream.rtxPayloadType
+
+					// Remove padding if present.
+					paddingLength := 0
+					originPayload := p.Payload()
+					if p.Header().Padding {
+						paddingLength = int(originPayload[len(originPayload)-1])
+						p.Header().Padding = false
+					}
+
+					// Write the original sequence number at the beginning of the payload.
+					payload := make([]byte, 2)
+					binary.BigEndian.PutUint16(payload, originalSequenceNumber)
+					payload = append(payload, originPayload[:len(originPayload)-paddingLength]...)
+
+					// Send RTX packet.
+					if _, err := stream.rtpWriter.Write(p.Header(), payload, interceptor.Attributes{}); err != nil {
+						n.log.Warnf("failed sending rtx packet: %+v", err)
+					}
+
+					// Resore the Padding and SSRC.
+					if paddingLength > 0 {
+						p.Header().Padding = true
+					}
+					p.Header().SequenceNumber = originalSequenceNumber
+				} else {
+					if _, err := stream.rtpWriter.Write(p.Header(), p.Payload(), interceptor.Attributes{}); err != nil {
+						n.log.Warnf("failed resending nacked packet: %+v", err)
+					}
 				}
 				p.Release()
 			}


Test patch:
diff --git a/pkg/nack/responder_interceptor_test.go b/pkg/nack/responder_interceptor_test.go
index 9eb5b23..68e88d1 100644
--- a/pkg/nack/responder_interceptor_test.go
+++ b/pkg/nack/responder_interceptor_test.go
@@ -4,6 +4,7 @@
 package nack
 
 import (
+	"encoding/binary"
 	"testing"
 	"time"
 
@@ -231,3 +232,60 @@ func TestResponderInterceptor_StreamFilter(t *testing.T) {
 	case <-time.After(10 * time.Millisecond):
 	}
 }
+
+func TestResponderInterceptor_RFC4588(t *testing.T) {
+	f, err := NewResponderInterceptor()
+	require.NoError(t, err)
+
+	i, err := f.NewInterceptor("")
+	require.NoError(t, err)
+
+	stream := test.NewMockStream(&interceptor.StreamInfo{
+		SSRC:                      1,
+		SSRCRetransmission:        2,
+		PayloadTypeRetransmission: 2,
+		RTCPFeedback:              []interceptor.RTCPFeedback{{Type: "nack"}},
+	}, i)
+	defer func() {
+		require.NoError(t, stream.Close())
+	}()
+
+	for _, seqNum := range []uint16{10, 11, 12, 14, 15} {
+		require.NoError(t, stream.WriteRTP(&rtp.Packet{Header: rtp.Header{SequenceNumber: seqNum}}))
+
+		select {
+		case p := <-stream.WrittenRTP():
+			require.Equal(t, seqNum, p.SequenceNumber)
+		case <-time.After(10 * time.Millisecond):
+			t.Fatal("written rtp packet not found")
+		}
+	}
+
+	stream.ReceiveRTCP([]rtcp.Packet{
+		&rtcp.TransportLayerNack{
+			MediaSSRC:  1,
+			SenderSSRC: 2,
+			Nacks: []rtcp.NackPair{
+				{PacketID: 11, LostPackets: 0b1011}, // sequence numbers: 11, 12, 13, 15
+			},
+		},
+	})
+
+	// seq number 13 was never sent, so it can't be resent
+	for _, seqNum := range []uint16{11, 12, 15} {
+		select {
+		case p := <-stream.WrittenRTP():
+			require.Equal(t, uint32(2), p.SSRC)
+			require.Equal(t, uint8(2), p.PayloadType)
+			require.Equal(t, binary.BigEndian.Uint16(p.Payload), seqNum)
+		case <-time.After(10 * time.Millisecond):
+			t.Fatal("written rtp packet not found")
+		}
+	}
+
+	select {
+	case p := <-stream.WrittenRTP():
+		t.Errorf("no more rtp packets expected, found sequence number: %v", p.SequenceNumber)
+	case <-time.After(10 * time.Millisecond):
+	}
+}