Repository: pion/interceptor
Pull request #4 (2020-11-27 16:57:23)

Added Nack interceptors
Ref: https://github.com/pion/interceptor/issues/3


codecov[bot]: # [Codecov](https://codecov.io/gh/pion/interceptor/pull/4?src=pr&el=h1) Report
> Merging [#4](https://codecov.io/gh/pion/interceptor/pull/4?src=pr&el=desc) (4174c4b) into [master](https://codecov.io/gh/pion/interceptor/commit/1e87aa9abf785b2121bf5bc642c015158215640e?el=desc) (1e87aa9) will **increase** coverage by `47.94%`.
> The diff coverage is `80.53%`.

[![Impacted file tree graph](https://codecov.io/gh/pion/interceptor/pull/4/graphs/tree.svg?width=650&height=150&src=pr&token=FNH9xUG6AX)](https://codecov.io/gh/pion/interceptor/pull/4?src=pr&el=tree)

```diff
@@             Coverage Diff             @@
##           master       #4       +/-   ##
===========================================
+ Coverage   23.52%   71.47%   +47.94%     
===========================================
  Files           6       10        +4     
  Lines          51      312      +261     
===========================================
+ Hits           12      223      +211     
- Misses         36       65       +29     
- Partials        3       24       +21     
```

| Flag | Coverage Δ | |
|---|---|---|
| go | `?` | |
| wasm | `71.47% <80.53%> (+4.80%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/pion/interceptor/pull/4?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [chain.go](https://codecov.io/gh/pion/interceptor/pull/4/diff?src=pr&el=tree#diff-Y2hhaW4uZ28=) | `0.00% <ø> (ø)` | |
| [interceptor.go](https://codecov.io/gh/pion/interceptor/pull/4/diff?src=pr&el=tree#diff-aW50ZXJjZXB0b3IuZ28=) | `0.00% <ø> (ø)` | |
| [noop.go](https://codecov.io/gh/pion/interceptor/pull/4/diff?src=pr&el=tree#diff-bm9vcC5nbw==) | `0.00% <ø> (ø)` | |
| [registry.go](https://codecov.io/gh/pion/interceptor/pull/4/diff?src=pr&el=tree#diff-cmVnaXN0cnkuZ28=) | `0.00% <ø> (ø)` | |
| [nack/receiver\_interceptor.go](https://codecov.io/gh/pion/interceptor/pull/4/diff?src=pr&el=tree#diff-bmFjay9yZWNlaXZlcl9pbnRlcmNlcHRvci5nbw==) | `76.00% <76.00%> (ø)` | |
| [nack/sender\_interceptor.go](https://codecov.io/gh/pion/interceptor/pull/4/diff?src=pr&el=tree#diff-bmFjay9zZW5kZXJfaW50ZXJjZXB0b3IuZ28=) | `76.59% <76.59%> (ø)` | |
| [nack/send\_buffer.go](https://codecov.io/gh/pion/interceptor/pull/4/diff?src=pr&el=tree#diff-bmFjay9zZW5kX2J1ZmZlci5nbw==) | `76.66% <76.66%> (ø)` | |
| [nack/receive\_log.go](https://codecov.io/gh/pion/interceptor/pull/4/diff?src=pr&el=tree#diff-bmFjay9yZWNlaXZlX2xvZy5nbw==) | `83.33% <83.33%> (ø)` | |
| [test/stream.go](https://codecov.io/gh/pion/interceptor/pull/4/diff?src=pr&el=tree#diff-dGVzdC9zdHJlYW0uZ28=) | `89.28% <89.28%> (ø)` | |
| ... and [4 more](https://codecov.io/gh/pion/interceptor/pull/4/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pion/interceptor/pull/4?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/pion/interceptor/pull/4?src=pr&el=footer). Last update [1e87aa9...4174c4b](https://codecov.io/gh/pion/interceptor/pull/4?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

Sean-Der: @masterada this is really fantastic! I wish I had this 5 years ago when learning. The code is so clear, I can't wait for people to go read this and understand RTC in a week instead of years :) 
Sean-Der: What do you think of doing a functional pattern for config, then we can add/remove options as we please. @at-wat did this in `pion/srtp` https://github.com/pion/srtp/blob/master/context.go#L61

In the future I think we can expand things a lot more!
* Distinct stream for NACKs
* Do dynamic instead of static etc...

Or we can have distinict objects for the different types.
Sean-Der: @masterada rebased off of master, reviewing again!
Sean-Der: @masterada this is really great, approved! I especially love how readable the tests are. I think the Public API is rock solid, and the internals are good. Lots of WaitGroup, Mutex and Channels make for challenging code, but that is the problem space!

I think if we have this + Receiver/Sender reports and enable them by default that would make a perfect `v3.0.0`

@at-wat would love if you could review/see if this captures the spirit? This will be a blueprint for lots of future work, so would love to have it. 


Sean-Der: @masterada If you don't have the availability I am happy to pull in the suggested changes :)

No rush, I just don't want to overload you! What do you think of having `pion/webrtc` go out with Receiver Reports + NACK by default? You think good/bad idea.

Or should we push `v3.0.0` with no interceptors, and do a minor release with them?
masterada: @Sean-Der sorry, I just started work at a new place and it comsumes more energy than I expected. Not sure when I could finish these comments, probably only next week (my weekend is kinda budy too). Please if you have the time feel free to finish this up. Also there is a TODO to move sequence numbers -> nack pair conversion to rtcp packet, because the other way is alread there. It's not really important, can be done sometime later too.

I replied to 2 comments @at-wat , other than that I agree with you in everything.

> What do you think of having pion/webrtc go out with Receiver Reports + NACK by default? You think good/bad idea.

I think it would be nice to add at least 1 interceptor to 3.0.0, so we can be sure the interceptor integration works, and also create the RegisterDefaultInterceptors method that actually does something. RR/SR and TCC would be nice too but they can come at 3.1 too or something.

Also note that in order for nack.GeneratorInterceptor to work, you need to call s.SetSRTPReplayProtectionWindow(AT_LEAST_NACK_SIZE), see: https://github.com/pion/webrtc/pull/1549/files#diff-6e2e56e295d82766f7fdce880ab65f299e92aede458043e232610c9fe9166e03R58

Maybe a higher default value for that could work? I think google uses 10000 for the replay protection window size, not sure though, their c code is very hard to follow.
Sean-Der: @masterada congrats on the new job! No worries at all, you have done some amazing work here. I am more then happy to finish out the rest :) 

Sean-Der: @masterada @at-wat mind giving it another look? I went through and address everything in your reviews.

thanks!
masterada: LGTM, but i think this should wait until the raw read and attributes changes are in. 

Patch:
diff --git a/go.mod b/go.mod
index a1e57d6..067c0d4 100644
--- a/go.mod
+++ b/go.mod
@@ -3,6 +3,7 @@ module github.com/pion/interceptor
 go 1.15
 
 require (
+	github.com/pion/logging v0.2.2
 	github.com/pion/rtcp v1.2.6
 	github.com/pion/rtp v1.6.2
 	github.com/stretchr/testify v1.6.1
diff --git a/go.sum b/go.sum
index 33c52ce..d927bde 100644
--- a/go.sum
+++ b/go.sum
@@ -1,5 +1,7 @@
 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.6 h1:1zvwBbyd0TeEuuWftrd/4d++m+/kZSeiguxU61LFWpo=
diff --git a/nack.go b/nack.go
deleted file mode 100644
index 736f562..0000000
--- a/nack.go
+++ /dev/null
@@ -1,12 +0,0 @@
-package interceptor
-
-// NACK interceptor generates/responds to nack messages.
-type NACK struct {
-	NoOp
-}
-
-// 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 *NACK) BindRemoteStream(_ *StreamInfo, reader RTPReader) RTPReader {
-	return reader
-}
diff --git a/pkg/nack/errors.go b/pkg/nack/errors.go
new file mode 100644
index 0000000..bbfc773
--- /dev/null
+++ b/pkg/nack/errors.go
@@ -0,0 +1,6 @@
+package nack
+
+import "errors"
+
+// ErrInvalidSize is returned by newReceiveLog/newSendBuffer, when an incorrect buffer size is supplied.
+var ErrInvalidSize = errors.New("invalid buffer size")
diff --git a/pkg/nack/generator_interceptor.go b/pkg/nack/generator_interceptor.go
new file mode 100644
index 0000000..b02fa90
--- /dev/null
+++ b/pkg/nack/generator_interceptor.go
@@ -0,0 +1,158 @@
+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
+
+	remoteStreamBuf rtp.Packet
+}
+
+// NewGeneratorInterceptor returns a new GeneratorInterceptor interceptor
+func NewGeneratorInterceptor(opts ...GeneratorOption) (*GeneratorInterceptor, error) {
+	r := &GeneratorInterceptor{
+		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 {
+		if err := opt(r); err != nil {
+			return nil, err
+		}
+	}
+
+	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()
+
+	if n.isClosed() {
+		return writer
+	}
+
+	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 {
+	if !streamSupportNack(info) {
+		return reader
+	}
+
+	// error is already checked in NewGeneratorInterceptor
+	receiveLog, _ := newReceiveLog(n.size)
+	n.receiveLogs.Store(info.SSRC, receiveLog)
+
+	return interceptor.RTPReaderFunc(func(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) {
+		i, attr, err := reader.Read(b, a)
+		if err != nil {
+			return 0, nil, err
+		}
+
+		if err = n.remoteStreamBuf.Unmarshal(b[:i]); err != nil {
+			return 0, nil, err
+		}
+		receiveLog.add(n.remoteStreamBuf.Header.SequenceNumber)
+
+		return i, 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()
+
+	if !n.isClosed() {
+		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
+		}
+	}
+}
+
+func (n *GeneratorInterceptor) isClosed() bool {
+	select {
+	case <-n.close:
+		return true
+	default:
+		return false
+	}
+}
diff --git a/pkg/nack/generator_option.go b/pkg/nack/generator_option.go
new file mode 100644
index 0000000..092f5db
--- /dev/null
+++ b/pkg/nack/generator_option.go
@@ -0,0 +1,44 @@
+package nack
+
+import (
+	"time"
+
+	"github.com/pion/logging"
+)
+
+// GeneratorOption can be used to configure GeneratorInterceptor
+type GeneratorOption func(r *GeneratorInterceptor) error
+
+// 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) error {
+		r.size = size
+		return nil
+	}
+}
+
+// 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) error {
+		r.skipLastN = skipLastN
+		return nil
+	}
+}
+
+// GeneratorLog sets a logger for the interceptor
+func GeneratorLog(log logging.LeveledLogger) GeneratorOption {
+	return func(r *GeneratorInterceptor) error {
+		r.log = log
+		return nil
+	}
+}
+
+// GeneratorInterval sets the nack send interval for the interceptor
+func GeneratorInterval(interval time.Duration) GeneratorOption {
+	return func(r *GeneratorInterceptor) error {
+		r.interval = interval
+		return nil
+	}
+}
diff --git a/pkg/nack/nack.go b/pkg/nack/nack.go
new file mode 100644
index 0000000..a658e7f
--- /dev/null
+++ b/pkg/nack/nack.go
@@ -0,0 +1,14 @@
+// Package nack provides interceptors to implement sending and receiving negative acknowledgements
+package nack
+
+import "github.com/pion/interceptor"
+
+func streamSupportNack(info *interceptor.StreamInfo) bool {
+	for _, fb := range info.RTCPFeedback {
+		if fb.Type == "nack" && fb.Parameter == "" {
+			return true
+		}
+	}
+
+	return false
+}
diff --git a/pkg/nack/receive_log.go b/pkg/nack/receive_log.go
new file mode 100644
index 0000000..8107f59
--- /dev/null
+++ b/pkg/nack/receive_log.go
@@ -0,0 +1,134 @@
+package nack
+
+import (
+	"fmt"
+	"sync"
+)
+
+type receiveLog struct {
+	packets         []uint64
+	size            uint16
+	end             uint16
+	started         bool
+	lastConsecutive uint16
+	m               sync.RWMutex
+}
+
+func newReceiveLog(size uint16) (*receiveLog, error) {
+	allowedSizes := make([]uint16, 0)
+	correctSize := false
+	for i := 6; i < 16; i++ {
+		if size == 1<<i {
+			correctSize = true
+			break
+		}
+		allowedSizes = append(allowedSizes, 1<<i)
+	}
+
+	if !correctSize {
+		return nil, fmt.Errorf("%w: %d is not a valid size, allowed sizes: %v", ErrInvalidSize, size, allowedSizes)
+	}
+
+	return &receiveLog{
+		packets: make([]uint64, size/64),
+		size:    size,
+	}, nil
+}
+
+func (s *receiveLog) add(seq uint16) {
+	s.m.Lock()
+	defer s.m.Unlock()
+
+	if !s.started {
+		s.setReceived(seq)
+		s.end = seq
+		s.started = true
+		s.lastConsecutive = seq
+		return
+	}
+
+	diff := seq - s.end
+	switch {
+	case diff == 0:
+		return
+	case diff < uint16SizeHalf:
+		// this means a positive diff, in other words seq > end (with counting for rollovers)
+		for i := s.end + 1; i != seq; i++ {
+			// clear packets between end and seq (these may contain packets from a "size" ago)
+			s.delReceived(i)
+		}
+		s.end = seq
+
+		if s.lastConsecutive+1 == seq {
+			s.lastConsecutive = seq
+		} else if seq-s.lastConsecutive > s.size {
+			s.lastConsecutive = seq - s.size
+			s.fixLastConsecutive() // there might be valid packets at the beginning of the buffer now
+		}
+	case s.lastConsecutive+1 == seq:
+		// negative diff, seq < end (with counting for rollovers)
+		s.lastConsecutive = seq
+		s.fixLastConsecutive() // there might be other valid packets after seq
+	}
+
+	s.setReceived(seq)
+}
+
+func (s *receiveLog) get(seq uint16) bool {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	diff := s.end - seq
+	if diff >= uint16SizeHalf {
+		return false
+	}
+
+	if diff >= s.size {
+		return false
+	}
+
+	return s.getReceived(seq)
+}
+
+func (s *receiveLog) missingSeqNumbers(skipLastN uint16) []uint16 {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	until := s.end - skipLastN
+	if until-s.lastConsecutive >= uint16SizeHalf {
+		// until < s.lastConsecutive (counting for rollover)
+		return nil
+	}
+
+	missingPacketSeqNums := make([]uint16, 0)
+	for i := s.lastConsecutive + 1; i != until+1; i++ {
+		if !s.getReceived(i) {
+			missingPacketSeqNums = append(missingPacketSeqNums, i)
+		}
+	}
+
+	return missingPacketSeqNums
+}
+
+func (s *receiveLog) setReceived(seq uint16) {
+	pos := seq % s.size
+	s.packets[pos/64] |= 1 << (pos % 64)
+}
+
+func (s *receiveLog) delReceived(seq uint16) {
+	pos := seq % s.size
+	s.packets[pos/64] &^= 1 << (pos % 64)
+}
+
+func (s *receiveLog) getReceived(seq uint16) bool {
+	pos := seq % s.size
+	return (s.packets[pos/64] & (1 << (pos % 64))) != 0
+}
+
+func (s *receiveLog) fixLastConsecutive() {
+	i := s.lastConsecutive + 1
+	for ; i != s.end+1 && s.getReceived(i); i++ {
+		// find all consecutive packets
+	}
+	s.lastConsecutive = i - 1
+}
diff --git a/pkg/nack/responder_interceptor.go b/pkg/nack/responder_interceptor.go
new file mode 100644
index 0000000..d054a95
--- /dev/null
+++ b/pkg/nack/responder_interceptor.go
@@ -0,0 +1,114 @@
+package nack
+
+import (
+	"sync"
+
+	"github.com/pion/interceptor"
+	"github.com/pion/logging"
+	"github.com/pion/rtcp"
+	"github.com/pion/rtp"
+)
+
+// ResponderInterceptor responds to nack feedback messages
+type ResponderInterceptor struct {
+	interceptor.NoOp
+	size    uint16
+	streams *sync.Map
+	log     logging.LeveledLogger
+}
+
+type localStream struct {
+	sendBuffer *sendBuffer
+	rtpWriter  interceptor.RTPWriter
+}
+
+// NewResponderInterceptor returns a new GeneratorInterceptor interceptor
+func NewResponderInterceptor(opts ...ResponderOption) (*ResponderInterceptor, error) {
+	r := &ResponderInterceptor{
+		size:    8192,
+		streams: &sync.Map{},
+		log:     logging.NewDefaultLoggerFactory().NewLogger("nack_responder"),
+	}
+
+	for _, opt := range opts {
+		if err := opt(r); err != nil {
+			return nil, err
+		}
+	}
+
+	if _, err := newSendBuffer(r.size); err != nil {
+		return nil, err
+	}
+
+	return r, nil
+}
+
+// BindRTCPReader lets you modify any incoming RTCP packets. It is called once per sender/receiver, however this might
+// change in the future. The returned method will be called once per packet batch.
+func (n *ResponderInterceptor) BindRTCPReader(reader interceptor.RTCPReader) interceptor.RTCPReader {
+	return interceptor.RTCPReaderFunc(func(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) {
+		i, attr, err := reader.Read(b, a)
+		if err != nil {
+			return 0, nil, err
+		}
+
+		pkts, err := rtcp.Unmarshal(b[:i])
+		if err != nil {
+			return 0, nil, err
+		}
+		for _, rtcpPacket := range pkts {
+			nack, ok := rtcpPacket.(*rtcp.TransportLayerNack)
+			if !ok {
+				continue
+			}
+
+			go n.resendPackets(nack)
+		}
+
+		return i, attr, err
+	})
+}
+
+// BindLocalStream lets you modify any outgoing RTP packets. It is called once for per LocalStream. The returned method
+// will be called once per rtp packet.
+func (n *ResponderInterceptor) BindLocalStream(info *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter {
+	if !streamSupportNack(info) {
+		return writer
+	}
+
+	// error is already checked in NewGeneratorInterceptor
+	sendBuffer, _ := newSendBuffer(n.size)
+	n.streams.Store(info.SSRC, &localStream{sendBuffer: sendBuffer, rtpWriter: writer})
+
+	return interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
+		sendBuffer.add(&rtp.Packet{Header: *header, Payload: payload})
+
+		return writer.Write(header, payload, attributes)
+	})
+}
+
+// UnbindLocalStream is called when the Stream is removed. It can be used to clean up any data related to that track.
+func (n *ResponderInterceptor) UnbindLocalStream(info *interceptor.StreamInfo) {
+	n.streams.Delete(info.SSRC)
+}
+
+func (n *ResponderInterceptor) resendPackets(nack *rtcp.TransportLayerNack) {
+	v, ok := n.streams.Load(nack.MediaSSRC)
+	if !ok {
+		return
+	}
+
+	stream := v.(*localStream)
+
+	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)
+				}
+			}
+
+			return true
+		})
+	}
+}
diff --git a/pkg/nack/responder_option.go b/pkg/nack/responder_option.go
new file mode 100644
index 0000000..7ad52c8
--- /dev/null
+++ b/pkg/nack/responder_option.go
@@ -0,0 +1,23 @@
+package nack
+
+import "github.com/pion/logging"
+
+// ResponderOption can be used to configure ResponderInterceptor
+type ResponderOption func(s *ResponderInterceptor) error
+
+// ResponderSize sets the size of the interceptor.
+// Size must be one of: 1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192, 16384, 32768
+func ResponderSize(size uint16) ResponderOption {
+	return func(r *ResponderInterceptor) error {
+		r.size = size
+		return nil
+	}
+}
+
+// ResponderLog sets a logger for the interceptor
+func ResponderLog(log logging.LeveledLogger) ResponderOption {
+	return func(r *ResponderInterceptor) error {
+		r.log = log
+		return nil
+	}
+}
diff --git a/pkg/nack/send_buffer.go b/pkg/nack/send_buffer.go
new file mode 100644
index 0000000..cf3f020
--- /dev/null
+++ b/pkg/nack/send_buffer.go
@@ -0,0 +1,74 @@
+package nack
+
+import (
+	"fmt"
+
+	"github.com/pion/rtp"
+)
+
+const (
+	uint16SizeHalf = 1 << 15
+)
+
+type sendBuffer struct {
+	packets   []*rtp.Packet
+	size      uint16
+	lastAdded uint16
+	started   bool
+}
+
+func newSendBuffer(size uint16) (*sendBuffer, error) {
+	allowedSizes := make([]uint16, 0)
+	correctSize := false
+	for i := 0; i < 16; i++ {
+		if size == 1<<i {
+			correctSize = true
+			break
+		}
+		allowedSizes = append(allowedSizes, 1<<i)
+	}
+
+	if !correctSize {
+		return nil, fmt.Errorf("%w: %d is not a valid size, allowed sizes: %v", ErrInvalidSize, size, allowedSizes)
+	}
+
+	return &sendBuffer{
+		packets: make([]*rtp.Packet, size),
+		size:    size,
+	}, nil
+}
+
+func (s *sendBuffer) add(packet *rtp.Packet) {
+	seq := packet.SequenceNumber
+	if !s.started {
+		s.packets[seq%s.size] = packet
+		s.lastAdded = seq
+		s.started = true
+		return
+	}
+
+	diff := seq - s.lastAdded
+	if diff == 0 {
+		return
+	} else if diff < uint16SizeHalf {
+		for i := s.lastAdded + 1; i != seq; i++ {
+			s.packets[i%s.size] = nil
+		}
+	}
+
+	s.packets[seq%s.size] = packet
+	s.lastAdded = seq
+}
+
+func (s *sendBuffer) get(seq uint16) *rtp.Packet {
+	diff := s.lastAdded - seq
+	if diff >= uint16SizeHalf {
+		return nil
+	}
+
+	if diff >= s.size {
+		return nil
+	}
+
+	return s.packets[seq%s.size]
+}


Test patch:
diff --git a/pkg/nack/generator_interceptor_test.go b/pkg/nack/generator_interceptor_test.go
new file mode 100644
index 0000000..481a42c
--- /dev/null
+++ b/pkg/nack/generator_interceptor_test.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)
+}
diff --git a/pkg/nack/receive_log_test.go b/pkg/nack/receive_log_test.go
new file mode 100644
index 0000000..75e5e27
--- /dev/null
+++ b/pkg/nack/receive_log_test.go
@@ -0,0 +1,137 @@
+package nack
+
+import (
+	"fmt"
+	"reflect"
+	"testing"
+)
+
+func TestReceivedBuffer(t *testing.T) {
+	for _, start := range []uint16{0, 1, 127, 128, 129, 511, 512, 513, 32767, 32768, 32769, 65407, 65408, 65409, 65534, 65535} {
+		start := start
+
+		t.Run(fmt.Sprintf("StartFrom%d", start), func(t *testing.T) {
+			rl, err := newReceiveLog(128)
+			if err != nil {
+				t.Fatalf("%+v", err)
+			}
+
+			all := func(min uint16, max uint16) []uint16 {
+				result := make([]uint16, 0)
+				for i := min; i != max+1; i++ {
+					result = append(result, i)
+				}
+				return result
+			}
+			join := func(parts ...[]uint16) []uint16 {
+				result := make([]uint16, 0)
+				for _, p := range parts {
+					result = append(result, p...)
+				}
+				return result
+			}
+
+			add := func(nums ...uint16) {
+				for _, n := range nums {
+					seq := start + n
+					rl.add(seq)
+				}
+			}
+
+			assertGet := func(nums ...uint16) {
+				t.Helper()
+				for _, n := range nums {
+					seq := start + n
+					if !rl.get(seq) {
+						t.Errorf("not found: %d", seq)
+					}
+				}
+			}
+			assertNOTGet := func(nums ...uint16) {
+				t.Helper()
+				for _, n := range nums {
+					seq := start + n
+					if rl.get(seq) {
+						t.Errorf("packet found for %d", seq)
+					}
+				}
+			}
+			assertMissing := func(skipLastN uint16, nums []uint16) {
+				t.Helper()
+				missing := rl.missingSeqNumbers(skipLastN)
+				if missing == nil {
+					missing = []uint16{}
+				}
+				want := make([]uint16, 0, len(nums))
+				for _, n := range nums {
+					want = append(want, start+n)
+				}
+				if !reflect.DeepEqual(want, missing) {
+					t.Errorf("missing want/got %v / %v", want, missing)
+				}
+			}
+			assertLastConsecutive := func(lastConsecutive uint16) {
+				want := lastConsecutive + start
+				if rl.lastConsecutive != want {
+					t.Errorf("invalid lastConsecutive want %d got %d", want, rl.lastConsecutive)
+				}
+			}
+
+			add(0)
+			assertGet(0)
+			assertMissing(0, []uint16{})
+			assertLastConsecutive(0) // first element added
+
+			add(all(1, 127)...)
+			assertGet(all(1, 127)...)
+			assertMissing(0, []uint16{})
+			assertLastConsecutive(127)
+
+			add(128)
+			assertGet(128)
+			assertNOTGet(0)
+			assertMissing(0, []uint16{})
+			assertLastConsecutive(128)
+
+			add(130)
+			assertGet(130)
+			assertNOTGet(1, 2, 129)
+			assertMissing(0, []uint16{129})
+			assertLastConsecutive(128)
+
+			add(333)
+			assertGet(333)
+			assertNOTGet(all(0, 332)...)
+			assertMissing(0, all(206, 332))  // all 127 elements missing before 333
+			assertMissing(10, all(206, 323)) // skip last 10 packets (324-333) from check
+			assertLastConsecutive(205)       // lastConsecutive is still out of the buffer
+
+			add(329)
+			assertGet(329)
+			assertMissing(0, join(all(206, 328), all(330, 332)))
+			assertMissing(5, join(all(206, 328))) // skip last 5 packets (329-333) from check
+			assertLastConsecutive(205)
+
+			add(all(207, 320)...)
+			assertGet(all(207, 320)...)
+			assertMissing(0, join([]uint16{206}, all(321, 328), all(330, 332)))
+			assertLastConsecutive(205)
+
+			add(334)
+			assertGet(334)
+			assertNOTGet(206)
+			assertMissing(0, join(all(321, 328), all(330, 332)))
+			assertLastConsecutive(320) // head of buffer is full of consecutive packages
+
+			add(all(322, 328)...)
+			assertGet(all(322, 328)...)
+			assertMissing(0, join([]uint16{321}, all(330, 332)))
+			assertLastConsecutive(320)
+
+			add(321)
+			assertGet(321)
+			assertMissing(0, all(330, 332))
+			assertLastConsecutive(329) // after adding a single missing packet, lastConsecutive should jump forward
+		})
+	}
+}
diff --git a/pkg/nack/responder_interceptor_test.go b/pkg/nack/responder_interceptor_test.go
new file mode 100644
index 0000000..5ff322d
--- /dev/null
+++ b/pkg/nack/responder_interceptor_test.go
@@ -0,0 +1,71 @@
+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 TestResponderInterceptor(t *testing.T) {
+	i, err := NewResponderInterceptor(
+		ResponderSize(8),
+		ResponderLog(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, 15} {
+		assert.NoError(t, stream.WriteRTP(&rtp.Packet{Header: rtp.Header{SequenceNumber: seqNum}}))
+
+		select {
+		case p := <-stream.WrittenRTP():
+			assert.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():
+			assert.Equal(t, seqNum, p.SequenceNumber)
+		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):
+	}
+}
+
+func TestResponderInterceptor_InvalidSize(t *testing.T) {
+	_, err := NewResponderInterceptor(ResponderSize(5))
+	assert.Error(t, err, ErrInvalidSize)
+}
diff --git a/pkg/nack/send_buffer_test.go b/pkg/nack/send_buffer_test.go
new file mode 100644
index 0000000..afba216
--- /dev/null
+++ b/pkg/nack/send_buffer_test.go
@@ -0,0 +1,64 @@
+package nack
+
+import (
+	"testing"
+
+	"github.com/pion/rtp"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSendBuffer(t *testing.T) {
+	for _, start := range []uint16{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 511, 512, 513, 32767, 32768, 32769, 65527, 65528, 65529, 65530, 65531, 65532, 65533, 65534, 65535} {
+		start := start
+
+		sb, err := newSendBuffer(8)
+		assert.NoError(t, err)
+
+		add := func(nums ...uint16) {
+			for _, n := range nums {
+				seq := start + n
+				sb.add(&rtp.Packet{Header: rtp.Header{SequenceNumber: seq}})
+			}
+		}
+
+		assertGet := func(nums ...uint16) {
+			t.Helper()
+			for _, n := range nums {
+				seq := start + n
+				packet := sb.get(seq)
+				if packet == nil {
+					t.Errorf("packet not found: %d", seq)
+					continue
+				}
+				if packet.SequenceNumber != seq {
+					t.Errorf("packet for %d returned with incorrect SequenceNumber: %d", seq, packet.SequenceNumber)
+				}
+			}
+		}
+		assertNOTGet := func(nums ...uint16) {
+			t.Helper()
+			for _, n := range nums {
+				seq := start + n
+				packet := sb.get(seq)
+				if packet != nil {
+					t.Errorf("packet found for %d: %d", seq, packet.SequenceNumber)
+				}
+			}
+		}
+
+		add(0, 1, 2, 3, 4, 5, 6, 7)
+		assertGet(0, 1, 2, 3, 4, 5, 6, 7)
+
+		add(8)
+		assertGet(8)
+		assertNOTGet(0)
+
+		add(10)
+		assertGet(10)
+		assertNOTGet(1, 2, 9)
+
+		add(22)
+		assertGet(22)
+		assertNOTGet(3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21)
+	}
+}