Repository: pion/interceptor
Pull request #304 (2025-01-17 17:01:12)

Binary formatters in packetdump
#### Description

This PR adds new packet dumper options that let you use RTP/RTCP formatters returning `[]byte` instead of strings. This makes it easier to create dumpers that save RTP data in binary formats like PCAP. The formatter and filter APIs are also updated to handle one RTCP packet at a time, making them consistent with how RTP is handled.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pion/interceptor/pull/304?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) Report
Attention: Patch coverage is `68.25397%` with `20 lines` in your changes missing coverage. Please review.
> Project coverage is 69.32%. 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 [(`5a55af3`)](https://app.codecov.io/gh/pion/interceptor/commit/5a55af3eb17bd67784672c572d4635a4738d2734?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).
> Report is 1 commits behind head on master.

| [Files with missing lines](https://app.codecov.io/gh/pion/interceptor/pull/304?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | Patch % | Lines |
|---|---|---|
| [pkg/packetdump/packet\_dumper.go](https://app.codecov.io/gh/pion/interceptor/pull/304?src=pr&el=tree&filepath=pkg%2Fpacketdump%2Fpacket_dumper.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3BhY2tldGR1bXAvcGFja2V0X2R1bXBlci5nbw==) | 68.75% | [10 Missing and 5 partials :warning: ](https://app.codecov.io/gh/pion/interceptor/pull/304?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) |
| [pkg/packetdump/option.go](https://app.codecov.io/gh/pion/interceptor/pull/304?src=pr&el=tree&filepath=pkg%2Fpacketdump%2Foption.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3BhY2tldGR1bXAvb3B0aW9uLmdv) | 66.66% | [5 Missing :warning: ](https://app.codecov.io/gh/pion/interceptor/pull/304?src=pr&el=tree&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     #304      +/-   ##
==========================================
- Coverage   71.31%   69.32%   -2.00%     
==========================================
  Files          80       80              
  Lines        4483     4534      +51     
==========================================
- Hits         3197     3143      -54     
- Misses       1153     1242      +89     
- Partials      133      149      +16     
```

| [Flag](https://app.codecov.io/gh/pion/interceptor/pull/304/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/304/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | `?` | |
| [wasm](https://app.codecov.io/gh/pion/interceptor/pull/304/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | `69.32% <68.25%> (-0.08%)` | :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/304?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).

mengelbart: This looks generally good to me. The only concern I have is that setting both the deprecated and the binary formatters seems to print both results to the same output. I would be happy to remove string formatters completely since the interceptors package is still v0 and since the new API can do everything the old one could, it should be easy to upgrade.
aalekseevx: The code would be much simpler if we could break backward compatibility, but I'm not sure whether this is common practice in `pion/interceptor`. @joeturki, @Sean-Der, what do you think? I see a few usages in open source:

https://github.com/search?type=code&q=github.com%2Fpion%2Finterceptor%2Fpkg%2Fpacketdump

mengelbart: I think I broke backward compatibility before, but that was a while ago, so maybe @Sean-Der could confirm if this is still ok. About the open source usages: all except the ducksoup one are my projects or forks of my projects, and as far as I know, most of them are no longer used and maintained.
aalekseevx: @mengelbart, I’ve added validation to ensure that both formatters are not used simultaneously. Ready to remove the deprecated API if this decision is approved.
joeturki: If I got to vote, I'd vote against hard deprecating (removing) APIs unless we're doing a major version bump. Even if it this is uncommonly used API It sets a precedent, but I’m not 100% sure.
aalekseevx: @joeturki, We can merge the current version and remove the deprecated API in the next major release. Does that work for you?
mengelbart: Note that the package is still in v0, which, according to https://go.dev/doc/modules/version-numbers, indicates that it is in development and API changes are to be expected. Bumping to v1 would mean that we fix the API from then on (until the next major version).
joeturki: > @joeturki, We can merge the current version and remove the deprecated API in the next major release. Does that work for you?

@aalekseevx  This makes sense :) 

> Note that the package is still in v0, which, according to https://go.dev/doc/modules/version-numbers, indicates that it is in development and API changes are to be expected. Bumping to v1 would mean that we fix the API from then on (until the next major version).

@mengelbart  Maybe we keep deprecation tags, until we have a stable API then we can bump the version, and delete the APIs?


Would love to hear @Sean-Der thought on this, and maybe we can have a standard policy for all packages / repos?


mengelbart: Feel free to merge as is. We can remove the old API later.

Patch:
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
 }


Test patch:
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())
+}