Add Nack Interceptors

Add ResponderInterceptor which responds to NACK Requests

Add GeneratorInterceptor which generates NACK Requests

Author: masterada

chain.go

@@ -1,5 +1,3 @@
-// +build !js
-
 package interceptor
 
 // Chain is an interceptor that runs all child interceptors in order.

go.mod

@@ -3,6 +3,8 @@ module github.com/pion/interceptor
 go 1.15
 
 require (
-	github.com/pion/rtcp v1.2.4
+	github.com/pion/logging v0.2.2
+	github.com/pion/rtcp v1.2.6
 	github.com/pion/rtp v1.6.1
+	github.com/stretchr/testify v1.6.1
 )

go.sum

@@ -1,9 +1,11 @@
 github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+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/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.4 h1:NT3H5LkUGgaEapvp0HGik+a+CpflRF7KTD7H+o7OWIM=
-github.com/pion/rtcp v1.2.4/go.mod h1:52rMNPWFsjr39z9B9MhnkqhPLoeHTv1aN63o/42bWE0=
+github.com/pion/rtcp v1.2.6 h1:1zvwBbyd0TeEuuWftrd/4d++m+/kZSeiguxU61LFWpo=
+github.com/pion/rtcp v1.2.6/go.mod h1:52rMNPWFsjr39z9B9MhnkqhPLoeHTv1aN63o/42bWE0=
 github.com/pion/rtp v1.6.1 h1:2Y2elcVBrahYnHKN2X7rMHX/r1R4TEBMP1LaVu/wNhk=
 github.com/pion/rtp v1.6.1/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=

interceptor.go

@@ -1,5 +1,3 @@
-// +build !js
-
 // Package interceptor contains the Interceptor interface, with some useful interceptors that should be safe to use
 // in most cases.
 package interceptor

nack.go

@@ -1,5 +1,3 @@
-// +build !js
-
 package interceptor
 
 // NoOp is an Interceptor that does not modify any packets. It can embedded in other interceptors, so it's

noop.go

@@ -0,0 +1,7 @@
+// Package nack provides interceptors to implement sending and receiving negative acknowledgements
+package nack
+
+import "errors"
+
+// ErrInvalidSize is returned by newReceiveLog/newSendBuffer, when an incorrect buffer size is supplied.
+var ErrInvalidSize = errors.New("invalid buffer size")

pkg/nack/errors.go

@@ -0,0 +1,157 @@
+package nack
+
+import (
+	"math/rand"
+	"sync"
+	"time"
+
+	"github.com/pion/interceptor"
+	"github.com/pion/logging"
+	"github.com/pion/rtcp"
+	"github.com/pion/rtp"
+)
+
+// GeneratorInterceptor interceptor generates nack feedback messages.
+type GeneratorInterceptor struct {
+	interceptor.NoOp
+	size        uint16
+	skipLastN   uint16
+	interval    time.Duration
+	receiveLogs *sync.Map
+	m           sync.Mutex
+	wg          sync.WaitGroup
+	close       chan struct{}
+	log         logging.LeveledLogger
+}
+
+// NewGeneratorInterceptor returns a new GeneratorInterceptor interceptor
+func NewGeneratorInterceptor(opts ...GeneratorOption) (*GeneratorInterceptor, error) {
+	r := &GeneratorInterceptor{
+		NoOp:        interceptor.NoOp{},
+		size:        8192,
+		skipLastN:   0,
+		interval:    time.Millisecond * 100,
+		receiveLogs: &sync.Map{},
+		close:       make(chan struct{}),
+		log:         logging.NewDefaultLoggerFactory().NewLogger("nack_generator"),
+	}
+
+	for _, opt := range opts {
+		opt(r)
+	}
+
+	if _, err := newReceiveLog(r.size); err != nil {
+		return nil, err
+	}
+
+	return r, nil
+}
+
+// BindRTCPWriter lets you modify any outgoing RTCP packets. It is called once per PeerConnection. The returned method
+// will be called once per packet batch.
+func (n *GeneratorInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter {
+	n.m.Lock()
+	defer n.m.Unlock()
+	select {
+	case <-n.close:
+		// already closed
+		return writer
+	default:
+	}
+
+	n.wg.Add(1)
+
+	go n.loop(writer)
+
+	return writer
+}
+
+// BindRemoteStream lets you modify any incoming RTP packets. It is called once for per RemoteStream. The returned method
+// will be called once per rtp packet.
+func (n *GeneratorInterceptor) BindRemoteStream(info *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader {
+	hasNack := false
+	for _, fb := range info.RTCPFeedback {
+		if fb.Type == "nack" && fb.Parameter == "" {
+			hasNack = true
+		}
+	}
+
+	if !hasNack {
+		return reader
+	}
+
+	// error is already checked in NewGeneratorInterceptor
+	receiveLog, _ := newReceiveLog(n.size)
+	n.receiveLogs.Store(info.SSRC, receiveLog)
+
+	return interceptor.RTPReaderFunc(func() (*rtp.Packet, interceptor.Attributes, error) {
+		p, attr, err := reader.Read()
+		if err != nil {
+			return nil, nil, err
+		}
+
+		receiveLog.add(p.SequenceNumber)
+
+		return p, attr, nil
+	})
+}
+
+// UnbindLocalStream is called when the Stream is removed. It can be used to clean up any data related to that track.
+func (n *GeneratorInterceptor) UnbindLocalStream(info *interceptor.StreamInfo) {
+	n.receiveLogs.Delete(info.SSRC)
+}
+
+// Close closes the interceptor
+func (n *GeneratorInterceptor) Close() error {
+	defer n.wg.Wait()
+	n.m.Lock()
+	defer n.m.Unlock()
+
+	select {
+	case <-n.close:
+		// already closed
+		return nil
+	default:
+	}
+
+	close(n.close)
+
+	return nil
+}
+
+func (n *GeneratorInterceptor) loop(rtcpWriter interceptor.RTCPWriter) {
+	defer n.wg.Done()
+
+	senderSSRC := rand.Uint32() // #nosec
+
+	ticker := time.NewTicker(n.interval)
+	for {
+		select {
+		case <-ticker.C:
+			n.receiveLogs.Range(func(key, value interface{}) bool {
+				ssrc := key.(uint32)
+				receiveLog := value.(*receiveLog)
+
+				missing := receiveLog.missingSeqNumbers(n.skipLastN)
+				if len(missing) == 0 {
+					return true
+				}
+
+				nack := &rtcp.TransportLayerNack{
+					SenderSSRC: senderSSRC,
+					MediaSSRC:  ssrc,
+					Nacks:      rtcp.NackPairsFromSequenceNumbers(missing),
+				}
+
+				if _, err := rtcpWriter.Write([]rtcp.Packet{nack}, interceptor.Attributes{}); err != nil {
+					n.log.Warnf("failed sending nack: %+v", err)
+				}
+
+				return true
+			})
+
+		case <-n.close:
+			return
+		}
+	}
+}

pkg/nack/generator_interceptor.go

@@ -0,0 +1,70 @@
+package nack
+
+import (
+	"testing"
+	"time"
+
+	"github.com/pion/interceptor"
+	"github.com/pion/interceptor/internal/test"
+	"github.com/pion/logging"
+	"github.com/pion/rtcp"
+	"github.com/pion/rtp"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGeneratorInterceptor(t *testing.T) {
+	const interval = time.Millisecond * 10
+	i, err := NewGeneratorInterceptor(
+		GeneratorSize(64),
+		GeneratorSkipLastN(2),
+		GeneratorInterval(interval),
+		GeneratorLog(logging.NewDefaultLoggerFactory().NewLogger("test")),
+	)
+	assert.NoError(t, err)
+
+	stream := test.NewMockStream(&interceptor.StreamInfo{
+		SSRC:         1,
+		RTCPFeedback: []interceptor.RTCPFeedback{{Type: "nack"}},
+	}, i)
+	defer func() {
+		assert.NoError(t, stream.Close())
+	}()
+
+	for _, seqNum := range []uint16{10, 11, 12, 14, 16, 18} {
+		stream.ReceiveRTP(&rtp.Packet{Header: rtp.Header{SequenceNumber: seqNum}})
+
+		select {
+		case r := <-stream.ReadRTP():
+			assert.NoError(t, r.Err)
+			assert.Equal(t, seqNum, r.Packet.SequenceNumber)
+		case <-time.After(10 * time.Millisecond):
+			t.Fatal("receiver rtp packet not found")
+		}
+	}
+
+	time.Sleep(interval * 2) // wait for at least 2 nack packets
+
+	select {
+	case <-stream.WrittenRTCP():
+		// ignore the first nack, it might only contain the sequence id 13 as missing
+	default:
+	}
+
+	select {
+	case pkts := <-stream.WrittenRTCP():
+		assert.Equal(t, len(pkts), 1, "single packet RTCP Compound Packet expected")
+
+		p, ok := pkts[0].(*rtcp.TransportLayerNack)
+		assert.True(t, ok, "TransportLayerNack rtcp packet expected, found: %T", pkts[0])
+
+		assert.Equal(t, uint16(13), p.Nacks[0].PacketID)
+		assert.Equal(t, rtcp.PacketBitmap(0b10), p.Nacks[0].LostPackets) // we want packets: 13, 15 (not packet 17, because skipLastN is setReceived to 2)
+	case <-time.After(10 * time.Millisecond):
+		t.Fatal("written rtcp packet not found")
+	}
+}
+
+func TestGeneratorInterceptor_InvalidSize(t *testing.T) {
+	_, err := NewGeneratorInterceptor(GeneratorSize(5))
+	assert.Error(t, err, ErrInvalidSize)
+}

pkg/nack/generator_interceptor_test.go

@@ -0,0 +1,40 @@
+package nack
+
+import (
+	"time"
+
+	"github.com/pion/logging"
+)
+
+// GeneratorOption can be used to configure GeneratorInterceptor
+type GeneratorOption func(r *GeneratorInterceptor)
+
+// GeneratorSize sets the size of the interceptor.
+// Size must be one of: 64, 128, 256, 512, 1024, 2048, 4096, 8192, 16384, 32768
+func GeneratorSize(size uint16) GeneratorOption {
+	return func(r *GeneratorInterceptor) {
+		r.size = size
+	}
+}
+
+// GeneratorSkipLastN sets the number of packets (n-1 packets before the last received packets) to ignore when generating
+// nack requests.
+func GeneratorSkipLastN(skipLastN uint16) GeneratorOption {
+	return func(r *GeneratorInterceptor) {
+		r.skipLastN = skipLastN
+	}
+}
+
+// GeneratorLog sets a logger for the interceptor
+func GeneratorLog(log logging.LeveledLogger) GeneratorOption {
+	return func(r *GeneratorInterceptor) {
+		r.log = log
+	}
+}
+
+// GeneratorInterval sets the nack send interval for the interceptor
+func GeneratorInterval(interval time.Duration) GeneratorOption {
+	return func(r *GeneratorInterceptor) {
+		r.interval = interval
+	}
+}