Repository: pion/interceptor
Pull request #246 (2024-04-11 16:34:26)

Add SetPlayoutHead
Allow the user to Get+Set PlayoutHead of the JitterBuffer

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pion/interceptor/pull/246?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 70.10%. Comparing base [(`bf4e938`)](https://app.codecov.io/gh/pion/interceptor/commit/bf4e93840eae8f6aba4d0b7ba4172ed58d950565?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) to head [(`490d213`)](https://app.codecov.io/gh/pion/interceptor/pull/246?dropdown=coverage&src=pr&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     #246      +/-   ##
==========================================
- Coverage   71.92%   70.10%   -1.83%     
==========================================
  Files          77       77              
  Lines        3573     3586      +13     
==========================================
- Hits         2570     2514      -56     
- Misses        868      927      +59     
- Partials      135      145      +10     
```

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

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/246?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/pkg/jitterbuffer/jitter_buffer.go b/pkg/jitterbuffer/jitter_buffer.go
index 56cb6c9..e7a9f43 100644
--- a/pkg/jitterbuffer/jitter_buffer.go
+++ b/pkg/jitterbuffer/jitter_buffer.go
@@ -91,10 +91,18 @@ type Stats struct {
 
 // New will initialize a jitter buffer and its associated statistics
 func New(opts ...Option) *JitterBuffer {
-	jb := &JitterBuffer{state: Buffering, stats: Stats{0, 0, 0}, minStartCount: 50, packets: NewQueue(), listeners: make(map[Event][]EventListener)}
+	jb := &JitterBuffer{
+		state:         Buffering,
+		stats:         Stats{0, 0, 0},
+		minStartCount: 50,
+		packets:       NewQueue(),
+		listeners:     make(map[Event][]EventListener),
+	}
+
 	for _, o := range opts {
 		o(jb)
 	}
+
 	return jb
 }
 
@@ -113,6 +121,23 @@ func (jb *JitterBuffer) Listen(event Event, cb EventListener) {
 	jb.listeners[event] = append(jb.listeners[event], cb)
 }
 
+// PlayoutHead returns the SequenceNumber that will be attempted to Pop next
+func (jb *JitterBuffer) PlayoutHead() uint16 {
+	jb.mutex.Lock()
+	defer jb.mutex.Unlock()
+
+	return jb.playoutHead
+}
+
+// SetPlayoutHead allows you to manually specify the packet you wish to pop next
+// If you have encountered a packet that hasn't resolved you can skip it
+func (jb *JitterBuffer) SetPlayoutHead(playoutHead uint16) {
+	jb.mutex.Lock()
+	defer jb.mutex.Unlock()
+
+	jb.playoutHead = playoutHead
+}
+
 func (jb *JitterBuffer) updateStats(lastPktSeqNo uint16) {
 	// If we have at least one packet, and the next packet being pushed in is not
 	// at the expected sequence number increment the out of order count


Test patch:
diff --git a/pkg/jitterbuffer/jitter_buffer_test.go b/pkg/jitterbuffer/jitter_buffer_test.go
index b326cd2..0ed7302 100644
--- a/pkg/jitterbuffer/jitter_buffer_test.go
+++ b/pkg/jitterbuffer/jitter_buffer_test.go
@@ -13,6 +13,7 @@ import (
 
 func TestJitterBuffer(t *testing.T) {
 	assert := assert.New(t)
+
 	t.Run("Appends packets in order", func(*testing.T) {
 		jb := New()
 		assert.Equal(jb.lastSequence, uint16(0))
@@ -29,6 +30,7 @@ func TestJitterBuffer(t *testing.T) {
 		assert.Equal(jb.packets.Length(), uint16(4))
 		assert.Equal(jb.lastSequence, uint16(5012))
 	})
+
 	t.Run("Appends packets and begins playout", func(*testing.T) {
 		jb := New()
 		for i := 0; i < 100; i++ {
@@ -53,6 +55,7 @@ func TestJitterBuffer(t *testing.T) {
 		assert.Equal(head.SequenceNumber, uint16(5012))
 		assert.Equal(err, nil)
 	})
+
 	t.Run("Wraps playout correctly", func(*testing.T) {
 		jb := New()
 		for i := 0; i < 100; i++ {
@@ -75,6 +78,7 @@ func TestJitterBuffer(t *testing.T) {
 			}
 		}
 	})
+
 	t.Run("Pops at timestamp correctly", func(*testing.T) {
 		jb := New()
 		for i := 0; i < 100; i++ {
@@ -94,6 +98,7 @@ func TestJitterBuffer(t *testing.T) {
 		assert.Equal(head.SequenceNumber, uint16(math.MaxUint16-32))
 		assert.Equal(err, nil)
 	})
+
 	t.Run("Can peek at a packet", func(*testing.T) {
 		jb := New()
 		jb.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: 5000, Timestamp: 500}, Payload: []byte{0x02}})
@@ -110,6 +115,7 @@ func TestJitterBuffer(t *testing.T) {
 		assert.Equal(pkt.SequenceNumber, uint16(5000))
 		assert.Equal(err, nil)
 	})
+
 	t.Run("Pops at sequence with an invalid sequence number", func(*testing.T) {
 		jb := New()
 		for i := 0; i < 50; i++ {
@@ -124,6 +130,7 @@ func TestJitterBuffer(t *testing.T) {
 		assert.Equal(head, (*rtp.Packet)(nil))
 		assert.NotEqual(err, nil)
 	})
+
 	t.Run("Pops at timestamp with multiple packets", func(*testing.T) {
 		jb := New()
 		for i := 0; i < 50; i++ {
@@ -145,6 +152,7 @@ func TestJitterBuffer(t *testing.T) {
 		assert.Equal(head.SequenceNumber, uint16(math.MaxUint16-32))
 		assert.Equal(err, nil)
 	})
+
 	t.Run("Peeks at timestamp with multiple packets", func(*testing.T) {
 		jb := New()
 		for i := 0; i < 50; i++ {
@@ -166,4 +174,44 @@ func TestJitterBuffer(t *testing.T) {
 		assert.Equal(head.SequenceNumber, uint16(math.MaxUint16-32))
 		assert.Equal(err, nil)
 	})
+
+	t.Run("SetPlayoutHead", func(*testing.T) {
+		jb := New(WithMinimumPacketCount(1))
+
+		// Push packets 0-9, but no packet 4
+		for i := uint16(0); i < 10; i++ {
+			if i == 4 {
+				continue
+			}
+			jb.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: i, Timestamp: uint32(512 + i)}, Payload: []byte{0x00}})
+		}
+
+		// The first 3 packets will be able to popped
+		for i := 0; i < 4; i++ {
+			pkt, err := jb.Pop()
+			assert.NoError(err)
+			assert.NotNil(pkt)
+		}
+
+		// The next pop will fail because of gap
+		pkt, err := jb.Pop()
+		assert.ErrorIs(err, ErrNotFound)
+		assert.Nil(pkt)
+		assert.Equal(jb.PlayoutHead(), uint16(4))
+
+		// Assert that PlayoutHead isn't modified with pushing/popping again
+		jb.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: 10, Timestamp: uint32(522)}, Payload: []byte{0x00}})
+		pkt, err = jb.Pop()
+		assert.ErrorIs(err, ErrNotFound)
+		assert.Nil(pkt)
+		assert.Equal(jb.PlayoutHead(), uint16(4))
+
+		// Increment the PlayoutHead and popping will work again
+		jb.SetPlayoutHead(jb.PlayoutHead() + 1)
+		for i := 0; i < 6; i++ {
+			pkt, err := jb.Pop()
+			assert.NoError(err)
+			assert.NotNil(pkt)
+		}
+	})
 }