Repository: pion/interceptor
Pull request #112 (2022-03-16 16:11:22)

RFC 8888 interceptor

codecov[bot]: # [Codecov](https://codecov.io/gh/pion/interceptor/pull/112?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) Report
> Merging [#112](https://codecov.io/gh/pion/interceptor/pull/112?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) (fbd9a06) into [master](https://codecov.io/gh/pion/interceptor/commit/ccd4d72f6947a6d5e041a53b68001f3117df2520?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) (ccd4d72) will **decrease** coverage by `3.07%`.
> The diff coverage is `39.34%`.

```diff
@@            Coverage Diff             @@
##           master     #112      +/-   ##
==========================================
- Coverage   79.30%   76.23%   -3.08%     
==========================================
  Files          51       55       +4     
  Lines        2459     2642     +183     
==========================================
+ Hits         1950     2014      +64     
- Misses        419      534     +115     
- Partials       90       94       +4     
```

| Flag | Coverage Δ | |
|---|---|---|
| go | `76.23% <39.34%> (-3.08%)` | :arrow_down: |
| wasm | `74.98% <39.34%> (-2.66%)` | :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.

| [Impacted Files](https://codecov.io/gh/pion/interceptor/pull/112?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | Coverage Δ | |
|---|---|---|
| [pkg/rfc8888/interceptor.go](https://codecov.io/gh/pion/interceptor/pull/112/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3JmYzg4ODgvaW50ZXJjZXB0b3IuZ28=) | `0.00% <0.00%> (ø)` | |
| [pkg/rfc8888/recorder.go](https://codecov.io/gh/pion/interceptor/pull/112/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3JmYzg4ODgvcmVjb3JkZXIuZ28=) | `0.00% <0.00%> (ø)` | |
| [pkg/rfc8888/unwrapper.go](https://codecov.io/gh/pion/interceptor/pull/112/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3JmYzg4ODgvdW53cmFwcGVyLmdv) | `72.72% <72.72%> (ø)` | |
| [pkg/rfc8888/stream\_log.go](https://codecov.io/gh/pion/interceptor/pull/112/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3JmYzg4ODgvc3RyZWFtX2xvZy5nbw==) | `100.00% <100.00%> (ø)` | |
| [pkg/gcc/arrival\_group\_accumulator.go](https://codecov.io/gh/pion/interceptor/pull/112/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9hcnJpdmFsX2dyb3VwX2FjY3VtdWxhdG9yLmdv) | `90.62% <0.00%> (-9.38%)` | :arrow_down: |
| [pkg/gcc/kalman.go](https://codecov.io/gh/pion/interceptor/pull/112/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9rYWxtYW4uZ28=) | `94.33% <0.00%> (-5.67%)` | :arrow_down: |
| [pkg/gcc/delay\_based\_bwe.go](https://codecov.io/gh/pion/interceptor/pull/112/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9kZWxheV9iYXNlZF9id2UuZ28=) | `83.33% <0.00%> (-3.04%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pion/interceptor/pull/112?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/pion/interceptor/pull/112?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion). Last update [ccd4d72...fbd9a06](https://codecov.io/gh/pion/interceptor/pull/112?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).


Patch:
diff --git a/AUTHORS.txt b/AUTHORS.txt
index ae86894..83d9064 100644
--- a/AUTHORS.txt
+++ b/AUTHORS.txt
@@ -10,6 +10,7 @@ Aditya Kumar <k.aditya00@gmail.com>
 aler9 <46489434+aler9@users.noreply.github.com>
 Antoine Baché <antoine@tenten.app>
 Atsushi Watanabe <atsushi.w@ieee.org>
+Bobby Peck <rpeck@mux.com>
 boks1971 <raja.gobi@tutanota.com>
 David Zhao <david@davidzhao.com>
 Jonathan Müller <jonathan@fotokite.com>
diff --git a/go.mod b/go.mod
index 74927d2..443efa3 100644
--- a/go.mod
+++ b/go.mod
@@ -4,7 +4,7 @@ go 1.15
 
 require (
 	github.com/pion/logging v0.2.2
-	github.com/pion/rtcp v1.2.9
+	github.com/pion/rtcp v1.2.10
 	github.com/pion/rtp v1.7.13
 	github.com/stretchr/testify v1.7.1
 )
diff --git a/go.sum b/go.sum
index 4bce19c..e947dec 100644
--- a/go.sum
+++ b/go.sum
@@ -4,14 +4,13 @@ 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.9 h1:1ujStwg++IOLIEoOiIQ2s+qBuJ1VN81KW+9pMPsif+U=
-github.com/pion/rtcp v1.2.9/go.mod h1:qVPhiCzAm4D/rxb6XzKeyZiQK69yJpbUDJSF7TgrqNo=
+github.com/pion/rtcp v1.2.10 h1:nkr3uj+8Sp97zyItdN60tE/S6vk4al5CPRR6Gejsdjc=
+github.com/pion/rtcp v1.2.10/go.mod h1:ztfEwXZNLGyF1oQDttz/ZKIBaeeg/oWbRYqzBM9TL1I=
 github.com/pion/rtp v1.7.13 h1:qcHwlmtiI50t1XivvoawdCGTP4Uiypzfrsap+bijcoA=
 github.com/pion/rtp v1.7.13/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMTY=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
diff --git a/pkg/rfc8888/interceptor.go b/pkg/rfc8888/interceptor.go
new file mode 100644
index 0000000..1529d5c
--- /dev/null
+++ b/pkg/rfc8888/interceptor.go
@@ -0,0 +1,179 @@
+// Package rfc8888 provides an interceptor that generates congestion control
+// feedback reports as defined by RFC 8888.
+package rfc8888
+
+import (
+	"sync"
+	"time"
+
+	"github.com/pion/interceptor"
+	"github.com/pion/logging"
+	"github.com/pion/rtcp"
+)
+
+// TickerFactory is a factory to create new tickers
+type TickerFactory func(d time.Duration) ticker
+
+// SenderInterceptorFactory is a interceptor.Factory for a SenderInterceptor
+type SenderInterceptorFactory struct {
+	opts []Option
+}
+
+// NewInterceptor constructs a new SenderInterceptor
+func (s *SenderInterceptorFactory) NewInterceptor(id string) (interceptor.Interceptor, error) {
+	i := &SenderInterceptor{
+		NoOp:          interceptor.NoOp{},
+		log:           logging.NewDefaultLoggerFactory().NewLogger("rfc8888_interceptor"),
+		lock:          sync.Mutex{},
+		wg:            sync.WaitGroup{},
+		recorder:      NewRecorder(),
+		interval:      100 * time.Millisecond,
+		maxReportSize: 1200,
+		packetChan:    make(chan packet),
+		newTicker: func(d time.Duration) ticker {
+			return &timeTicker{time.NewTicker(d)}
+		},
+		now:   time.Now,
+		close: make(chan struct{}),
+	}
+	for _, opt := range s.opts {
+		err := opt(i)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return i, nil
+}
+
+// NewSenderInterceptor returns a new SenderInterceptorFactory configured with the given options.
+func NewSenderInterceptor(opts ...Option) (*SenderInterceptorFactory, error) {
+	return &SenderInterceptorFactory{opts: opts}, nil
+}
+
+// SenderInterceptor sends congestion control feedback as specified in RFC 8888.
+type SenderInterceptor struct {
+	interceptor.NoOp
+	log           logging.LeveledLogger
+	lock          sync.Mutex
+	wg            sync.WaitGroup
+	recorder      *Recorder
+	interval      time.Duration
+	maxReportSize int64
+	packetChan    chan packet
+	newTicker     TickerFactory
+	now           func() time.Time
+	close         chan struct{}
+}
+
+type packet struct {
+	arrival        time.Time
+	ssrc           uint32
+	sequenceNumber uint16
+	ecn            uint8
+}
+
+// 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 (s *SenderInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	if s.isClosed() {
+		return writer
+	}
+
+	s.wg.Add(1)
+	go s.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 (s *SenderInterceptor) BindRemoteStream(info *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader {
+	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 attr == nil {
+			attr = make(interceptor.Attributes)
+		}
+		header, err := attr.GetRTPHeader(b[:i])
+		if err != nil {
+			return 0, nil, err
+		}
+
+		p := packet{
+			arrival:        s.now(),
+			ssrc:           header.SSRC,
+			sequenceNumber: header.SequenceNumber,
+			ecn:            0, // ECN is not supported (yet).
+		}
+		s.packetChan <- p
+		return i, attr, nil
+	})
+}
+
+// Close closes the interceptor.
+func (s *SenderInterceptor) Close() error {
+	s.log.Trace("close")
+	defer s.wg.Wait()
+
+	if !s.isClosed() {
+		close(s.close)
+	}
+
+	return nil
+}
+
+func (s *SenderInterceptor) isClosed() bool {
+	select {
+	case <-s.close:
+		return true
+	default:
+		return false
+	}
+}
+
+func (s *SenderInterceptor) loop(writer interceptor.RTCPWriter) {
+	defer s.wg.Done()
+
+	select {
+	case <-s.close:
+		return
+	case pkt := <-s.packetChan:
+		s.log.Tracef("got first packet: %v", pkt)
+		s.recorder.AddPacket(pkt.arrival, pkt.ssrc, pkt.sequenceNumber, pkt.ecn)
+	}
+
+	s.log.Trace("start loop")
+	t := s.newTicker(s.interval)
+	for {
+		select {
+		case <-s.close:
+			t.Stop()
+			return
+
+		case pkt := <-s.packetChan:
+			s.log.Tracef("got packet: %v", pkt)
+			s.recorder.AddPacket(pkt.arrival, pkt.ssrc, pkt.sequenceNumber, pkt.ecn)
+
+		case now := <-t.Ch():
+			s.log.Tracef("report triggered at %v", now)
+			if writer == nil {
+				s.log.Trace("no writer added, continue")
+				continue
+			}
+			pkts := s.recorder.BuildReport(now, int(s.maxReportSize))
+			if pkts == nil {
+				continue
+			}
+			s.log.Tracef("got report: %v", pkts)
+			if _, err := writer.Write([]rtcp.Packet{pkts}, nil); err != nil {
+				s.log.Error(err.Error())
+			}
+		}
+	}
+}
diff --git a/pkg/rfc8888/option.go b/pkg/rfc8888/option.go
new file mode 100644
index 0000000..865d1a9
--- /dev/null
+++ b/pkg/rfc8888/option.go
@@ -0,0 +1,30 @@
+package rfc8888
+
+import "time"
+
+// An Option is a function that can be used to configure a SenderInterceptor
+type Option func(*SenderInterceptor) error
+
+// SenderTicker sets an alternative for time.Ticker.
+func SenderTicker(f TickerFactory) Option {
+	return func(i *SenderInterceptor) error {
+		i.newTicker = f
+		return nil
+	}
+}
+
+// SenderNow sets an alternative for the time.Now function.
+func SenderNow(f func() time.Time) Option {
+	return func(i *SenderInterceptor) error {
+		i.now = f
+		return nil
+	}
+}
+
+// SendInterval sets the feedback send interval for the interceptor
+func SendInterval(interval time.Duration) Option {
+	return func(s *SenderInterceptor) error {
+		s.interval = interval
+		return nil
+	}
+}
diff --git a/pkg/rfc8888/recorder.go b/pkg/rfc8888/recorder.go
new file mode 100644
index 0000000..6cf0f5d
--- /dev/null
+++ b/pkg/rfc8888/recorder.go
@@ -0,0 +1,75 @@
+package rfc8888
+
+import (
+	"time"
+
+	"github.com/pion/rtcp"
+)
+
+type packetReport struct {
+	arrivalTime time.Time
+	ecn         uint8
+}
+
+// Recorder records incoming RTP packets and their arrival times. Recorder can
+// be used to create feedback reports as defined by RFC 8888.
+type Recorder struct {
+	ssrc    uint32
+	streams map[uint32]*streamLog
+}
+
+// NewRecorder creates a new Recorder
+func NewRecorder() *Recorder {
+	return &Recorder{
+		streams: map[uint32]*streamLog{},
+	}
+}
+
+// AddPacket writes a packet to the underlying stream.
+func (r *Recorder) AddPacket(ts time.Time, ssrc uint32, seq uint16, ecn uint8) {
+	stream, ok := r.streams[ssrc]
+	if !ok {
+		stream = newStreamLog(ssrc)
+		r.streams[ssrc] = stream
+	}
+	stream.add(ts, seq, ecn)
+}
+
+// BuildReport creates a new rtcp.CCFeedbackReport containing all packets that
+// were added by AddPacket and missing packets.
+func (r *Recorder) BuildReport(now time.Time, maxSize int) *rtcp.CCFeedbackReport {
+	report := &rtcp.CCFeedbackReport{
+		SenderSSRC:      r.ssrc,
+		ReportBlocks:    []rtcp.CCFeedbackReportBlock{},
+		ReportTimestamp: ntpTime32(now),
+	}
+
+	maxReportBlocks := (maxSize - 12 - (8 * len(r.streams))) / 2
+	var maxReportBlocksPerStream int
+	if len(r.streams) > 1 {
+		maxReportBlocksPerStream = maxReportBlocks / (len(r.streams) - 1)
+	} else {
+		maxReportBlocksPerStream = maxReportBlocks
+	}
+
+	for i, log := range r.streams {
+		if len(r.streams) > 1 && int(i) == len(r.streams)-1 {
+			maxReportBlocksPerStream = maxReportBlocks % len(r.streams)
+		}
+		block := log.metricsAfter(now, int64(maxReportBlocksPerStream))
+		report.ReportBlocks = append(report.ReportBlocks, block)
+	}
+
+	return report
+}
+
+func ntpTime32(t time.Time) uint32 {
+	// seconds since 1st January 1900
+	s := (float64(t.UnixNano()) / 1000000000.0) + 2208988800
+
+	integerPart := uint32(s)
+	fractionalPart := uint32((s - float64(integerPart)) * 0xFFFFFFFF)
+
+	// higher 32 bits are the integer part, lower 32 bits are the fractional part
+	return uint32(((uint64(integerPart)<<32 | uint64(fractionalPart)) >> 16) & 0xFFFFFFFF)
+}
diff --git a/pkg/rfc8888/stream_log.go b/pkg/rfc8888/stream_log.go
new file mode 100644
index 0000000..56f0b08
--- /dev/null
+++ b/pkg/rfc8888/stream_log.go
@@ -0,0 +1,107 @@
+package rfc8888
+
+import (
+	"time"
+
+	"github.com/pion/rtcp"
+)
+
+const maxReportsPerReportBlock = 16384
+
+type streamLog struct {
+	ssrc                       uint32
+	sequence                   unwrapper
+	init                       bool
+	nextSequenceNumberToReport int64 // next to report
+	lastSequenceNumberReceived int64 // highest received
+	log                        map[int64]*packetReport
+}
+
+func newStreamLog(ssrc uint32) *streamLog {
+	return &streamLog{
+		ssrc:                       ssrc,
+		sequence:                   unwrapper{},
+		init:                       false,
+		nextSequenceNumberToReport: 0,
+		lastSequenceNumberReceived: 0,
+		log:                        map[int64]*packetReport{},
+	}
+}
+
+func (l *streamLog) add(ts time.Time, sequenceNumber uint16, ecn uint8) {
+	unwrappedSequenceNumber := l.sequence.unwrap(sequenceNumber)
+	if !l.init {
+		l.init = true
+		l.nextSequenceNumberToReport = unwrappedSequenceNumber
+	}
+	l.log[unwrappedSequenceNumber] = &packetReport{
+		arrivalTime: ts,
+		ecn:         ecn,
+	}
+	if l.lastSequenceNumberReceived < unwrappedSequenceNumber {
+		l.lastSequenceNumberReceived = unwrappedSequenceNumber
+	}
+}
+
+// metricsAfter iterates over all packets order of their sequence number.
+// Packets are removed until the first loss is detected.
+func (l *streamLog) metricsAfter(reference time.Time, maxReportBlocks int64) rtcp.CCFeedbackReportBlock {
+	if len(l.log) == 0 {
+		return rtcp.CCFeedbackReportBlock{
+			MediaSSRC:     l.ssrc,
+			BeginSequence: uint16(l.nextSequenceNumberToReport),
+			MetricBlocks:  []rtcp.CCFeedbackMetricBlock{},
+		}
+	}
+	numReports := l.lastSequenceNumberReceived - l.nextSequenceNumberToReport + 1
+	if numReports > maxReportBlocks {
+		numReports = maxReportBlocks
+		l.nextSequenceNumberToReport = l.lastSequenceNumberReceived - maxReportBlocks + 1
+	}
+	metricBlocks := make([]rtcp.CCFeedbackMetricBlock, numReports)
+	offset := l.nextSequenceNumberToReport
+	lastReceived := l.nextSequenceNumberToReport
+	gapDetected := false
+	for i := offset; i <= l.lastSequenceNumberReceived; i++ {
+		received := false
+		ecn := uint8(0)
+		ato := uint16(0)
+		if report, ok := l.log[i]; ok {
+			received = true
+			ecn = report.ecn
+			ato = getArrivalTimeOffset(reference, report.arrivalTime)
+		}
+		metricBlocks[i-offset] = rtcp.CCFeedbackMetricBlock{
+			Received:          received,
+			ECN:               rtcp.ECN(ecn),
+			ArrivalTimeOffset: ato,
+		}
+
+		if !gapDetected {
+			if received && i == l.nextSequenceNumberToReport {
+				delete(l.log, i)
+				l.nextSequenceNumberToReport++
+				lastReceived = i
+			}
+			if i > lastReceived+1 {
+				gapDetected = true
+			}
+		}
+	}
+	return rtcp.CCFeedbackReportBlock{
+		MediaSSRC:     l.ssrc,
+		BeginSequence: uint16(offset),
+		MetricBlocks:  metricBlocks,
+	}
+}
+
+func getArrivalTimeOffset(base time.Time, arrival time.Time) uint16 {
+	if base.Before(arrival) {
+		return 0x1FFF
+	}
+	ato := uint16(base.Sub(arrival).Seconds() * 1024.0)
+	if ato > 0x1FFD {
+		return 0x1FFE
+	}
+	return ato
+}
diff --git a/pkg/rfc8888/ticker.go b/pkg/rfc8888/ticker.go
new file mode 100644
index 0000000..aabc480
--- /dev/null
+++ b/pkg/rfc8888/ticker.go
@@ -0,0 +1,16 @@
+package rfc8888
+
+import "time"
+
+type ticker interface {
+	Ch() <-chan time.Time
+	Stop()
+}
+
+type timeTicker struct {
+	*time.Ticker
+}
+
+func (t *timeTicker) Ch() <-chan time.Time {
+	return t.C
+}
diff --git a/pkg/rfc8888/unwrapper.go b/pkg/rfc8888/unwrapper.go
new file mode 100644
index 0000000..a0894f2
--- /dev/null
+++ b/pkg/rfc8888/unwrapper.go
@@ -0,0 +1,39 @@
+package rfc8888
+
+const (
+	maxSequenceNumberPlusOne = int64(65536)
+	breakpoint               = 32768 // half of max uint16
+)
+
+type unwrapper struct {
+	init          bool
+	lastUnwrapped int64
+}
+
+func isNewer(value, previous uint16) bool {
+	if value-previous == breakpoint {
+		return value > previous
+	}
+	return value != previous && (value-previous) < breakpoint
+}
+
+func (u *unwrapper) unwrap(i uint16) int64 {
+	if !u.init {
+		u.init = true
+		u.lastUnwrapped = int64(i)
+		return u.lastUnwrapped
+	}
+
+	lastWrapped := uint16(u.lastUnwrapped)
+	delta := int64(i - lastWrapped)
+	if isNewer(i, lastWrapped) {
+		if delta < 0 {
+			delta += maxSequenceNumberPlusOne
+		}
+	} else if delta > 0 && u.lastUnwrapped+delta-maxSequenceNumberPlusOne >= 0 {
+		delta -= maxSequenceNumberPlusOne
+	}
+
+	u.lastUnwrapped += delta
+	return u.lastUnwrapped
+}


Test patch:
diff --git a/internal/test/mock_ticker.go b/internal/test/mock_ticker.go
new file mode 100644
index 0000000..659832e
--- /dev/null
+++ b/internal/test/mock_ticker.go
@@ -0,0 +1,24 @@
+package test
+
+import (
+	"time"
+)
+
+// MockTicker is a helper to replace time.Ticker for testing purposes.
+type MockTicker struct {
+	C chan time.Time
+}
+
+// Stop stops the MockTicker.
+func (t *MockTicker) Stop() {
+}
+
+// Ch returns the tickers channel
+func (t *MockTicker) Ch() <-chan time.Time {
+	return t.C
+}
+
+// Tick sends now to the channel
+func (t *MockTicker) Tick(now time.Time) {
+	t.C <- now
+}
diff --git a/pkg/rfc8888/interceptor_test.go b/pkg/rfc8888/interceptor_test.go
new file mode 100644
index 0000000..47bda3d
--- /dev/null
+++ b/pkg/rfc8888/interceptor_test.go
@@ -0,0 +1,279 @@
+package rfc8888
+
+import (
+	"testing"
+	"time"
+
+	"github.com/pion/interceptor"
+	"github.com/pion/interceptor/internal/test"
+	"github.com/pion/rtcp"
+	"github.com/pion/rtp"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestInterceptor(t *testing.T) {
+	t.Run("before any packet", func(t *testing.T) {
+		f, err := NewSenderInterceptor()
+		assert.NoError(t, err)
+
+		i, err := f.NewInterceptor("")
+		assert.NoError(t, err)
+
+		stream := test.NewMockStream(&interceptor.StreamInfo{
+			SSRC: 123456,
+		}, i)
+		defer func() {
+			assert.NoError(t, stream.Close())
+		}()
+
+		var pkts []rtcp.Packet
+		select {
+		case pkts = <-stream.WrittenRTCP():
+		case <-time.After(300 * time.Millisecond):
+		}
+		assert.Equal(t, len(pkts), 0)
+	})
+
+	t.Run("after RTP packets", func(t *testing.T) {
+		f, err := NewSenderInterceptor()
+		assert.NoError(t, err)
+
+		i, err := f.NewInterceptor("")
+		assert.NoError(t, err)
+
+		stream := test.NewMockStream(&interceptor.StreamInfo{
+			SSRC: 123456,
+		}, i)
+		defer func() {
+			assert.NoError(t, stream.Close())
+		}()
+
+		for i := 0; i < 10; i++ {
+			stream.ReceiveRTP(&rtp.Packet{
+				Header: rtp.Header{
+					Version:          0,
+					Padding:          false,
+					Extension:        false,
+					Marker:           false,
+					PayloadType:      0,
+					SequenceNumber:   uint16(i),
+					Timestamp:        0,
+					SSRC:             123456,
+					CSRC:             []uint32{},
+					ExtensionProfile: 0,
+					Extensions:       []rtp.Extension{},
+				},
+				Payload:     []byte{},
+				PaddingSize: 0,
+			})
+		}
+
+		pkts := <-stream.WrittenRTCP()
+		assert.Equal(t, len(pkts), 1)
+		fb, ok := pkts[0].(*rtcp.CCFeedbackReport)
+		assert.True(t, ok)
+		assert.Equal(t, 1, len(fb.ReportBlocks))
+		assert.Equal(t, uint32(123456), fb.ReportBlocks[0].MediaSSRC)
+		assert.Equal(t, 10, len(fb.ReportBlocks[0].MetricBlocks))
+	})
+
+	t.Run("different delays between RTP packets", func(t *testing.T) {
+		mNow := &test.MockTime{}
+		mTick := &test.MockTicker{
+			C: make(chan time.Time),
+		}
+		f, err := NewSenderInterceptor(
+			SenderTicker(func(d time.Duration) ticker {
+				return mTick
+			}),
+			SenderNow(mNow.Now),
+		)
+		assert.NoError(t, err)
+
+		i, err := f.NewInterceptor("")
+		assert.NoError(t, err)
+
+		stream := test.NewMockStream(&interceptor.StreamInfo{
+			SSRC: 123456,
+		}, i)
+		defer func() {
+			assert.NoError(t, stream.Close())
+		}()
+
+		zero := time.Date(1900, time.January, 1, 0, 0, 0, 0, time.UTC)
+
+		delays := []time.Duration{
+			0,
+			250 * time.Millisecond,
+			500 * time.Millisecond,
+			time.Second,
+		}
+		for i, d := range delays {
+			mNow.SetNow(zero.Add(d))
+			stream.ReceiveRTP(&rtp.Packet{
+				Header: rtp.Header{
+					SequenceNumber: uint16(i),
+					SSRC:           123456,
+				},
+			})
+			select {
+			case r := <-stream.ReadRTP():
+				assert.NoError(t, r.Err)
+			case <-time.After(10 * time.Millisecond):
+				t.Fatal("receiver rtp packet not found")
+			}
+		}
+		mTick.Tick(zero.Add(time.Second))
+		pkts := <-stream.WrittenRTCP()
+		assert.Equal(t, 1, len(pkts))
+		ccfb, ok := pkts[0].(*rtcp.CCFeedbackReport)
+		assert.True(t, ok)
+		assert.Equal(t, uint32(1<<16), ccfb.ReportTimestamp)
+		assert.Equal(t, 1, len(ccfb.ReportBlocks))
+		assert.Equal(t, uint32(123456), ccfb.ReportBlocks[0].MediaSSRC)
+		assert.Equal(t, 4, len(ccfb.ReportBlocks[0].MetricBlocks))
+		assert.Equal(t, uint16(0), ccfb.ReportBlocks[0].BeginSequence)
+		assert.Equal(t, []rtcp.CCFeedbackMetricBlock{
+			{
+				Received:          true,
+				ECN:               0,
+				ArrivalTimeOffset: 1024,
+			},
+			{
+				Received:          true,
+				ECN:               0,
+				ArrivalTimeOffset: 512 + 256,
+			},
+			{
+				Received:          true,
+				ECN:               0,
+				ArrivalTimeOffset: 512,
+			},
+			{
+				Received:          true,
+				ECN:               0,
+				ArrivalTimeOffset: 0,
+			},
+		}, ccfb.ReportBlocks[0].MetricBlocks)
+	})
+
+	t.Run("packet loss", func(t *testing.T) {
+		mNow := &test.MockTime{}
+		mTick := &test.MockTicker{
+			C: make(chan time.Time),
+		}
+		f, err := NewSenderInterceptor(
+			SenderTicker(func(d time.Duration) ticker {
+				return mTick
+			}),
+			SenderNow(mNow.Now),
+		)
+		assert.NoError(t, err)
+
+		i, err := f.NewInterceptor("")
+		assert.NoError(t, err)
+
+		stream := test.NewMockStream(&interceptor.StreamInfo{
+			SSRC: 123456,
+		}, i)
+		defer func() {
+			assert.NoError(t, stream.Close())
+		}()
+
+		zero := time.Date(1900, time.January, 1, 0, 0, 0, 0, time.UTC)
+
+		sequenceNumberToDelay := map[int]int{
+			0:  0,
+			1:  125,
+			4:  250,
+			8:  500,
+			9:  750,
+			10: 1000,
+		}
+		for i := 0; i <= 10; i++ {
+			if _, ok := sequenceNumberToDelay[i]; !ok {
+				continue
+			}
+			mNow.SetNow(zero.Add(time.Duration(sequenceNumberToDelay[i]) * time.Millisecond))
+			stream.ReceiveRTP(&rtp.Packet{
+				Header: rtp.Header{
+					SequenceNumber: uint16(i),
+					SSRC:           123456,
+				},
+			})
+			select {
+			case r := <-stream.ReadRTP():
+				assert.NoError(t, r.Err)
+			case <-time.After(10 * time.Millisecond):
+				t.Fatal("receiver rtp packet not found")
+			}
+		}
+		mTick.Tick(zero.Add(time.Second))
+		pkts := <-stream.WrittenRTCP()
+		assert.Equal(t, 1, len(pkts))
+		ccfb, ok := pkts[0].(*rtcp.CCFeedbackReport)
+		assert.True(t, ok)
+		assert.Equal(t, uint32(1<<16), ccfb.ReportTimestamp)
+		assert.Equal(t, 1, len(ccfb.ReportBlocks))
+		assert.Equal(t, uint32(123456), ccfb.ReportBlocks[0].MediaSSRC)
+		assert.Equal(t, 11, len(ccfb.ReportBlocks[0].MetricBlocks))
+		assert.Equal(t, uint16(0), ccfb.ReportBlocks[0].BeginSequence)
+		assert.Equal(t, []rtcp.CCFeedbackMetricBlock{
+			{
+				Received:          true,
+				ECN:               0,
+				ArrivalTimeOffset: 1024,
+			},
+			{
+				Received:          true,
+				ECN:               0,
+				ArrivalTimeOffset: 1024 - 128,
+			},
+			{
+				Received:          false,
+				ECN:               0,
+				ArrivalTimeOffset: 0,
+			},
+			{
+				Received:          false,
+				ECN:               0,
+				ArrivalTimeOffset: 0,
+			},
+			{
+				Received:          true,
+				ECN:               0,
+				ArrivalTimeOffset: 1024 - 256,
+			},
+			{
+				Received:          false,
+				ECN:               0,
+				ArrivalTimeOffset: 0,
+			},
+			{
+				Received:          false,
+				ECN:               0,
+				ArrivalTimeOffset: 0,
+			},
+			{
+				Received:          false,
+				ECN:               0,
+				ArrivalTimeOffset: 0,
+			},
+			{
+				Received:          true,
+				ECN:               0,
+				ArrivalTimeOffset: 512,
+			},
+			{
+				Received:          true,
+				ECN:               0,
+				ArrivalTimeOffset: 256,
+			},
+			{
+				Received:          true,
+				ECN:               0,
+				ArrivalTimeOffset: 0,
+			},
+		}, ccfb.ReportBlocks[0].MetricBlocks)
+	})
+}
diff --git a/pkg/rfc8888/recorder_test.go b/pkg/rfc8888/recorder_test.go
new file mode 100644
index 0000000..6d1e5cb
--- /dev/null
+++ b/pkg/rfc8888/recorder_test.go
@@ -0,0 +1,195 @@
+package rfc8888
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/pion/rtcp"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetArrivalTimeOffset(t *testing.T) {
+	for _, test := range []struct {
+		base    time.Time
+		arrival time.Time
+		want    uint16
+	}{
+		{
+			base:    time.Time{}.Add(time.Second),
+			arrival: time.Time{},
+			want:    1024,
+		},
+		{
+			base:    time.Time{}.Add(500 * time.Millisecond),
+			arrival: time.Time{},
+			want:    512,
+		},
+		{
+			base:    time.Time{}.Add(8 * time.Second),
+			arrival: time.Time{},
+			want:    0x1FFE,
+		},
+		{
+			base:    time.Time{},
+			arrival: time.Time{}.Add(time.Second),
+			want:    0x1FFF,
+		},
+	} {
+		assert.Equal(t, test.want, getArrivalTimeOffset(test.base, test.arrival))
+	}
+}
+
+func TestRecorder(t *testing.T) {
+	t.Run("normal", func(t *testing.T) {
+		recorder := NewRecorder()
+		now := time.Time{}
+		recorder.AddPacket(now, 123456, 0, 0)
+		recorder.AddPacket(now.Add(125*time.Millisecond), 123456, 1, 0)
+		recorder.AddPacket(now.Add(250*time.Millisecond), 123456, 2, 0)
+		recorder.AddPacket(now.Add(500*time.Millisecond), 123456, 3, 0)
+		recorder.AddPacket(now.Add(625*time.Millisecond), 123456, 4, 0)
+		recorder.AddPacket(now.Add(750*time.Millisecond), 123456, 5, 0)
+
+		report := recorder.BuildReport(now.Add(time.Second), 1500)
+		assert.Equal(t, 1, len(report.ReportBlocks))
+		assert.Equal(t, rtcp.CCFeedbackReportBlock{
+			MediaSSRC:     123456,
+			BeginSequence: 0,
+			MetricBlocks: []rtcp.CCFeedbackMetricBlock{
+				{
+					Received:          true,
+					ECN:               0,
+					ArrivalTimeOffset: 1024,
+				},
+				{
+					Received:          true,
+					ECN:               0,
+					ArrivalTimeOffset: 1024 - 128,
+				},
+				{
+					Received:          true,
+					ECN:               0,
+					ArrivalTimeOffset: 1024 - 256,
+				},
+				{
+					Received:          true,
+					ECN:               0,
+					ArrivalTimeOffset: 1024 - 512,
+				},
+				{
+					Received:          true,
+					ECN:               0,
+					ArrivalTimeOffset: 1024 - 640,
+				},
+				{
+					Received:          true,
+					ECN:               0,
+					ArrivalTimeOffset: 1024 - 768,
+				},
+			},
+		}, report.ReportBlocks[0])
+	})
+
+	t.Run("packet loss", func(t *testing.T) {
+		recorder := NewRecorder()
+		now := time.Time{}
+		recorder.AddPacket(now, 123456, 0, 0)
+		recorder.AddPacket(now.Add(250*time.Millisecond), 123456, 2, 0)
+		recorder.AddPacket(now.Add(625*time.Millisecond), 123456, 4, 0)
+		recorder.AddPacket(now.Add(750*time.Millisecond), 123456, 5, 0)
+
+		report := recorder.BuildReport(now.Add(time.Second), 1500)
+		assert.Equal(t, 1, len(report.ReportBlocks))
+		assert.Equal(t, 6, len(report.ReportBlocks[0].MetricBlocks))
+		assert.Equal(t, rtcp.CCFeedbackReportBlock{
+			MediaSSRC:     123456,
+			BeginSequence: 0,
+			MetricBlocks: []rtcp.CCFeedbackMetricBlock{
+				{
+					Received:          true,
+					ECN:               0,
+					ArrivalTimeOffset: 1024,
+				},
+				{
+					Received:          false,
+					ECN:               0,
+					ArrivalTimeOffset: 0,
+				},
+				{
+					Received:          true,
+					ECN:               0,
+					ArrivalTimeOffset: 1024 - 256,
+				},
+				{
+					Received:          false,
+					ECN:               0,
+					ArrivalTimeOffset: 0,
+				},
+				{
+					Received:          true,
+					ECN:               0,
+					ArrivalTimeOffset: 1024 - 640,
+				},
+				{
+					Received:          true,
+					ECN:               0,
+					ArrivalTimeOffset: 1024 - 768,
+				},
+			},
+		}, report.ReportBlocks[0])
+	})
+
+	t.Run("MaxreportsPerStream", func(t *testing.T) {
+		recorder := NewRecorder()
+		now := time.Time{}
+
+		// Add 1000 packets on 10 different streams
+		for i := 0; i < 10; i++ {
+			for j := 0; j < 100; j++ {
+				recorder.AddPacket(now, uint32(i), uint16(j), 0)
+			}
+		}
+		reports := recorder.BuildReport(time.Time{}, 1380)
+
+		for i := 0; i < 10; i++ {
+			assert.Greater(t, 72, len(reports.ReportBlocks[i].MetricBlocks))
+			assert.Less(t, 3, len(reports.ReportBlocks[i].MetricBlocks))
+		}
+	})
+}
+
+func TestNTPTime32(t *testing.T) {
+	zero := time.Date(1900, time.January, 1, 0, 0, 0, 0, time.UTC)
+	notSoLongAgo := time.Date(2022, time.May, 5, 14, 48, 20, 0, time.UTC)
+	for i, cc := range []struct {
+		input    time.Time
+		expected uint32
+	}{
+		{
+			input:    zero,
+			expected: 0,
+		},
+		{
+			input:    zero.Add(time.Second),
+			expected: 1 << 16,
+		},
+		{
+			input:    notSoLongAgo,
+			expected: uint32(int(notSoLongAgo.Sub(zero).Seconds())&0xffff) << 16,
+		},
+		{
+			input:    zero.Add(400 * time.Millisecond),
+			expected: 26214,
+		},
+		{
+			input:    zero.Add(1400 * time.Millisecond),
+			expected: 1<<16 + 26214,
+		},
+	} {
+		t.Run(fmt.Sprintf("%v", i), func(t *testing.T) {
+			res := ntpTime32(cc.input)
+			assert.Equalf(t, cc.expected, res, "%b != %b", cc.expected, res)
+		})
+	}
+}
diff --git a/pkg/rfc8888/stream_log_test.go b/pkg/rfc8888/stream_log_test.go
new file mode 100644
index 0000000..fc35b2c
--- /dev/null
+++ b/pkg/rfc8888/stream_log_test.go
@@ -0,0 +1,580 @@
+package rfc8888
+
+import (
+	"testing"
+	"time"
+
+	"github.com/pion/rtcp"
+	"github.com/stretchr/testify/assert"
+)
+
+type input struct {
+	ts  time.Time
+	nr  uint16
+	ecn uint8
+}
+
+func TestStreamLogAdd(t *testing.T) {
+	tests := []struct {
+		name         string
+		inputs       []input
+		expectedNext int64
+		expectedLast int64
+		expectedLog  map[int64]*packetReport
+	}{
+		{
+			name:         "emptyLog",
+			inputs:       []input{},
+			expectedNext: 0,
+			expectedLast: 0,
+			expectedLog:  map[int64]*packetReport{},
+		},
+		//nolint
+		{
+			name: "addInOrderSequence",
+			inputs: []input{
+				{
+					ts:  time.Time{},
+					nr:  0,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(10 * time.Millisecond),
+					nr:  1,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(20 * time.Millisecond),
+					nr:  2,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(30 * time.Millisecond),
+					nr:  3,
+					ecn: 0,
+				},
+			},
+			expectedNext: 0,
+			expectedLast: 3,
+			expectedLog: map[int64]*packetReport{
+				0: {
+					arrivalTime: time.Time{},
+					ecn:         0,
+				},
+				1: {
+					arrivalTime: time.Time{}.Add(10 * time.Millisecond),
+					ecn:         0,
+				},
+				2: {
+					arrivalTime: time.Time{}.Add(20 * time.Millisecond),
+					ecn:         0,
+				},
+				3: {
+					arrivalTime: time.Time{}.Add(30 * time.Millisecond),
+					ecn:         0,
+				},
+			},
+		},
+		//nolint
+		{
+			name: "reorderedSequence",
+			inputs: []input{
+				{
+					ts:  time.Time{},
+					nr:  0,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(10 * time.Millisecond),
+					nr:  2,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(20 * time.Millisecond),
+					nr:  1,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(30 * time.Millisecond),
+					nr:  3,
+					ecn: 0,
+				},
+			},
+			expectedNext: 0,
+			expectedLast: 3,
+			expectedLog: map[int64]*packetReport{
+				0: {
+					arrivalTime: time.Time{},
+					ecn:         0,
+				},
+				1: {
+					arrivalTime: time.Time{}.Add(20 * time.Millisecond),
+					ecn:         0,
+				},
+				2: {
+					arrivalTime: time.Time{}.Add(10 * time.Millisecond),
+					ecn:         0,
+				},
+				3: {
+					arrivalTime: time.Time{}.Add(30 * time.Millisecond),
+					ecn:         0,
+				},
+			},
+		},
+		{
+			name: "reorderedWrappingSequence",
+			inputs: []input{
+				{
+					ts:  time.Time{},
+					nr:  65534,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(10 * time.Millisecond),
+					nr:  0,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(20 * time.Millisecond),
+					nr:  65535,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(30 * time.Millisecond),
+					nr:  2,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(40 * time.Millisecond),
+					nr:  1,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(50 * time.Millisecond),
+					nr:  3,
+					ecn: 0,
+				},
+			},
+			expectedNext: 65534,
+			expectedLast: 65539,
+			expectedLog: map[int64]*packetReport{
+				65534: {
+					arrivalTime: time.Time{},
+					ecn:         0,
+				},
+				65535: {
+					arrivalTime: time.Time{}.Add(20 * time.Millisecond),
+					ecn:         0,
+				},
+				65536: {
+					arrivalTime: time.Time{}.Add(10 * time.Millisecond),
+					ecn:         0,
+				},
+				65537: {
+					arrivalTime: time.Time{}.Add(40 * time.Millisecond),
+					ecn:         0,
+				},
+				65538: {
+					arrivalTime: time.Time{}.Add(30 * time.Millisecond),
+					ecn:         0,
+				},
+				65539: {
+					arrivalTime: time.Time{}.Add(50 * time.Millisecond),
+					ecn:         0,
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			sl := newStreamLog(0)
+			for _, input := range test.inputs {
+				sl.add(input.ts, input.nr, input.ecn)
+			}
+			assert.Equal(t, test.expectedNext, sl.nextSequenceNumberToReport)
+			assert.Equal(t, test.expectedLast, sl.lastSequenceNumberReceived)
+			assert.Equal(t, test.expectedLog, sl.log)
+		})
+	}
+}
+
+func TestStreamLogMetricsAfter(t *testing.T) {
+	tests := []struct {
+		name   string
+		inputs []input
+
+		expectedLast int64
+
+		expectedNextBefore int64
+		expectedLogBefore  map[int64]*packetReport
+
+		expectedNextAfter int64
+		expectedLogAfter  map[int64]*packetReport
+
+		expectedMetrics rtcp.CCFeedbackReportBlock
+	}{
+		{
+			name:               "emptyLog",
+			inputs:             []input{},
+			expectedNextBefore: 0,
+			expectedLast:       0,
+			expectedLogBefore:  map[int64]*packetReport{},
+			expectedNextAfter:  0,
+			expectedLogAfter:   map[int64]*packetReport{},
+			expectedMetrics: rtcp.CCFeedbackReportBlock{
+				MediaSSRC:     0,
+				BeginSequence: 0,
+				MetricBlocks:  []rtcp.CCFeedbackMetricBlock{},
+			},
+		},
+		//nolint
+		{
+			name: "addInOrderSequence",
+			inputs: []input{
+				{
+					ts:  time.Time{},
+					nr:  0,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(10 * time.Millisecond),
+					nr:  1,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(20 * time.Millisecond),
+					nr:  2,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(30 * time.Millisecond),
+					nr:  3,
+					ecn: 0,
+				},
+			},
+			expectedNextBefore: 0,
+			expectedLast:       3,
+			expectedLogBefore: map[int64]*packetReport{
+				0: {
+					arrivalTime: time.Time{},
+					ecn:         0,
+				},
+				1: {
+					arrivalTime: time.Time{}.Add(10 * time.Millisecond),
+					ecn:         0,
+				},
+				2: {
+					arrivalTime: time.Time{}.Add(20 * time.Millisecond),
+					ecn:         0,
+				},
+				3: {
+					arrivalTime: time.Time{}.Add(30 * time.Millisecond),
+					ecn:         0,
+				},
+			},
+			expectedNextAfter: 4,
+			expectedLogAfter:  map[int64]*packetReport{},
+			expectedMetrics: rtcp.CCFeedbackReportBlock{
+				MetricBlocks: []rtcp.CCFeedbackMetricBlock{
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 1024,
+					},
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 1013,
+					},
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 1003,
+					},
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 993,
+					},
+				},
+			},
+		},
+		//nolint
+		{
+			name: "reorderedSequence",
+			inputs: []input{
+				{
+					ts:  time.Time{},
+					nr:  0,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(10 * time.Millisecond),
+					nr:  2,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(20 * time.Millisecond),
+					nr:  1,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(30 * time.Millisecond),
+					nr:  3,
+					ecn: 0,
+				},
+			},
+			expectedNextBefore: 0,
+			expectedLast:       3,
+			expectedLogBefore: map[int64]*packetReport{
+				0: {
+					arrivalTime: time.Time{},
+					ecn:         0,
+				},
+				1: {
+					arrivalTime: time.Time{}.Add(20 * time.Millisecond),
+					ecn:         0,
+				},
+				2: {
+					arrivalTime: time.Time{}.Add(10 * time.Millisecond),
+					ecn:         0,
+				},
+				3: {
+					arrivalTime: time.Time{}.Add(30 * time.Millisecond),
+					ecn:         0,
+				},
+			},
+			expectedNextAfter: 4,
+			expectedLogAfter:  map[int64]*packetReport{},
+			expectedMetrics: rtcp.CCFeedbackReportBlock{
+				MetricBlocks: []rtcp.CCFeedbackMetricBlock{
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 1024,
+					},
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 1003,
+					},
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 1013,
+					},
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 993,
+					},
+				},
+			},
+		},
+		{
+			name: "reorderedWrappingSequence",
+			inputs: []input{
+				{
+					ts:  time.Time{},
+					nr:  65534,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(10 * time.Millisecond),
+					nr:  0,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(20 * time.Millisecond),
+					nr:  65535,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(30 * time.Millisecond),
+					nr:  2,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(40 * time.Millisecond),
+					nr:  1,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(50 * time.Millisecond),
+					nr:  3,
+					ecn: 0,
+				},
+			},
+			expectedNextBefore: 65534,
+			expectedLast:       65539,
+			expectedLogBefore: map[int64]*packetReport{
+				65534: {
+					arrivalTime: time.Time{},
+					ecn:         0,
+				},
+				65535: {
+					arrivalTime: time.Time{}.Add(20 * time.Millisecond),
+					ecn:         0,
+				},
+				65536: {
+					arrivalTime: time.Time{}.Add(10 * time.Millisecond),
+					ecn:         0,
+				},
+				65537: {
+					arrivalTime: time.Time{}.Add(40 * time.Millisecond),
+					ecn:         0,
+				},
+				65538: {
+					arrivalTime: time.Time{}.Add(30 * time.Millisecond),
+					ecn:         0,
+				},
+				65539: {
+					arrivalTime: time.Time{}.Add(50 * time.Millisecond),
+					ecn:         0,
+				},
+			},
+			expectedNextAfter: 65540,
+			expectedLogAfter:  map[int64]*packetReport{},
+			expectedMetrics: rtcp.CCFeedbackReportBlock{
+				BeginSequence: 65534,
+				MetricBlocks: []rtcp.CCFeedbackMetricBlock{
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 1024,
+					},
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 1003,
+					},
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 1013,
+					},
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 983,
+					},
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 993,
+					},
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 972,
+					},
+				},
+			},
+		},
+		{
+			name: "addMissingPacketSequence",
+			inputs: []input{
+				{
+					ts:  time.Time{},
+					nr:  0,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(20 * time.Millisecond),
+					nr:  2,
+					ecn: 0,
+				},
+				{
+					ts:  time.Time{}.Add(30 * time.Millisecond),
+					nr:  3,
+					ecn: 0,
+				},
+			},
+			expectedNextBefore: 0,
+			expectedLast:       3,
+			expectedLogBefore: map[int64]*packetReport{
+				0: {
+					arrivalTime: time.Time{},
+					ecn:         0,
+				},
+				2: {
+					arrivalTime: time.Time{}.Add(20 * time.Millisecond),
+					ecn:         0,
+				},
+				3: {
+					arrivalTime: time.Time{}.Add(30 * time.Millisecond),
+					ecn:         0,
+				},
+			},
+			expectedNextAfter: 1,
+			expectedLogAfter: map[int64]*packetReport{
+				2: {
+					arrivalTime: time.Time{}.Add(20 * time.Millisecond),
+					ecn:         0,
+				},
+				3: {
+					arrivalTime: time.Time{}.Add(30 * time.Millisecond),
+					ecn:         0,
+				},
+			},
+			expectedMetrics: rtcp.CCFeedbackReportBlock{
+				MetricBlocks: []rtcp.CCFeedbackMetricBlock{
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 1024,
+					},
+					{
+						Received:          false,
+						ECN:               0,
+						ArrivalTimeOffset: 0,
+					},
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 1003,
+					},
+					{
+						Received:          true,
+						ECN:               0,
+						ArrivalTimeOffset: 993,
+					},
+				},
+			},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			sl := newStreamLog(0)
+			for _, input := range test.inputs {
+				sl.add(input.ts, input.nr, input.ecn)
+			}
+
+			assert.Equal(t, test.expectedNextBefore, sl.nextSequenceNumberToReport)
+			assert.Equal(t, test.expectedLast, sl.lastSequenceNumberReceived)
+			assert.Equal(t, test.expectedLogBefore, sl.log)
+
+			metrics := sl.metricsAfter(time.Time{}.Add(time.Second), 500)
+
+			assert.Equal(t, test.expectedNextAfter, sl.nextSequenceNumberToReport)
+			assert.Equal(t, test.expectedLast, sl.lastSequenceNumberReceived)
+			assert.Equal(t, test.expectedLogAfter, sl.log)
+
+			assert.Equal(t, test.expectedMetrics, metrics)
+		})
+	}
+}
+
+func TestRemoveOldestPackets(t *testing.T) {
+	sl := newStreamLog(0)
+	sl.add(time.Time{}.Add(time.Second), 1, 0)
+	now := time.Now().Add(10 * time.Second)
+	for i := 2; i < 16386; i++ {
+		now = now.Add(10 * time.Millisecond)
+		sl.add(now, uint16(i), 0)
+	}
+	metrics := sl.metricsAfter(now, maxReportsPerReportBlock)
+	assert.Equal(t, uint16(2), metrics.BeginSequence)
+	assert.Lenf(t, metrics.MetricBlocks, 16384, "%v != %v", len(metrics.MetricBlocks), 16384)
+}
diff --git a/pkg/rfc8888/unwrapper_test.go b/pkg/rfc8888/unwrapper_test.go
new file mode 100644
index 0000000..9744a61
--- /dev/null
+++ b/pkg/rfc8888/unwrapper_test.go
@@ -0,0 +1,54 @@
+package rfc8888
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestUnwrapper(t *testing.T) {
+	cases := []struct {
+		input    []uint16
+		expected []int64
+	}{
+		{
+			input:    []uint16{},
+			expected: []int64{},
+		},
+		{
+			input:    []uint16{0, 1, 2, 3, 4},
+			expected: []int64{0, 1, 2, 3, 4},
+		},
+		{
+			input:    []uint16{65534, 65535, 0, 1, 2},
+			expected: []int64{65534, 65535, 65536, 65537, 65538},
+		},
+		{
+			input:    []uint16{32769, 0},
+			expected: []int64{32769, 65536},
+		},
+		{
+			input:    []uint16{32767, 0},
+			expected: []int64{32767, 0},
+		},
+		{
+			input:    []uint16{0, 1, 4, 3, 2, 5},
+			expected: []int64{0, 1, 4, 3, 2, 5},
+		},
+		{
+			input:    []uint16{65534, 0, 1, 65535, 4, 3, 2, 5},
+			expected: []int64{65534, 65536, 65537, 65535, 65540, 65539, 65538, 65541},
+		},
+	}
+	for i, tc := range cases {
+		t.Run(fmt.Sprintf("%v", i), func(t *testing.T) {
+			u := &unwrapper{}
+			result := []int64{}
+			for _, i := range tc.input {
+				result = append(result, u.unwrap(i))
+			}
+			assert.Equal(t, tc.expected, result)
+		})
+	}
+}