Repository: pion/interceptor
Pull request #302 (2025-01-16 16:11:36)

Deduplicate unwrapper
Reuse unwrapper code for RFC8888 implementation.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pion/interceptor/pull/302?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 71.37%. Comparing base [(`e187410`)](https://app.codecov.io/gh/pion/interceptor/commit/e1874104865b23ba465ecc505f959daf156cc2a5?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) to head [(`2f7577f`)](https://app.codecov.io/gh/pion/interceptor/commit/2f7577f7511acd892c499a32bc70ef833ebbd1cf?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).

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


```diff
@@            Coverage Diff             @@
##           master     #302      +/-   ##
==========================================
+ Coverage   71.31%   71.37%   +0.06%     
==========================================
  Files          80       79       -1     
  Lines        4483     4461      -22     
==========================================
- Hits         3197     3184      -13     
+ Misses       1153     1147       -6     
+ Partials      133      130       -3     
```

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

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

</details>

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


Patch:
diff --git a/go.mod b/go.mod
index fe03158..a5a0579 100644
--- a/go.mod
+++ b/go.mod
@@ -5,7 +5,7 @@ go 1.20
 require (
 	github.com/pion/logging v0.2.2
 	github.com/pion/rtcp v1.2.15
-	github.com/pion/rtp v1.8.10
+	github.com/pion/rtp v1.8.11
 	github.com/pion/transport/v3 v3.0.7
 	github.com/stretchr/testify v1.10.0
 )
diff --git a/go.sum b/go.sum
index 0e87c34..413b877 100644
--- a/go.sum
+++ b/go.sum
@@ -6,8 +6,8 @@ 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.15 h1:LZQi2JbdipLOj4eBjK4wlVoQWfrZbh3Q6eHtWtJBZBo=
 github.com/pion/rtcp v1.2.15/go.mod h1:jlGuAjHMEXwMUHK78RgX0UmEJFV4zUKOFHR7OP+D3D0=
-github.com/pion/rtp v1.8.10 h1:puphjdbjPB+L+NFaVuZ5h6bt1g5q4kFIoI+r5q/g0CU=
-github.com/pion/rtp v1.8.10/go.mod h1:8uMBJj32Pa1wwx8Fuv/AsFhn8jsgw+3rUC2PfoBZ8p4=
+github.com/pion/rtp v1.8.11 h1:17xjnY5WO5hgO6SD3/NTIUPvSFw/PbLsIJyz1r1yNIk=
+github.com/pion/rtp v1.8.11/go.mod h1:8uMBJj32Pa1wwx8Fuv/AsFhn8jsgw+3rUC2PfoBZ8p4=
 github.com/pion/transport/v3 v3.0.7 h1:iRbMH05BzSNwhILHoBoAPxoB9xQgOaJk+591KC9P1o0=
 github.com/pion/transport/v3 v3.0.7/go.mod h1:YleKiTZ4vqNxVwh77Z0zytYi7rXHl7j6uPLGhhz9rwo=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
diff --git a/pkg/packetdump/filter.go b/pkg/packetdump/filter.go
index c20026b..627fb43 100644
--- a/pkg/packetdump/filter.go
+++ b/pkg/packetdump/filter.go
@@ -14,4 +14,9 @@ type RTPFilterCallback func(pkt *rtp.Packet) bool
 
 // RTCPFilterCallback can be used to filter RTCP packets to dump.
 // The callback returns whether or not to print dump the packet's content.
+// Deprecated: prefer RTCPPerPacketFilterCallback
 type RTCPFilterCallback func(pkt []rtcp.Packet) bool
+
+// RTCPPerPacketFilterCallback can be used to filter RTCP packets to dump.
+// It's called once per every packet opposing to RTCPFilterCallback which is called once per packet batch
+type RTCPPerPacketFilterCallback func(pkt rtcp.Packet) bool
diff --git a/pkg/packetdump/format.go b/pkg/packetdump/format.go
index 889e084..c570318 100644
--- a/pkg/packetdump/format.go
+++ b/pkg/packetdump/format.go
@@ -14,19 +14,29 @@ import (
 // RTPFormatCallback can be used to apply custom formatting to each dumped RTP
 // packet. If new lines should be added after each packet, they must be included
 // in the returned format.
+// Deprecated: prefer RTPBinaryFormatCallback
 type RTPFormatCallback func(*rtp.Packet, interceptor.Attributes) string
 
 // RTCPFormatCallback can be used to apply custom formatting to each dumped RTCP
 // packet. If new lines should be added after each packet, they must be included
 // in the returned format.
+// Deprecated: prefer RTCPBinaryFormatCallback
 type RTCPFormatCallback func([]rtcp.Packet, interceptor.Attributes) string
 
 // DefaultRTPFormatter returns the default log format for RTP packets
+// Deprecated: useless export since set by default
 func DefaultRTPFormatter(pkt *rtp.Packet, _ interceptor.Attributes) string {
 	return fmt.Sprintf("%s\n", pkt)
 }
 
 // DefaultRTCPFormatter returns the default log format for RTCP packets
+// Deprecated: useless export since set by default
 func DefaultRTCPFormatter(pkts []rtcp.Packet, _ interceptor.Attributes) string {
 	return fmt.Sprintf("%s\n", pkts)
 }
+
+// RTPBinaryFormatCallback can be used to apply custom formatting or marshaling to each dumped RTP packet
+type RTPBinaryFormatCallback func(*rtp.Packet, interceptor.Attributes) ([]byte, error)
+
+// RTCPBinaryFormatCallback can be used to apply custom formatting or marshaling to each dumped RTCP packet
+type RTCPBinaryFormatCallback func(rtcp.Packet, interceptor.Attributes) ([]byte, error)
diff --git a/pkg/packetdump/option.go b/pkg/packetdump/option.go
index 9be23ea..9b3c83f 100644
--- a/pkg/packetdump/option.go
+++ b/pkg/packetdump/option.go
@@ -37,6 +37,7 @@ func RTCPWriter(w io.Writer) PacketDumperOption {
 }
 
 // RTPFormatter sets the RTP format
+// Deprecated: prefer RTPBinaryFormatter
 func RTPFormatter(f RTPFormatCallback) PacketDumperOption {
 	return func(d *PacketDumper) error {
 		d.rtpFormat = f
@@ -45,6 +46,7 @@ func RTPFormatter(f RTPFormatCallback) PacketDumperOption {
 }
 
 // RTCPFormatter sets the RTCP format
+// Deprecated: prefer RTCPBinaryFormatter
 func RTCPFormatter(f RTCPFormatCallback) PacketDumperOption {
 	return func(d *PacketDumper) error {
 		d.rtcpFormat = f
@@ -52,6 +54,22 @@ func RTCPFormatter(f RTCPFormatCallback) PacketDumperOption {
 	}
 }
 
+// RTPBinaryFormatter sets the RTP binary formatter
+func RTPBinaryFormatter(f RTPBinaryFormatCallback) PacketDumperOption {
+	return func(d *PacketDumper) error {
+		d.rtpFormatBinary = f
+		return nil
+	}
+}
+
+// RTCPBinaryFormatter sets the RTCP binary formatter
+func RTCPBinaryFormatter(f RTCPBinaryFormatCallback) PacketDumperOption {
+	return func(d *PacketDumper) error {
+		d.rtcpFormatBinary = f
+		return nil
+	}
+}
+
 // RTPFilter sets the RTP filter.
 func RTPFilter(callback RTPFilterCallback) PacketDumperOption {
 	return func(d *PacketDumper) error {
@@ -61,9 +79,18 @@ func RTPFilter(callback RTPFilterCallback) PacketDumperOption {
 }
 
 // RTCPFilter sets the RTCP filter.
+// Deprecated: prefer RTCPPerPacketFilter
 func RTCPFilter(callback RTCPFilterCallback) PacketDumperOption {
 	return func(d *PacketDumper) error {
 		d.rtcpFilter = callback
 		return nil
 	}
 }
+
+// RTCPPerPacketFilter sets the RTCP per-packet filter.
+func RTCPPerPacketFilter(callback RTCPPerPacketFilterCallback) PacketDumperOption {
+	return func(d *PacketDumper) error {
+		d.rtcpPerPacketFilter = callback
+		return nil
+	}
+}
diff --git a/pkg/packetdump/packet_dumper.go b/pkg/packetdump/packet_dumper.go
index 12e0806..c1b1d78 100644
--- a/pkg/packetdump/packet_dumper.go
+++ b/pkg/packetdump/packet_dumper.go
@@ -15,6 +15,9 @@ import (
 	"github.com/pion/rtp"
 )
 
+// ErrBothBinaryAndDeprecatedFormat is returned when both binary and deprecated format callbacks are set
+var ErrBothBinaryAndDeprecatedFormat = fmt.Errorf("both binary and deprecated format callbacks are set")
+
 // PacketDumper dumps packet to a io.Writer
 type PacketDumper struct {
 	log logging.LeveledLogger
@@ -28,31 +31,44 @@ type PacketDumper struct {
 	rtpStream  io.Writer
 	rtcpStream io.Writer
 
+	rtpFormatBinary  RTPBinaryFormatCallback
+	rtcpFormatBinary RTCPBinaryFormatCallback
+
 	rtpFormat  RTPFormatCallback
 	rtcpFormat RTCPFormatCallback
 
-	rtpFilter  RTPFilterCallback
-	rtcpFilter RTCPFilterCallback
+	rtpFilter           RTPFilterCallback
+	rtcpFilter          RTCPFilterCallback
+	rtcpPerPacketFilter RTCPPerPacketFilterCallback
 }
 
 // NewPacketDumper creates a new PacketDumper
 func NewPacketDumper(opts ...PacketDumperOption) (*PacketDumper, error) {
 	d := &PacketDumper{
-		log:        logging.NewDefaultLoggerFactory().NewLogger("packet_dumper"),
-		wg:         sync.WaitGroup{},
-		close:      make(chan struct{}),
-		rtpChan:    make(chan *rtpDump),
-		rtcpChan:   make(chan *rtcpDump),
-		rtpStream:  os.Stdout,
-		rtcpStream: os.Stdout,
-		rtpFormat:  DefaultRTPFormatter,
-		rtcpFormat: DefaultRTCPFormatter,
+		log:              logging.NewDefaultLoggerFactory().NewLogger("packet_dumper"),
+		wg:               sync.WaitGroup{},
+		close:            make(chan struct{}),
+		rtpChan:          make(chan *rtpDump),
+		rtcpChan:         make(chan *rtcpDump),
+		rtpStream:        os.Stdout,
+		rtcpStream:       os.Stdout,
+		rtpFormat:        nil,
+		rtcpFormat:       nil,
+		rtpFormatBinary:  nil,
+		rtcpFormatBinary: nil,
 		rtpFilter: func(*rtp.Packet) bool {
 			return true
 		},
 		rtcpFilter: func([]rtcp.Packet) bool {
 			return true
 		},
+		rtcpPerPacketFilter: func(rtcp.Packet) bool {
+			return true
+		},
+	}
+
+	if d.rtpFormat != nil && d.rtpFormatBinary != nil {
+		return nil, ErrBothBinaryAndDeprecatedFormat
 	}
 
 	for _, opt := range opts {
@@ -61,6 +77,14 @@ func NewPacketDumper(opts ...PacketDumperOption) (*PacketDumper, error) {
 		}
 	}
 
+	if d.rtpFormat == nil && d.rtpFormatBinary == nil {
+		d.rtpFormat = DefaultRTPFormatter
+	}
+
+	if d.rtcpFormat == nil && d.rtcpFormatBinary == nil {
+		d.rtcpFormat = DefaultRTCPFormatter
+	}
+
 	d.wg.Add(1)
 	go d.loop()
 
@@ -117,17 +141,72 @@ func (d *PacketDumper) loop() {
 		case <-d.close:
 			return
 		case dump := <-d.rtpChan:
-			if d.rtpFilter(dump.packet) {
-				if _, err := fmt.Fprint(d.rtpStream, d.rtpFormat(dump.packet, dump.attributes)); err != nil {
-					d.log.Errorf("could not dump RTP packet %v", err)
-				}
+			err := d.writeDumpedRTP(dump)
+			if err != nil {
+				d.log.Errorf("could not dump RTP packet: %v", err)
 			}
 		case dump := <-d.rtcpChan:
-			if d.rtcpFilter(dump.packets) {
-				if _, err := fmt.Fprint(d.rtcpStream, d.rtcpFormat(dump.packets, dump.attributes)); err != nil {
-					d.log.Errorf("could not dump RTCP packet %v", err)
-				}
+			err := d.writeDumpedRTCP(dump)
+			if err != nil {
+				d.log.Errorf("could not dump RTCP packets: %v", err)
+			}
+		}
+	}
+}
+
+func (d *PacketDumper) writeDumpedRTP(dump *rtpDump) error {
+	if !d.rtpFilter(dump.packet) {
+		return nil
+	}
+
+	if d.rtpFormatBinary != nil {
+		dumped, err := d.rtpFormatBinary(dump.packet, dump.attributes)
+		if err != nil {
+			return fmt.Errorf("rtp format binary: %w", err)
+		}
+		_, err = d.rtpStream.Write(dumped)
+		if err != nil {
+			return fmt.Errorf("rtp stream write: %w", err)
+		}
+	}
+
+	if d.rtpFormat != nil {
+		if _, err := fmt.Fprint(d.rtpStream, d.rtpFormat(dump.packet, dump.attributes)); err != nil {
+			return fmt.Errorf("rtp stream Fprint: %w", err)
+		}
+	}
+
+	return nil
+}
+
+func (d *PacketDumper) writeDumpedRTCP(dump *rtcpDump) error {
+	if !d.rtcpFilter(dump.packets) {
+		return nil
+	}
+
+	for _, pkt := range dump.packets {
+		if !d.rtcpPerPacketFilter(pkt) {
+			continue
+		}
+
+		if d.rtcpFormatBinary != nil {
+			dumped, err := d.rtcpFormatBinary(pkt, dump.attributes)
+			if err != nil {
+				return fmt.Errorf("rtcp format binary: %w", err)
+			}
+
+			_, err = d.rtcpStream.Write(dumped)
+			if err != nil {
+				return fmt.Errorf("rtcp stream write: %w", err)
 			}
 		}
 	}
+
+	if d.rtcpFormat != nil {
+		if _, err := fmt.Fprint(d.rtcpStream, d.rtcpFormat(dump.packets, dump.attributes)); err != nil {
+			return fmt.Errorf("rtсp stream Fprint: %w", err)
+		}
+	}
+
+	return nil
 }
diff --git a/pkg/rfc8888/stream_log.go b/pkg/rfc8888/stream_log.go
index 8dfb14f..2d18d9c 100644
--- a/pkg/rfc8888/stream_log.go
+++ b/pkg/rfc8888/stream_log.go
@@ -6,6 +6,7 @@ package rfc8888
 import (
 	"time"
 
+	"github.com/pion/interceptor/internal/sequencenumber"
 	"github.com/pion/rtcp"
 )
 
@@ -13,7 +14,7 @@ const maxReportsPerReportBlock = 16384
 
 type streamLog struct {
 	ssrc                       uint32
-	sequence                   unwrapper
+	sequence                   sequencenumber.Unwrapper
 	init                       bool
 	nextSequenceNumberToReport int64 // next to report
 	lastSequenceNumberReceived int64 // highest received
@@ -23,7 +24,7 @@ type streamLog struct {
 func newStreamLog(ssrc uint32) *streamLog {
 	return &streamLog{
 		ssrc:                       ssrc,
-		sequence:                   unwrapper{},
+		sequence:                   sequencenumber.Unwrapper{},
 		init:                       false,
 		nextSequenceNumberToReport: 0,
 		lastSequenceNumberReceived: 0,
@@ -32,7 +33,7 @@ func newStreamLog(ssrc uint32) *streamLog {
 }
 
 func (l *streamLog) add(ts time.Time, sequenceNumber uint16, ecn uint8) {
-	unwrappedSequenceNumber := l.sequence.unwrap(sequenceNumber)
+	unwrappedSequenceNumber := l.sequence.Unwrap(sequenceNumber)
 	if !l.init {
 		l.init = true
 		l.nextSequenceNumberToReport = unwrappedSequenceNumber
diff --git a/pkg/rfc8888/unwrapper.go b/pkg/rfc8888/unwrapper.go
deleted file mode 100644
index f15f33e..0000000
--- a/pkg/rfc8888/unwrapper.go
+++ /dev/null
@@ -1,42 +0,0 @@
-// SPDX-FileCopyrightText: 2023 The Pion community <https://pion.ly>
-// SPDX-License-Identifier: MIT
-
-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/sequencenumber/unwrapper_test.go b/internal/sequencenumber/unwrapper_test.go
index e5129b2..47c4154 100644
--- a/internal/sequencenumber/unwrapper_test.go
+++ b/internal/sequencenumber/unwrapper_test.go
@@ -83,6 +83,14 @@ func TestUnwrapper(t *testing.T) {
 			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},
+		},
 		{
 			input: []uint16{
 				0, 32767, 32768, 32769, 32770,
diff --git a/pkg/packetdump/receiver_interceptor_test.go b/pkg/packetdump/receiver_interceptor_test.go
index 0402db1..737e659 100644
--- a/pkg/packetdump/receiver_interceptor_test.go
+++ b/pkg/packetdump/receiver_interceptor_test.go
@@ -108,3 +108,106 @@ func TestReceiverFilterNothing(t *testing.T) {
 
 	assert.NotZero(t, buf.Len())
 }
+
+func TestReceiverCustomBinaryFormatter(t *testing.T) {
+	rtpBuf := bytes.Buffer{}
+	rtcpBuf := bytes.Buffer{}
+
+	factory, err := NewReceiverInterceptor(
+		RTPWriter(&rtpBuf),
+		RTCPWriter(&rtcpBuf),
+		Log(logging.NewDefaultLoggerFactory().NewLogger("test")),
+		// custom binary formatter to dump only seqno mod 256
+		RTPBinaryFormatter(func(p *rtp.Packet, _ interceptor.Attributes) ([]byte, error) {
+			return []byte{byte(p.SequenceNumber)}, nil
+		}),
+		// custom binary formatter to dump only DestinationSSRCs mod 256
+		RTCPBinaryFormatter(func(p rtcp.Packet, _ interceptor.Attributes) ([]byte, error) {
+			b := make([]byte, 0)
+			for _, ssrc := range p.DestinationSSRC() {
+				b = append(b, byte(ssrc))
+			}
+			return b, nil
+		}),
+	)
+	assert.NoError(t, err)
+
+	i, err := factory.NewInterceptor("")
+	assert.NoError(t, err)
+
+	assert.EqualValues(t, 0, rtpBuf.Len())
+	assert.EqualValues(t, 0, rtcpBuf.Len())
+
+	stream := test.NewMockStream(&interceptor.StreamInfo{
+		SSRC:      123456,
+		ClockRate: 90000,
+	}, i)
+	defer func() {
+		assert.NoError(t, stream.Close())
+	}()
+
+	stream.ReceiveRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{
+		SenderSSRC: 123,
+		MediaSSRC:  45,
+	}})
+	stream.ReceiveRTP(&rtp.Packet{Header: rtp.Header{
+		SequenceNumber: uint16(123),
+	}})
+
+	// Give time for packets to be handled and stream written to.
+	time.Sleep(50 * time.Millisecond)
+
+	err = i.Close()
+	assert.NoError(t, err)
+
+	// check that there is custom formatter results in buffer
+	assert.Equal(t, []byte{123}, rtpBuf.Bytes())
+	assert.Equal(t, []byte{45}, rtcpBuf.Bytes())
+}
+
+func TestReceiverRTCPPerPacketFilter(t *testing.T) {
+	buf := bytes.Buffer{}
+
+	factory, err := NewReceiverInterceptor(
+		RTCPWriter(&buf),
+		Log(logging.NewDefaultLoggerFactory().NewLogger("test")),
+		RTCPPerPacketFilter(func(packet rtcp.Packet) bool {
+			_, isPli := packet.(*rtcp.PictureLossIndication)
+			return isPli
+		}),
+		RTCPBinaryFormatter(func(p rtcp.Packet, _ interceptor.Attributes) ([]byte, error) {
+			assert.IsType(t, &rtcp.PictureLossIndication{}, p)
+			return []byte{123}, nil
+		}),
+	)
+	assert.NoError(t, err)
+
+	i, err := factory.NewInterceptor("")
+	assert.NoError(t, err)
+
+	assert.EqualValues(t, 0, buf.Len())
+
+	stream := test.NewMockStream(&interceptor.StreamInfo{
+		SSRC:      123456,
+		ClockRate: 90000,
+	}, i)
+	defer func() {
+		assert.NoError(t, stream.Close())
+	}()
+
+	stream.ReceiveRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{
+		SenderSSRC: 123,
+		MediaSSRC:  456,
+	}})
+	stream.ReceiveRTCP([]rtcp.Packet{&rtcp.ReceiverReport{
+		SSRC: 789,
+	}})
+	// Give time for packets to be handled and stream written to.
+	time.Sleep(50 * time.Millisecond)
+
+	err = i.Close()
+	assert.NoError(t, err)
+
+	// Only single PictureLossIndication should have been written.
+	assert.Equal(t, []byte{123}, buf.Bytes())
+}
diff --git a/pkg/packetdump/sender_interceptor_test.go b/pkg/packetdump/sender_interceptor_test.go
index b005b55..f97bbef 100644
--- a/pkg/packetdump/sender_interceptor_test.go
+++ b/pkg/packetdump/sender_interceptor_test.go
@@ -114,3 +114,113 @@ func TestSenderFilterNothing(t *testing.T) {
 
 	assert.NotZero(t, buf.Len())
 }
+
+func TestSenderCustomBinaryFormatter(t *testing.T) {
+	rtpBuf := bytes.Buffer{}
+	rtcpBuf := bytes.Buffer{}
+
+	factory, err := NewSenderInterceptor(
+		RTPWriter(&rtpBuf),
+		RTCPWriter(&rtcpBuf),
+		Log(logging.NewDefaultLoggerFactory().NewLogger("test")),
+		// custom binary formatter to dump only seqno mod 256
+		RTPBinaryFormatter(func(p *rtp.Packet, _ interceptor.Attributes) ([]byte, error) {
+			return []byte{byte(p.SequenceNumber)}, nil
+		}),
+		// custom binary formatter to dump only DestinationSSRCs mod 256
+		RTCPBinaryFormatter(func(p rtcp.Packet, _ interceptor.Attributes) ([]byte, error) {
+			b := make([]byte, 0)
+			for _, ssrc := range p.DestinationSSRC() {
+				b = append(b, byte(ssrc))
+			}
+			return b, nil
+		}),
+	)
+	assert.NoError(t, err)
+
+	i, err := factory.NewInterceptor("")
+	assert.NoError(t, err)
+
+	assert.EqualValues(t, 0, rtpBuf.Len())
+	assert.EqualValues(t, 0, rtcpBuf.Len())
+
+	stream := test.NewMockStream(&interceptor.StreamInfo{
+		SSRC:      123456,
+		ClockRate: 90000,
+	}, i)
+	defer func() {
+		assert.NoError(t, stream.Close())
+	}()
+
+	err = stream.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{
+		SenderSSRC: 123,
+		MediaSSRC:  45,
+	}})
+	assert.NoError(t, err)
+
+	err = stream.WriteRTP(&rtp.Packet{Header: rtp.Header{
+		SequenceNumber: uint16(123),
+	}})
+	assert.NoError(t, err)
+
+	// Give time for packets to be handled and stream written to.
+	time.Sleep(50 * time.Millisecond)
+
+	err = i.Close()
+	assert.NoError(t, err)
+
+	// check that there is custom formatter results in buffer
+	assert.Equal(t, []byte{123}, rtpBuf.Bytes())
+	assert.Equal(t, []byte{45}, rtcpBuf.Bytes())
+}
+
+func TestSenderRTCPPerPacketFilter(t *testing.T) {
+	buf := bytes.Buffer{}
+
+	factory, err := NewSenderInterceptor(
+		RTCPWriter(&buf),
+		Log(logging.NewDefaultLoggerFactory().NewLogger("test")),
+		RTCPPerPacketFilter(func(packet rtcp.Packet) bool {
+			_, isPli := packet.(*rtcp.PictureLossIndication)
+			return isPli
+		}),
+		RTCPBinaryFormatter(func(p rtcp.Packet, _ interceptor.Attributes) ([]byte, error) {
+			assert.IsType(t, &rtcp.PictureLossIndication{}, p)
+			return []byte{123}, nil
+		}),
+	)
+	assert.NoError(t, err)
+
+	i, err := factory.NewInterceptor("")
+	assert.NoError(t, err)
+
+	assert.EqualValues(t, 0, buf.Len())
+
+	stream := test.NewMockStream(&interceptor.StreamInfo{
+		SSRC:      123456,
+		ClockRate: 90000,
+	}, i)
+	defer func() {
+		assert.NoError(t, stream.Close())
+	}()
+
+	err = stream.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{
+		SenderSSRC: 123,
+		MediaSSRC:  456,
+	}})
+	assert.NoError(t, err)
+
+	err = stream.WriteRTCP([]rtcp.Packet{&rtcp.ReceiverReport{
+		SSRC: 789,
+	}})
+	assert.NoError(t, err)
+
+	// Give time for packets to be handled and stream written to.
+	time.Sleep(50 * time.Millisecond)
+
+	err = i.Close()
+	assert.NoError(t, err)
+
+	// Only single PictureLossIndication should have been written.
+	assert.Equal(t, []byte{123}, buf.Bytes())
+}
diff --git a/pkg/rfc8888/unwrapper_test.go b/pkg/rfc8888/unwrapper_test.go
deleted file mode 100644
index 3e7de92..0000000
--- a/pkg/rfc8888/unwrapper_test.go
+++ /dev/null
@@ -1,57 +0,0 @@
-// SPDX-FileCopyrightText: 2023 The Pion community <https://pion.ly>
-// SPDX-License-Identifier: MIT
-
-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)
-		})
-	}
-}