Repository: pion/interceptor
Pull request #143 (2022-07-11 19:59:22)

Feat/getstats interceptor
#### Description

Implement the RTP and RTCP-related statistics from the WebRTC statistics API (https://w3c.github.io/webrtc-stats/) draft, that are easy to gather by intercepting RTP/RTCP streams.

codecov[bot]: # [Codecov](https://codecov.io/gh/pion/interceptor/pull/143?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) Report
> Merging [#143](https://codecov.io/gh/pion/interceptor/pull/143?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) (96ad9b7) into [master](https://codecov.io/gh/pion/interceptor/commit/0748586beeecf8c4bd66824ffde1b097d6bee431?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) (0748586) will **decrease** coverage by `1.13%`.
> The diff coverage is `66.66%`.

```diff
@@            Coverage Diff             @@
##           master     #143      +/-   ##
==========================================
- Coverage   79.30%   78.16%   -1.14%     
==========================================
  Files          51       55       +4     
  Lines        2489     2753     +264     
==========================================
+ Hits         1974     2152     +178     
- Misses        417      501      +84     
- Partials       98      100       +2     
```

| Flag | Coverage Δ | |
|---|---|---|
| go | `78.16% <66.66%> (-1.14%)` | :arrow_down: |
| wasm | `76.67% <66.66%> (-1.19%)` | :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/143?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | Coverage Δ | |
|---|---|---|
| [pkg/report/sender\_interceptor.go](https://codecov.io/gh/pion/interceptor/pull/143/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3JlcG9ydC9zZW5kZXJfaW50ZXJjZXB0b3IuZ28=) | `81.15% <ø> (-1.96%)` | :arrow_down: |
| [pkg/stats/interceptor.go](https://codecov.io/gh/pion/interceptor/pull/143/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3N0YXRzL2ludGVyY2VwdG9yLmdv) | `0.00% <0.00%> (ø)` | |
| [internal/sequencenumber/unwrapper.go](https://codecov.io/gh/pion/interceptor/pull/143/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-aW50ZXJuYWwvc2VxdWVuY2VudW1iZXIvdW53cmFwcGVyLmdv) | `86.36% <86.36%> (ø)` | |
| [pkg/stats/stats\_recorder.go](https://codecov.io/gh/pion/interceptor/pull/143/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3N0YXRzL3N0YXRzX3JlY29yZGVyLmdv) | `89.15% <89.15%> (ø)` | |
| [internal/ntp/ntp.go](https://codecov.io/gh/pion/interceptor/pull/143/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-aW50ZXJuYWwvbnRwL250cC5nbw==) | `100.00% <100.00%> (ø)` | |
| [pkg/report/sender\_stream.go](https://codecov.io/gh/pion/interceptor/pull/143/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3JlcG9ydC9zZW5kZXJfc3RyZWFtLmdv) | `100.00% <100.00%> (ø)` | |
| [pkg/gcc/delay\_based\_bwe.go](https://codecov.io/gh/pion/interceptor/pull/143/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9kZWxheV9iYXNlZF9id2UuZ28=) | `95.45% <0.00%> (+3.03%)` | :arrow_up: |
| [pkg/gcc/arrival\_group\_accumulator.go](https://codecov.io/gh/pion/interceptor/pull/143/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9hcnJpdmFsX2dyb3VwX2FjY3VtdWxhdG9yLmdv) | `100.00% <0.00%> (+9.37%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pion/interceptor/pull/143?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/143?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion). Last update [0748586...96ad9b7](https://codecov.io/gh/pion/interceptor/pull/143?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/internal/ntp/ntp.go b/internal/ntp/ntp.go
new file mode 100644
index 0000000..f706142
--- /dev/null
+++ b/internal/ntp/ntp.go
@@ -0,0 +1,27 @@
+// Package ntp provides conversion methods between time.Time and NTP timestamps
+// stored in uint64
+package ntp
+
+import (
+	"time"
+)
+
+// ToNTP converts a time.Time oboject to an uint64 NTP timestamp
+func ToNTP(t time.Time) uint64 {
+	// seconds since 1st January 1900
+	s := (float64(t.UnixNano()) / 1000000000) + 2208988800
+
+	// higher 32 bits are the integer part, lower 32 bits are the fractional part
+	integerPart := uint32(s)
+	fractionalPart := uint32((s - float64(integerPart)) * 0xFFFFFFFF)
+	return uint64(integerPart)<<32 | uint64(fractionalPart)
+}
+
+// ToTime converts a uint64 NTP timestamps to a time.Time object
+func ToTime(t uint64) time.Time {
+	seconds := (t & 0xFFFFFFFF00000000) >> 32
+	fractional := float64(t&0x00000000FFFFFFFF) / float64(0xFFFFFFFF)
+	d := time.Duration(seconds)*time.Second + time.Duration(fractional*1e9)*time.Nanosecond
+
+	return time.Unix(0, 0).Add(-2208988800 * time.Second).Add(d)
+}
diff --git a/internal/sequencenumber/unwrapper.go b/internal/sequencenumber/unwrapper.go
new file mode 100644
index 0000000..4c2457e
--- /dev/null
+++ b/internal/sequencenumber/unwrapper.go
@@ -0,0 +1,42 @@
+// Package sequencenumber provides a sequence number unwrapper
+package sequencenumber
+
+const (
+	maxSequenceNumberPlusOne = int64(65536)
+	breakpoint               = 32768 // half of max uint16
+)
+
+// Unwrapper stores an unwrapped sequence number
+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
+}
+
+// Unwrap unwraps the next sequencenumber
+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
+}
diff --git a/pkg/report/sender_interceptor.go b/pkg/report/sender_interceptor.go
index d74b1ab..d191392 100644
--- a/pkg/report/sender_interceptor.go
+++ b/pkg/report/sender_interceptor.go
@@ -126,13 +126,3 @@ func (s *SenderInterceptor) BindLocalStream(info *interceptor.StreamInfo, writer
 		return writer.Write(header, payload, a)
 	})
 }
-
-func ntpTime(t time.Time) uint64 {
-	// seconds since 1st January 1900
-	s := (float64(t.UnixNano()) / 1000000000) + 2208988800
-
-	// higher 32 bits are the integer part, lower 32 bits are the fractional part
-	integerPart := uint32(s)
-	fractionalPart := uint32((s - float64(integerPart)) * 0xFFFFFFFF)
-	return uint64(integerPart)<<32 | uint64(fractionalPart)
-}
diff --git a/pkg/report/sender_stream.go b/pkg/report/sender_stream.go
index 51a8d02..db03223 100644
--- a/pkg/report/sender_stream.go
+++ b/pkg/report/sender_stream.go
@@ -4,6 +4,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/pion/interceptor/internal/ntp"
 	"github.com/pion/rtcp"
 	"github.com/pion/rtp"
 )
@@ -45,7 +46,7 @@ func (stream *senderStream) generateReport(now time.Time) *rtcp.SenderReport {
 
 	return &rtcp.SenderReport{
 		SSRC:        stream.ssrc,
-		NTPTime:     ntpTime(now),
+		NTPTime:     ntp.ToNTP(now),
 		RTPTime:     stream.lastRTPTimeRTP + uint32(now.Sub(stream.lastRTPTimeTime).Seconds()*stream.clockRate),
 		PacketCount: stream.packetCount,
 		OctetCount:  stream.octetCount,
diff --git a/pkg/stats/interceptor.go b/pkg/stats/interceptor.go
new file mode 100644
index 0000000..bd146bc
--- /dev/null
+++ b/pkg/stats/interceptor.go
@@ -0,0 +1,204 @@
+// Package stats provides an interceptor that records RTP/RTCP stream statistics
+package stats
+
+import (
+	"sync"
+	"time"
+
+	"github.com/pion/interceptor"
+	"github.com/pion/rtcp"
+	"github.com/pion/rtp"
+)
+
+// Option can be used to configure the stats interceptor
+type Option func(*Interceptor) error
+
+// SetRecorderFactory sets the factory that is used to create new stats
+// recorders for new streams
+func SetRecorderFactory(f RecorderFactory) Option {
+	return func(i *Interceptor) error {
+		i.RecorderFactory = f
+		return nil
+	}
+}
+
+// SetNowFunc sets the function the interceptor uses to get a current timestamp.
+// This is mostly useful for testing.
+func SetNowFunc(now func() time.Time) Option {
+	return func(i *Interceptor) error {
+		i.now = now
+		return nil
+	}
+}
+
+// Getter returns the most recent stats of a stream
+type Getter interface {
+	Get(ssrc uint32) *Stats
+}
+
+// NewPeerConnectionCallback receives a new StatsGetter for a newly created
+// PeerConnection
+type NewPeerConnectionCallback func(string, Getter)
+
+// InterceptorFactory is a interceptor.Factory for a stats Interceptor
+type InterceptorFactory struct {
+	opts              []Option
+	addPeerConnection NewPeerConnectionCallback
+}
+
+// NewInterceptor creates a new InterceptorFactory
+func NewInterceptor(opts ...Option) (*InterceptorFactory, error) {
+	return &InterceptorFactory{
+		opts:              opts,
+		addPeerConnection: nil,
+	}, nil
+}
+
+// OnNewPeerConnection sets the callback that is called when a new
+// PeerConnection is created.
+func (r *InterceptorFactory) OnNewPeerConnection(cb NewPeerConnectionCallback) {
+	r.addPeerConnection = cb
+}
+
+// NewInterceptor creates a new Interceptor
+func (r *InterceptorFactory) NewInterceptor(id string) (interceptor.Interceptor, error) {
+	i := &Interceptor{
+		NoOp: interceptor.NoOp{},
+		now:  time.Now,
+		lock: sync.Mutex{},
+		RecorderFactory: func(ssrc uint32, clockRate float64) Recorder {
+			return newRecorder(ssrc, clockRate)
+		},
+		recorders: map[uint32]Recorder{},
+		wg:        sync.WaitGroup{},
+	}
+	for _, opt := range r.opts {
+		if err := opt(i); err != nil {
+			return nil, err
+		}
+	}
+
+	if r.addPeerConnection != nil {
+		r.addPeerConnection(id, i)
+	}
+
+	return i, nil
+}
+
+// Recorder is the interface of a statistics recorder
+type Recorder interface {
+	QueueIncomingRTP(ts time.Time, buf []byte, attr interceptor.Attributes)
+	QueueIncomingRTCP(ts time.Time, buf []byte, attr interceptor.Attributes)
+	QueueOutgoingRTP(ts time.Time, header *rtp.Header, payload []byte, attr interceptor.Attributes)
+	QueueOutgoingRTCP(ts time.Time, pkts []rtcp.Packet, attr interceptor.Attributes)
+	GetStats() Stats
+	Stop()
+	Start()
+}
+
+// RecorderFactory creates new Recorders to be used by the interceptor
+type RecorderFactory func(ssrc uint32, clockRate float64) Recorder
+
+// Interceptor is the interceptor that collects stream stats
+type Interceptor struct {
+	interceptor.NoOp
+	now             func() time.Time
+	lock            sync.Mutex
+	RecorderFactory RecorderFactory
+	recorders       map[uint32]Recorder
+	wg              sync.WaitGroup
+}
+
+// Get returns the statistics for the stream with ssrc
+func (r *Interceptor) Get(ssrc uint32) *Stats {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	if rec, ok := r.recorders[ssrc]; ok {
+		stats := rec.GetStats()
+		return &stats
+	}
+	return nil
+}
+
+func (r *Interceptor) getRecorder(ssrc uint32, clockRate float64) Recorder {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	if rec, ok := r.recorders[ssrc]; ok {
+		return rec
+	}
+	rec := r.RecorderFactory(ssrc, clockRate)
+	r.wg.Add(1)
+	go func() {
+		defer r.wg.Done()
+		rec.Start()
+	}()
+	r.recorders[ssrc] = rec
+	return rec
+}
+
+// Close closes the interceptor and associated stats recorders
+func (r *Interceptor) Close() error {
+	defer r.wg.Wait()
+
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	for _, r := range r.recorders {
+		r.Stop()
+	}
+	return 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 (r *Interceptor) BindRTCPReader(reader interceptor.RTCPReader) interceptor.RTCPReader {
+	return interceptor.RTCPReaderFunc(func(bytes []byte, attributes interceptor.Attributes) (int, interceptor.Attributes, error) {
+		n, attattributes, err := reader.Read(bytes, attributes)
+		if err != nil {
+			return 0, attattributes, err
+		}
+		r.lock.Lock()
+		for _, recorder := range r.recorders {
+			recorder.QueueIncomingRTCP(r.now(), bytes[:n], attributes)
+		}
+		r.lock.Unlock()
+		return n, attattributes, err
+	})
+}
+
+// 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 (r *Interceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter {
+	return interceptor.RTCPWriterFunc(func(pkts []rtcp.Packet, attributes interceptor.Attributes) (int, error) {
+		r.lock.Lock()
+		for _, recorder := range r.recorders {
+			recorder.QueueOutgoingRTCP(r.now(), pkts, attributes)
+		}
+		r.lock.Unlock()
+		return writer.Write(pkts, attributes)
+	})
+}
+
+// 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 (r *Interceptor) BindLocalStream(info *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter {
+	recorder := r.getRecorder(info.SSRC, float64(info.ClockRate))
+	return interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
+		recorder.QueueOutgoingRTP(r.now(), header, payload, attributes)
+		return writer.Write(header, payload, attributes)
+	})
+}
+
+// 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 (r *Interceptor) BindRemoteStream(info *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader {
+	recorder := r.getRecorder(info.SSRC, float64(info.ClockRate))
+	return interceptor.RTPReaderFunc(func(bytes []byte, attributes interceptor.Attributes) (int, interceptor.Attributes, error) {
+		n, attributes, err := reader.Read(bytes, attributes)
+		if err != nil {
+			return 0, nil, err
+		}
+		recorder.QueueIncomingRTP(r.now(), bytes[:n], attributes)
+		return n, attributes, nil
+	})
+}
diff --git a/pkg/stats/received_stats.go b/pkg/stats/received_stats.go
new file mode 100644
index 0000000..c1e8924
--- /dev/null
+++ b/pkg/stats/received_stats.go
@@ -0,0 +1,68 @@
+package stats
+
+import (
+	"fmt"
+	"time"
+)
+
+// ReceivedRTPStreamStats contains common receiver stats of RTP streams
+type ReceivedRTPStreamStats struct {
+	PacketsReceived uint64
+	PacketsLost     int64
+	Jitter          float64
+}
+
+// String returns a string representation of ReceivedRTPStreamStats
+func (s ReceivedRTPStreamStats) String() string {
+	out := fmt.Sprintf("\tPacketsReceived: %v\n", s.PacketsReceived)
+	out += fmt.Sprintf("\tPacketsLost: %v\n", s.PacketsLost)
+	out += fmt.Sprintf("\tJitter: %v\n", s.Jitter)
+	return out
+}
+
+// InboundRTPStreamStats contains stats of inbound RTP streams
+type InboundRTPStreamStats struct {
+	ReceivedRTPStreamStats
+
+	LastPacketReceivedTimestamp time.Time
+	HeaderBytesReceived         uint64
+	BytesReceived               uint64
+	FIRCount                    uint32
+	PLICount                    uint32
+	NACKCount                   uint32
+}
+
+// String returns a string representation of InboundRTPStreamStats
+func (s InboundRTPStreamStats) String() string {
+	out := "InboundRTPStreamStats:\n"
+	out += s.ReceivedRTPStreamStats.String()
+	out += fmt.Sprintf("\tLastPacketReceivedTimestamp: %v\n", s.LastPacketReceivedTimestamp)
+	out += fmt.Sprintf("\tHeaderBytesReceived: %v\n", s.HeaderBytesReceived)
+	out += fmt.Sprintf("\tBytesReceived: %v\n", s.BytesReceived)
+	out += fmt.Sprintf("\tFIRCount: %v\n", s.FIRCount)
+	out += fmt.Sprintf("\tPLICount: %v\n", s.PLICount)
+	out += fmt.Sprintf("\tNACKCount: %v\n", s.NACKCount)
+	return out
+}
+
+// RemoteInboundRTPStreamStats contains stats of inbound RTP streams of the
+// remote peer
+type RemoteInboundRTPStreamStats struct {
+	ReceivedRTPStreamStats
+
+	RoundTripTime             time.Duration
+	TotalRoundTripTime        time.Duration
+	FractionLost              float64
+	RoundTripTimeMeasurements uint64
+}
+
+// String returns a string representation of RemoteInboundRTPStreamStats
+func (s RemoteInboundRTPStreamStats) String() string {
+	out := "RemoteInboundRTPStreamStats:\n"
+	out += s.ReceivedRTPStreamStats.String()
+	out += fmt.Sprintf("\tRoundTripTime: %v\n", s.RoundTripTime)
+	out += fmt.Sprintf("\tTotalRoundTripTime: %v\n", s.TotalRoundTripTime)
+	out += fmt.Sprintf("\tFractionLost: %v\n", s.FractionLost)
+	out += fmt.Sprintf("\tRoundTripTimeMeasurements: %v\n", s.RoundTripTimeMeasurements)
+	return out
+}
diff --git a/pkg/stats/sent_stats.go b/pkg/stats/sent_stats.go
new file mode 100644
index 0000000..de6ac3c
--- /dev/null
+++ b/pkg/stats/sent_stats.go
@@ -0,0 +1,64 @@
+package stats
+
+import (
+	"fmt"
+	"time"
+)
+
+// SentRTPStreamStats contains common sender stats of RTP streams
+type SentRTPStreamStats struct {
+	PacketsSent uint64
+	BytesSent   uint64
+}
+
+// String returns a string representation of SentRTPStreamStats
+func (s SentRTPStreamStats) String() string {
+	out := fmt.Sprintf("\tPacketsSent: %v\n", s.PacketsSent)
+	out += fmt.Sprintf("\tBytesSent: %v\n", s.BytesSent)
+	return out
+}
+
+// OutboundRTPStreamStats contains stats of outbound RTP streams
+type OutboundRTPStreamStats struct {
+	SentRTPStreamStats
+
+	HeaderBytesSent uint64
+	NACKCount       uint32
+	FIRCount        uint32
+	PLICount        uint32
+}
+
+// String returns a string representation of OutboundRTPStreamStats
+func (s OutboundRTPStreamStats) String() string {
+	out := "OutboundRTPStreamStats\n"
+	out += s.SentRTPStreamStats.String()
+	out += fmt.Sprintf("\tHeaderBytesSent: %v\n", s.HeaderBytesSent)
+	out += fmt.Sprintf("\tNACKCount: %v\n", s.NACKCount)
+	out += fmt.Sprintf("\tFIRCount: %v\n", s.FIRCount)
+	out += fmt.Sprintf("\tPLICount: %v\n", s.PLICount)
+	return out
+}
+
+// RemoteOutboundRTPStreamStats contains stats of outbound RTP streams of the
+// remote peer
+type RemoteOutboundRTPStreamStats struct {
+	SentRTPStreamStats
+
+	RemoteTimeStamp           time.Time
+	ReportsSent               uint64
+	RoundTripTime             time.Duration
+	TotalRoundTripTime        time.Duration
+	RoundTripTimeMeasurements uint64
+}
+
+// String returns a string representation of RemoteOutboundRTPStreamStats
+func (s RemoteOutboundRTPStreamStats) String() string {
+	out := "RemoteOutboundRTPStreamStats:\n"
+	out += s.SentRTPStreamStats.String()
+	out += fmt.Sprintf("\tRemoteTimeStamp: %v\n", s.RemoteTimeStamp)
+	out += fmt.Sprintf("\tReportsSent: %v\n", s.ReportsSent)
+	out += fmt.Sprintf("\tRoundTripTime: %v\n", s.RoundTripTime)
+	out += fmt.Sprintf("\tTotalRoundTripTime: %v\n", s.TotalRoundTripTime)
+	out += fmt.Sprintf("\tRoundTripTimeMeasurements: %v\n", s.RoundTripTimeMeasurements)
+	return out
+}
diff --git a/pkg/stats/stats_recorder.go b/pkg/stats/stats_recorder.go
new file mode 100644
index 0000000..e6206fb
--- /dev/null
+++ b/pkg/stats/stats_recorder.go
@@ -0,0 +1,352 @@
+package stats
+
+import (
+	"time"
+
+	"github.com/pion/interceptor"
+	"github.com/pion/interceptor/internal/ntp"
+	"github.com/pion/interceptor/internal/sequencenumber"
+	"github.com/pion/logging"
+	"github.com/pion/rtcp"
+	"github.com/pion/rtp"
+)
+
+// Stats contains all the available statistics of RTP streams
+type Stats struct {
+	InboundRTPStreamStats
+	OutboundRTPStreamStats
+	RemoteInboundRTPStreamStats
+	RemoteOutboundRTPStreamStats
+}
+
+type internalStats struct {
+	inboundSequencerNumber           sequencenumber.Unwrapper
+	inboundSequenceNumberInitialized bool
+	inboundFirstSequenceNumber       int64
+	inboundHighestSequenceNumber     int64
+
+	inboundLastArrivalInitialized bool
+	inboundLastArrival            time.Time
+	inboundLastTransit            int
+
+	remoteInboundFirstSequenceNumberInitialized bool
+	remoteInboundFirstSequenceNumber            int64
+
+	lastSenderReports []uint64
+
+	lastReceiverReferenceTimes []uint64
+
+	InboundRTPStreamStats
+	OutboundRTPStreamStats
+
+	RemoteInboundRTPStreamStats
+	RemoteOutboundRTPStreamStats
+}
+
+type incomingRTP struct {
+	ts         time.Time
+	header     rtp.Header
+	payloadLen int
+	attr       interceptor.Attributes
+}
+
+type incomingRTCP struct {
+	ts   time.Time
+	pkts []rtcp.Packet
+	attr interceptor.Attributes
+}
+
+type outgoingRTP struct {
+	ts         time.Time
+	header     rtp.Header
+	payloadLen int
+	attr       interceptor.Attributes
+}
+
+type outgoingRTCP struct {
+	ts   time.Time
+	pkts []rtcp.Packet
+	attr interceptor.Attributes
+}
+
+type recorder struct {
+	logger logging.LeveledLogger
+
+	ssrc      uint32
+	clockRate float64
+
+	maxLastSenderReports          int
+	maxLastReceiverReferenceTimes int
+
+	incomingRTPChan  chan *incomingRTP
+	incomingRTCPChan chan *incomingRTCP
+	outgoingRTPChan  chan *outgoingRTP
+	outgoingRTCPChan chan *outgoingRTCP
+	getStatsChan     chan Stats
+	done             chan struct{}
+}
+
+func newRecorder(ssrc uint32, clockRate float64) *recorder {
+	return &recorder{
+		logger:                        logging.NewDefaultLoggerFactory().NewLogger("stats_recorder"),
+		ssrc:                          ssrc,
+		clockRate:                     clockRate,
+		maxLastSenderReports:          5,
+		maxLastReceiverReferenceTimes: 5,
+		incomingRTPChan:               make(chan *incomingRTP),
+		incomingRTCPChan:              make(chan *incomingRTCP),
+		outgoingRTPChan:               make(chan *outgoingRTP),
+		outgoingRTCPChan:              make(chan *outgoingRTCP),
+		getStatsChan:                  make(chan Stats),
+		done:                          make(chan struct{}),
+	}
+}
+
+func (r *recorder) Stop() {
+	close(r.done)
+}
+
+func (r *recorder) GetStats() Stats {
+	return <-r.getStatsChan
+}
+
+func (r *recorder) recordIncomingRTP(latestStats internalStats, v *incomingRTP) internalStats {
+	sequenceNumber := latestStats.inboundSequencerNumber.Unwrap(v.header.SequenceNumber)
+	if !latestStats.inboundSequenceNumberInitialized {
+		latestStats.inboundFirstSequenceNumber = sequenceNumber
+		latestStats.inboundSequenceNumberInitialized = true
+	}
+	if sequenceNumber > latestStats.inboundHighestSequenceNumber {
+		latestStats.inboundHighestSequenceNumber = sequenceNumber
+	}
+
+	latestStats.InboundRTPStreamStats.PacketsReceived++
+	expectedPackets := latestStats.inboundHighestSequenceNumber - latestStats.inboundFirstSequenceNumber + 1
+	latestStats.InboundRTPStreamStats.PacketsLost = expectedPackets - int64(latestStats.InboundRTPStreamStats.PacketsReceived)
+
+	if !latestStats.inboundLastArrivalInitialized {
+		latestStats.inboundLastArrival = v.ts
+		latestStats.inboundLastArrivalInitialized = true
+	} else {
+		arrival := int(v.ts.Sub(latestStats.inboundLastArrival).Seconds() * r.clockRate)
+		transit := arrival - int(v.header.Timestamp)
+		d := transit - latestStats.inboundLastTransit
+		latestStats.inboundLastTransit = transit
+		if d < 0 {
+			d = -d
+		}
+		latestStats.InboundRTPStreamStats.Jitter += (1.0 / 16.0) * (float64(d) - latestStats.InboundRTPStreamStats.Jitter)
+		latestStats.inboundLastArrival = v.ts
+	}
+
+	latestStats.LastPacketReceivedTimestamp = v.ts
+	latestStats.HeaderBytesReceived += uint64(v.header.MarshalSize())
+	latestStats.BytesReceived += uint64(v.header.MarshalSize() + v.payloadLen)
+	return latestStats
+}
+
+func (r *recorder) recordOutgoingRTCP(latestStats internalStats, v *outgoingRTCP) internalStats {
+	for _, pkt := range v.pkts {
+		switch rtcpPkt := pkt.(type) {
+		case *rtcp.FullIntraRequest:
+			latestStats.InboundRTPStreamStats.FIRCount++
+		case *rtcp.PictureLossIndication:
+			latestStats.InboundRTPStreamStats.PLICount++
+		case *rtcp.TransportLayerNack:
+			latestStats.InboundRTPStreamStats.NACKCount++
+		case *rtcp.SenderReport:
+			latestStats.lastSenderReports = append(latestStats.lastSenderReports, rtcpPkt.NTPTime)
+			if len(latestStats.lastSenderReports) > r.maxLastSenderReports {
+				latestStats.lastSenderReports = latestStats.lastSenderReports[len(latestStats.lastSenderReports)-r.maxLastSenderReports:]
+			}
+		case *rtcp.ExtendedReport:
+			for _, block := range rtcpPkt.Reports {
+				if xr, ok := block.(*rtcp.ReceiverReferenceTimeReportBlock); ok {
+					latestStats.lastReceiverReferenceTimes = append(latestStats.lastReceiverReferenceTimes, xr.NTPTimestamp)
+					if len(latestStats.lastReceiverReferenceTimes) > r.maxLastReceiverReferenceTimes {
+						latestStats.lastReceiverReferenceTimes = latestStats.lastReceiverReferenceTimes[len(latestStats.lastReceiverReferenceTimes)-r.maxLastReceiverReferenceTimes:]
+					}
+				}
+			}
+		}
+	}
+	return latestStats
+}
+
+func (r *recorder) recordOutgoingRTP(latestStats internalStats, v *outgoingRTP) internalStats {
+	headerSize := v.header.MarshalSize()
+	latestStats.OutboundRTPStreamStats.PacketsSent++
+	latestStats.OutboundRTPStreamStats.BytesSent += uint64(headerSize + v.payloadLen)
+	latestStats.HeaderBytesSent += uint64(headerSize)
+	if !latestStats.remoteInboundFirstSequenceNumberInitialized {
+		latestStats.remoteInboundFirstSequenceNumber = int64(v.header.SequenceNumber)
+		latestStats.remoteInboundFirstSequenceNumberInitialized = true
+	}
+	return latestStats
+}
+
+func (r *recorder) recordIncomingRR(latestStats internalStats, pkt *rtcp.ReceiverReport, ts time.Time) internalStats {
+	for _, report := range pkt.Reports {
+		if report.SSRC == r.ssrc {
+			if latestStats.remoteInboundFirstSequenceNumberInitialized {
+				cycles := uint64(report.LastSequenceNumber & 0xFFFF0000)
+				nr := uint64(report.LastSequenceNumber & 0x0000FFFF)
+				highest := cycles*0xFFFF + nr
+				latestStats.RemoteInboundRTPStreamStats.PacketsReceived = highest - uint64(report.TotalLost) - uint64(latestStats.remoteInboundFirstSequenceNumber) + 1
+			}
+			latestStats.RemoteInboundRTPStreamStats.PacketsLost = int64(report.TotalLost)
+			latestStats.RemoteInboundRTPStreamStats.Jitter = float64(report.Jitter) / r.clockRate
+
+			if report.Delay != 0 && report.LastSenderReport != 0 {
+				for i := min(r.maxLastSenderReports, len(latestStats.lastSenderReports)) - 1; i >= 0; i-- {
+					lastReport := latestStats.lastSenderReports[i]
+					if (lastReport&0x0000FFFFFFFF0000)>>16 == uint64(report.LastSenderReport) {
+						dlsr := time.Duration(float64(report.Delay) / 65536.0 * float64(time.Second))
+						latestStats.RemoteInboundRTPStreamStats.RoundTripTime = (ts.Add(-dlsr)).Sub(ntp.ToTime(lastReport))
+						latestStats.RemoteInboundRTPStreamStats.TotalRoundTripTime += latestStats.RemoteInboundRTPStreamStats.RoundTripTime
+						latestStats.RemoteInboundRTPStreamStats.RoundTripTimeMeasurements++
+						break
+					}
+				}
+			}
+			latestStats.FractionLost = float64(report.FractionLost) / 256.0
+		}
+	}
+	return latestStats
+}
+
+func (r *recorder) recordIncomingXR(latestStats internalStats, pkt *rtcp.ExtendedReport, ts time.Time) internalStats {
+	for _, report := range pkt.Reports {
+		if xr, ok := report.(*rtcp.DLRRReportBlock); ok {
+			for _, xrReport := range xr.Reports {
+				if xrReport.LastRR != 0 && xrReport.DLRR != 0 {
+					for i := min(r.maxLastReceiverReferenceTimes, len(latestStats.lastReceiverReferenceTimes)) - 1; i >= 0; i-- {
+						lastRR := latestStats.lastReceiverReferenceTimes[i]
+						if (lastRR&0x0000FFFFFFFF0000)>>16 == uint64(xrReport.LastRR) {
+							dlrr := time.Duration(xrReport.DLRR/65536.0) * time.Second
+							latestStats.RemoteOutboundRTPStreamStats.RoundTripTime = (ts.Add(-dlrr)).Sub(ntp.ToTime(lastRR))
+							latestStats.RemoteOutboundRTPStreamStats.TotalRoundTripTime += latestStats.RemoteOutboundRTPStreamStats.RoundTripTime
+							latestStats.RemoteOutboundRTPStreamStats.RoundTripTimeMeasurements++
+						}
+					}
+				}
+			}
+		}
+	}
+	return latestStats
+}
+
+func (r *recorder) recordIncomingRTCP(latestStats internalStats, v *incomingRTCP) internalStats {
+	for _, pkt := range v.pkts {
+		switch pkt := pkt.(type) {
+		case *rtcp.TransportLayerNack:
+			latestStats.OutboundRTPStreamStats.NACKCount++
+		case *rtcp.FullIntraRequest:
+			latestStats.OutboundRTPStreamStats.FIRCount++
+		case *rtcp.PictureLossIndication:
+			latestStats.OutboundRTPStreamStats.PLICount++
+		case *rtcp.ReceiverReport:
+			return r.recordIncomingRR(latestStats, pkt, v.ts)
+		case *rtcp.SenderReport:
+			latestStats.RemoteOutboundRTPStreamStats.PacketsSent = uint64(pkt.PacketCount)
+			latestStats.RemoteOutboundRTPStreamStats.BytesSent = uint64(pkt.OctetCount)
+			latestStats.RemoteTimeStamp = ntp.ToTime(pkt.NTPTime)
+			latestStats.ReportsSent++
+
+		case *rtcp.ExtendedReport:
+			return r.recordIncomingXR(latestStats, pkt, v.ts)
+		}
+	}
+	return latestStats
+}
+
+func (r *recorder) Start() {
+	latestStats := &internalStats{}
+	for {
+		select {
+		case <-r.done:
+			return
+		case v := <-r.incomingRTPChan:
+			s := r.recordIncomingRTP(*latestStats, v)
+			latestStats = &s
+
+		case v := <-r.outgoingRTCPChan:
+			s := r.recordOutgoingRTCP(*latestStats, v)
+			latestStats = &s
+
+		case v := <-r.outgoingRTPChan:
+			s := r.recordOutgoingRTP(*latestStats, v)
+			latestStats = &s
+
+		case v := <-r.incomingRTCPChan:
+			s := r.recordIncomingRTCP(*latestStats, v)
+			latestStats = &s
+
+		case r.getStatsChan <- Stats{
+			InboundRTPStreamStats:        latestStats.InboundRTPStreamStats,
+			OutboundRTPStreamStats:       latestStats.OutboundRTPStreamStats,
+			RemoteInboundRTPStreamStats:  latestStats.RemoteInboundRTPStreamStats,
+			RemoteOutboundRTPStreamStats: latestStats.RemoteOutboundRTPStreamStats,
+		}:
+		}
+	}
+}
+
+func (r *recorder) QueueIncomingRTP(ts time.Time, buf []byte, attr interceptor.Attributes) {
+	if attr == nil {
+		attr = make(interceptor.Attributes)
+	}
+	header, err := attr.GetRTPHeader(buf)
+	if err != nil {
+		r.logger.Warnf("failed to get RTP Header, skipping incoming RTP packet in stats calculation: %v", err)
+		return
+	}
+	hdr := header.Clone()
+	r.incomingRTPChan <- &incomingRTP{
+		ts:         ts,
+		header:     hdr,
+		payloadLen: len(buf) - hdr.MarshalSize(),
+		attr:       attr,
+	}
+}
+
+func (r *recorder) QueueIncomingRTCP(ts time.Time, buf []byte, attr interceptor.Attributes) {
+	if attr == nil {
+		attr = make(interceptor.Attributes)
+	}
+	pkts, err := attr.GetRTCPPackets(buf)
+	if err != nil {
+		r.logger.Warnf("failed to get RTCP packets, skipping incoming RTCP packet in stats calculation: %v", err)
+		return
+	}
+	r.incomingRTCPChan <- &incomingRTCP{
+		ts:   ts,
+		pkts: pkts,
+		attr: attr,
+	}
+}
+
+func (r *recorder) QueueOutgoingRTP(ts time.Time, header *rtp.Header, payload []byte, attr interceptor.Attributes) {
+	hdr := header.Clone()
+	r.outgoingRTPChan <- &outgoingRTP{
+		ts:         ts,
+		header:     hdr,
+		payloadLen: len(payload),
+		attr:       attr,
+	}
+}
+
+func (r *recorder) QueueOutgoingRTCP(ts time.Time, pkts []rtcp.Packet, attr interceptor.Attributes) {
+	r.outgoingRTCPChan <- &outgoingRTCP{
+		ts:   ts,
+		pkts: pkts,
+		attr: attr,
+	}
+}
+
+func min(a, b int) int {
+	if a < b {
+		return a
+	}
+	return b
+}


Test patch:
diff --git a/internal/ntp/ntp_test.go b/internal/ntp/ntp_test.go
new file mode 100644
index 0000000..988e7f6
--- /dev/null
+++ b/internal/ntp/ntp_test.go
@@ -0,0 +1,49 @@
+package ntp
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNTPTimeConverstion(t *testing.T) {
+	for i, cc := range []struct {
+		ts time.Time
+	}{
+		{
+			ts: time.Now(),
+		},
+		{
+			ts: time.Unix(0, 0),
+		},
+	} {
+		t.Run(fmt.Sprintf("TimeToNTP/%v", i), func(t *testing.T) {
+			assert.InDelta(t, cc.ts.UnixNano(), ToTime(ToNTP(cc.ts)).UnixNano(), float64(time.Millisecond.Nanoseconds()))
+		})
+	}
+}
+
+func TestTimeToNTPConverstion(t *testing.T) {
+	for i, cc := range []struct {
+		ts uint64
+	}{
+		{
+			ts: 0,
+		},
+		{
+			ts: 65535,
+		},
+		{
+			ts: 16606669245815957503,
+		},
+		{
+			ts: 9487534653230284800,
+		},
+	} {
+		t.Run(fmt.Sprintf("TimeToNTP/%v", i), func(t *testing.T) {
+			assert.Equal(t, cc.ts, ToNTP(ToTime(cc.ts)))
+		})
+	}
+}
diff --git a/internal/sequencenumber/unwrapper_test.go b/internal/sequencenumber/unwrapper_test.go
new file mode 100644
index 0000000..67dd1a3
--- /dev/null
+++ b/internal/sequencenumber/unwrapper_test.go
@@ -0,0 +1,104 @@
+package sequencenumber
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsNewer(t *testing.T) {
+	cases := []struct {
+		a, b     uint16
+		expected bool
+	}{
+		{
+			a:        1,
+			b:        0,
+			expected: true,
+		},
+		{
+			a:        65534,
+			b:        65535,
+			expected: false,
+		},
+		{
+			a:        65535,
+			b:        65535,
+			expected: false,
+		},
+		{
+			a:        0,
+			b:        65535,
+			expected: true,
+		},
+		{
+			a:        0,
+			b:        32767,
+			expected: false,
+		},
+		{
+			a:        32770,
+			b:        2,
+			expected: true,
+		},
+		{
+			a:        3,
+			b:        32770,
+			expected: false,
+		},
+	}
+	for i, tc := range cases {
+		t.Run(fmt.Sprintf("%v", i), func(t *testing.T) {
+			assert.Equalf(t, tc.expected, isNewer(tc.a, tc.b), "expected isNewer(%v, %v) to be %v", tc.a, tc.b, tc.expected)
+		})
+	}
+}
+
+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, 32767, 32768, 32769, 32770,
+				1, 2, 32765, 32770, 65535,
+			},
+			expected: []int64{
+				0, 32767, 32768, 32769, 32770,
+				65537, 65538, 98301, 98306, 131071,
+			},
+		},
+	}
+	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)
+		})
+	}
+}
diff --git a/pkg/report/receiver_interceptor_test.go b/pkg/report/receiver_interceptor_test.go
index 2056689..412122c 100644
--- a/pkg/report/receiver_interceptor_test.go
+++ b/pkg/report/receiver_interceptor_test.go
@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/pion/interceptor"
+	"github.com/pion/interceptor/internal/ntp"
 	"github.com/pion/interceptor/internal/test"
 	"github.com/pion/logging"
 	"github.com/pion/rtcp"
@@ -123,7 +124,7 @@ func TestReceiverInterceptor(t *testing.T) {
 		stream.ReceiveRTCP([]rtcp.Packet{
 			&rtcp.SenderReport{
 				SSRC:        123456,
-				NTPTime:     ntpTime(now),
+				NTPTime:     ntp.ToNTP(now),
 				RTPTime:     987654321 + uint32(now.Sub(rtpTime).Seconds()*90000),
 				PacketCount: 10,
 				OctetCount:  0,
@@ -237,7 +238,7 @@ func TestReceiverInterceptor(t *testing.T) {
 		stream.ReceiveRTCP([]rtcp.Packet{
 			&rtcp.SenderReport{
 				SSRC:        123456,
-				NTPTime:     ntpTime(now),
+				NTPTime:     ntp.ToNTP(now),
 				RTPTime:     987654321 + uint32(now.Sub(rtpTime).Seconds()*90000),
 				PacketCount: 10,
 				OctetCount:  0,
@@ -419,7 +420,7 @@ func TestReceiverInterceptor(t *testing.T) {
 		stream.ReceiveRTCP([]rtcp.Packet{
 			&rtcp.SenderReport{
 				SSRC:        123456,
-				NTPTime:     ntpTime(time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)),
+				NTPTime:     ntp.ToNTP(time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)),
 				RTPTime:     987654321,
 				PacketCount: 0,
 				OctetCount:  0,
diff --git a/pkg/report/sender_interceptor_test.go b/pkg/report/sender_interceptor_test.go
index 6c891a8..37d6f19 100644
--- a/pkg/report/sender_interceptor_test.go
+++ b/pkg/report/sender_interceptor_test.go
@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/pion/interceptor"
+	"github.com/pion/interceptor/internal/ntp"
 	"github.com/pion/interceptor/internal/test"
 	"github.com/pion/logging"
 	"github.com/pion/rtcp"
@@ -40,7 +41,7 @@ func TestSenderInterceptor(t *testing.T) {
 		assert.True(t, ok)
 		assert.Equal(t, &rtcp.SenderReport{
 			SSRC:        123456,
-			NTPTime:     ntpTime(mt.Now()),
+			NTPTime:     ntp.ToNTP(mt.Now()),
 			RTPTime:     2269117121,
 			PacketCount: 0,
 			OctetCount:  0,
@@ -81,7 +82,7 @@ func TestSenderInterceptor(t *testing.T) {
 		assert.True(t, ok)
 		assert.Equal(t, &rtcp.SenderReport{
 			SSRC:        123456,
-			NTPTime:     ntpTime(mt.Now()),
+			NTPTime:     ntp.ToNTP(mt.Now()),
 			RTPTime:     2269117121,
 			PacketCount: 10,
 			OctetCount:  20,
diff --git a/pkg/stats/interceptor_test.go b/pkg/stats/interceptor_test.go
new file mode 100644
index 0000000..4babfd9
--- /dev/null
+++ b/pkg/stats/interceptor_test.go
@@ -0,0 +1,229 @@
+package stats
+
+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 packets", func(t *testing.T) {
+		f, err := NewInterceptor()
+		assert.NoError(t, err)
+		statsCh := make(chan Getter)
+		f.OnNewPeerConnection(func(_ string, g Getter) {
+			go func() {
+				statsCh <- g
+			}()
+		})
+
+		i, err := f.NewInterceptor("")
+		assert.NoError(t, err)
+
+		stream := test.NewMockStream(&interceptor.StreamInfo{SSRC: 0}, i)
+		defer func() {
+			assert.NoError(t, stream.Close())
+		}()
+
+		var statsGetter Getter
+		select {
+		case statsGetter = <-statsCh:
+		case <-time.After(time.Second):
+			assert.FailNow(t, "expected to receive statsgetter")
+		}
+
+		assert.Equal(t, statsGetter.Get(0), &Stats{})
+	})
+
+	t.Run("records packets", func(t *testing.T) {
+		mockRecorder := newMockRecorder()
+		now := time.Now()
+		f, err := NewInterceptor(
+			SetRecorderFactory(func(ssrc uint32, clockRate float64) Recorder {
+				return mockRecorder
+			}),
+			SetNowFunc(func() time.Time {
+				return now
+			}),
+		)
+		assert.NoError(t, err)
+		statsCh := make(chan Getter)
+		f.OnNewPeerConnection(func(_ string, g Getter) {
+			go func() {
+				statsCh <- g
+			}()
+		})
+
+		i, err := f.NewInterceptor("")
+		assert.NoError(t, err)
+
+		stream := test.NewMockStream(&interceptor.StreamInfo{SSRC: 0}, i)
+		defer func() {
+			assert.NoError(t, stream.Close())
+		}()
+
+		incomingRTP := &rtp.Packet{}
+		incomingRTCP := []rtcp.Packet{&rtcp.RawPacket{}}
+		outgoingRTP := &rtp.Packet{}
+		outgoingRTCP := []rtcp.Packet{&rtcp.RawPacket{}}
+
+		stream.ReceiveRTP(incomingRTP)
+		stream.ReceiveRTCP(incomingRTCP)
+		assert.NoError(t, stream.WriteRTP(outgoingRTP))
+		assert.NoError(t, stream.WriteRTCP(outgoingRTCP))
+
+		var statsGetter Getter
+		select {
+		case statsGetter = <-statsCh:
+		case <-time.After(time.Second):
+			assert.FailNow(t, "expected to receive statsgetter")
+		}
+
+		var riRTP recordedIncomingRTP
+		select {
+		case riRTP = <-mockRecorder.incomingRTPQueue:
+		case <-time.After(time.Second):
+			assert.FailNow(t, "expected to record RTP packet")
+		}
+
+		var riRTCP recordedIncomingRTCP
+		select {
+		case riRTCP = <-mockRecorder.incomingRTCPQueue:
+		case <-time.After(time.Second):
+		}
+
+		var roRTP recordedOutgoingRTP
+		select {
+		case roRTP = <-mockRecorder.outgoingRTPQueue:
+		case <-time.After(time.Second):
+		}
+
+		var roRTCP recordedOutgoingRTCP
+		select {
+		case roRTCP = <-mockRecorder.outgoingRTCPQueue:
+		case <-time.After(time.Second):
+		}
+
+		assert.Equal(t, &Stats{}, statsGetter.Get(0))
+
+		buf, err := incomingRTP.Marshal()
+		assert.NoError(t, err)
+		expectedIncomingRTP := recordedIncomingRTP{
+			ts:   now,
+			buf:  buf,
+			attr: map[interface{}]interface{}{},
+		}
+		assert.Equal(t, expectedIncomingRTP, riRTP)
+
+		buf, err = rtcp.Marshal(incomingRTCP)
+		assert.NoError(t, err)
+		expectedIncomingRTCP := recordedIncomingRTCP{
+			ts:   now,
+			buf:  buf,
+			attr: map[interface{}]interface{}{},
+		}
+		assert.Equal(t, expectedIncomingRTCP, riRTCP)
+
+		expectedOutgoingRTP := recordedOutgoingRTP{
+			ts:      now,
+			header:  &rtp.Header{},
+			payload: outgoingRTP.Payload,
+			attr:    map[interface{}]interface{}{},
+		}
+		assert.Equal(t, expectedOutgoingRTP, roRTP)
+
+		expectedOutgoingRTCP := recordedOutgoingRTCP{
+			ts:   now,
+			pkts: outgoingRTCP,
+			attr: map[interface{}]interface{}{},
+		}
+		assert.Equal(t, expectedOutgoingRTCP, roRTCP)
+	})
+}
+
+type recordedOutgoingRTP struct {
+	ts      time.Time
+	header  *rtp.Header
+	payload []byte
+	attr    interceptor.Attributes
+}
+
+type recordedOutgoingRTCP struct {
+	ts   time.Time
+	pkts []rtcp.Packet
+	attr interceptor.Attributes
+}
+
+type recordedIncomingRTP struct {
+	ts   time.Time
+	buf  []byte
+	attr interceptor.Attributes
+}
+
+type recordedIncomingRTCP struct {
+	ts   time.Time
+	buf  []byte
+	attr interceptor.Attributes
+}
+
+type mockRecorder struct {
+	incomingRTPQueue  chan recordedIncomingRTP
+	incomingRTCPQueue chan recordedIncomingRTCP
+	outgoingRTPQueue  chan recordedOutgoingRTP
+	outgoingRTCPQueue chan recordedOutgoingRTCP
+}
+
+func newMockRecorder() *mockRecorder {
+	return &mockRecorder{
+		incomingRTPQueue:  make(chan recordedIncomingRTP, 1),
+		incomingRTCPQueue: make(chan recordedIncomingRTCP, 1),
+		outgoingRTPQueue:  make(chan recordedOutgoingRTP, 1),
+		outgoingRTCPQueue: make(chan recordedOutgoingRTCP, 1),
+	}
+}
+
+func (r *mockRecorder) QueueIncomingRTP(ts time.Time, buf []byte, attr interceptor.Attributes) {
+	r.incomingRTPQueue <- recordedIncomingRTP{
+		ts:   ts,
+		buf:  buf,
+		attr: attr,
+	}
+}
+
+func (r *mockRecorder) QueueIncomingRTCP(ts time.Time, buf []byte, attr interceptor.Attributes) {
+	r.incomingRTCPQueue <- recordedIncomingRTCP{
+		ts:   ts,
+		buf:  buf,
+		attr: attr,
+	}
+}
+
+func (r *mockRecorder) QueueOutgoingRTP(ts time.Time, header *rtp.Header, payload []byte, attr interceptor.Attributes) {
+	r.outgoingRTPQueue <- recordedOutgoingRTP{
+		ts:      ts,
+		header:  header,
+		payload: payload,
+		attr:    attr,
+	}
+}
+
+func (r *mockRecorder) QueueOutgoingRTCP(ts time.Time, pkts []rtcp.Packet, attr interceptor.Attributes) {
+	r.outgoingRTCPQueue <- recordedOutgoingRTCP{
+		ts:   ts,
+		pkts: pkts,
+		attr: attr,
+	}
+}
+
+func (r *mockRecorder) GetStats() Stats {
+	return Stats{}
+}
+
+func (r *mockRecorder) Start() {}
+
+func (r *mockRecorder) Stop() {}
diff --git a/pkg/stats/stats_recorder_test.go b/pkg/stats/stats_recorder_test.go
new file mode 100644
index 0000000..d0a7938
--- /dev/null
+++ b/pkg/stats/stats_recorder_test.go
@@ -0,0 +1,291 @@
+package stats
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/pion/interceptor/internal/ntp"
+	"github.com/pion/rtcp"
+	"github.com/pion/rtp"
+	"github.com/stretchr/testify/assert"
+)
+
+func mustMarshalRTP(t *testing.T, pkt rtp.Packet) []byte {
+	buf, err := pkt.Marshal()
+	assert.NoError(t, err)
+	return buf
+}
+
+func mustMarshalRTCPs(t *testing.T, pkt rtcp.Packet) []byte {
+	buf, err := pkt.Marshal()
+	assert.NoError(t, err)
+	return buf
+}
+
+func TestStatsRecorder(t *testing.T) {
+	cname := &rtcp.SourceDescription{
+		Chunks: []rtcp.SourceDescriptionChunk{{
+			Source: 1234,
+			Items: []rtcp.SourceDescriptionItem{{
+				Type: rtcp.SDESCNAME,
+				Text: "cname",
+			}},
+		}},
+	}
+	type record struct {
+		ts      time.Time
+		content interface{}
+	}
+	type input struct {
+		name string
+
+		records []record
+
+		expectedInboundRTPStreamStats        InboundRTPStreamStats
+		expectedOutboundRTPStreamStats       OutboundRTPStreamStats
+		expectedRemoteInboundRTPStreamStats  RemoteInboundRTPStreamStats
+		expectedRemoteOutboundRTPStreamStats RemoteOutboundRTPStreamStats
+	}
+	now := time.Date(2022, time.July, 18, 0, 0, 0, 0, time.Local)
+	for i, cc := range []input{
+		{
+			name: "basicIncomingRTP",
+			records: []record{
+				{
+					ts: now,
+					content: incomingRTP{
+						header: rtp.Header{
+							SequenceNumber: 7,
+							Timestamp:      0,
+						},
+					},
+				},
+				{
+					ts: now.Add(1 * time.Second),
+					content: incomingRTP{
+						header: rtp.Header{
+							SequenceNumber: 10,
+							Timestamp:      90000,
+						},
+					},
+				},
+				{
+					ts: now.Add(2 * time.Second),
+					content: incomingRTP{
+						header: rtp.Header{
+							SequenceNumber: 11,
+							Timestamp:      2 * 90000,
+						},
+					},
+				},
+			},
+			expectedInboundRTPStreamStats: InboundRTPStreamStats{
+				ReceivedRTPStreamStats: ReceivedRTPStreamStats{
+					PacketsReceived: 3,
+					PacketsLost:     2,
+					Jitter:          90000 / 16,
+				},
+				LastPacketReceivedTimestamp: now.Add(2 * time.Second),
+				HeaderBytesReceived:         36,
+				BytesReceived:               36,
+			},
+		},
+		{
+			name: "basicOutgoingRTP",
+			records: []record{
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 1,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 3,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: incomingRTCP{
+						pkts: []rtcp.Packet{
+							&rtcp.ReceiverReport{
+								SSRC: 0,
+								Reports: []rtcp.ReceptionReport{
+									{
+										SSRC:               0,
+										FractionLost:       85,
+										TotalLost:          1,
+										LastSequenceNumber: 3,
+										Jitter:             45000,
+									},
+								},
+							},
+							cname,
+						},
+					},
+				},
+			},
+			expectedOutboundRTPStreamStats: OutboundRTPStreamStats{
+				SentRTPStreamStats: SentRTPStreamStats{
+					PacketsSent: 2,
+					BytesSent:   24,
+				},
+				HeaderBytesSent: 24,
+			},
+			expectedRemoteInboundRTPStreamStats: RemoteInboundRTPStreamStats{
+				ReceivedRTPStreamStats: ReceivedRTPStreamStats{
+					PacketsReceived: 2,
+					PacketsLost:     1,
+					Jitter:          0.5,
+				},
+				FractionLost: 0.33203125,
+			},
+		},
+		{
+			name: "basicOutgoingRTCP",
+			records: []record{
+				{
+					ts: now,
+					content: outgoingRTCP{
+						ts: now,
+						pkts: []rtcp.Packet{&rtcp.SenderReport{
+							NTPTime: ntp.ToNTP(now),
+						}},
+					},
+				},
+				{
+					ts: now.Add(2 * time.Second),
+					content: incomingRTCP{
+						pkts: []rtcp.Packet{
+							&rtcp.ReceiverReport{
+								SSRC: 0,
+								Reports: []rtcp.ReceptionReport{{
+									SSRC:             0,
+									LastSenderReport: uint32((ntp.ToNTP(now) & 0x0000FFFFFFFF0000) >> 16),
+									Delay:            1 * 65536.0,
+								}},
+							},
+							cname,
+						},
+					},
+				},
+			},
+			expectedRemoteInboundRTPStreamStats: RemoteInboundRTPStreamStats{
+				RoundTripTime:             time.Second,
+				TotalRoundTripTime:        time.Second,
+				RoundTripTimeMeasurements: 1,
+			},
+		},
+		{
+			name: "basicIncomingRTCP",
+			records: []record{
+				{
+					ts: now,
+					content: incomingRTCP{
+						pkts: []rtcp.Packet{
+							&rtcp.SenderReport{
+								NTPTime: ntp.ToNTP(now),
+							},
+							cname,
+						},
+					},
+				},
+			},
+
+			expectedRemoteOutboundRTPStreamStats: RemoteOutboundRTPStreamStats{
+				ReportsSent:     1,
+				RemoteTimeStamp: ntp.ToTime(ntp.ToNTP(now)),
+			},
+		},
+		{
+			name: "remoteOutboundRTT",
+			records: []record{
+				{
+					ts: now,
+					content: outgoingRTCP{
+						pkts: []rtcp.Packet{
+							&rtcp.ReceiverReport{},
+							&rtcp.ExtendedReport{
+								SenderSSRC: 0,
+								Reports: []rtcp.ReportBlock{
+									&rtcp.ReceiverReferenceTimeReportBlock{
+										NTPTimestamp: ntp.ToNTP(now),
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					ts: now.Add(2 * time.Second),
+					content: incomingRTCP{
+						pkts: []rtcp.Packet{
+							&rtcp.SenderReport{
+								NTPTime: ntp.ToNTP(now.Add(time.Second)),
+							},
+							cname,
+							&rtcp.ExtendedReport{
+								SenderSSRC: 0,
+								Reports: []rtcp.ReportBlock{
+									&rtcp.DLRRReportBlock{
+										Reports: []rtcp.DLRRReport{
+											{
+												SSRC:   0,
+												LastRR: uint32((ntp.ToNTP(now) >> 16) & 0xFFFFFFFF),
+												DLRR:   1 * 65536.0,
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedRemoteOutboundRTPStreamStats: RemoteOutboundRTPStreamStats{
+				RemoteTimeStamp:           now.Add(time.Second),
+				ReportsSent:               1,
+				RoundTripTime:             time.Second,
+				TotalRoundTripTime:        time.Second,
+				RoundTripTimeMeasurements: 1,
+			},
+		},
+	} {
+		t.Run(fmt.Sprintf("%v:%v", i, cc.name), func(t *testing.T) {
+			r := newRecorder(0, 90_000)
+
+			go r.Start()
+			defer r.Stop()
+
+			for _, record := range cc.records {
+				switch v := record.content.(type) {
+				case incomingRTP:
+					r.QueueIncomingRTP(record.ts, mustMarshalRTP(t, rtp.Packet{Header: v.header}), v.attr)
+				case incomingRTCP:
+					pkts := make(rtcp.CompoundPacket, len(v.pkts))
+					copy(pkts, v.pkts)
+					r.QueueIncomingRTCP(record.ts, mustMarshalRTCPs(t, &pkts), v.attr)
+				case outgoingRTP:
+					r.QueueOutgoingRTP(record.ts, &v.header, []byte{}, v.attr)
+				case outgoingRTCP:
+					r.QueueOutgoingRTCP(record.ts, v.pkts, v.attr)
+				default:
+					assert.FailNow(t, "invalid test case")
+				}
+			}
+
+			s := r.GetStats()
+
+			assert.Equal(t, cc.expectedInboundRTPStreamStats, s.InboundRTPStreamStats)
+			assert.Equal(t, cc.expectedOutboundRTPStreamStats, s.OutboundRTPStreamStats)
+			assert.Equal(t, cc.expectedRemoteInboundRTPStreamStats, s.RemoteInboundRTPStreamStats)
+			assert.Equal(t, cc.expectedRemoteOutboundRTPStreamStats, s.RemoteOutboundRTPStreamStats)
+		})
+	}
+}