Repository: pion/interceptor
Pull request #230 (2024-02-23 19:16:12)

JitterBuffer Enhancements: Allow setting a minimum length
#### Description
For the tests (and probably generally to start)  we need a way to set when the jitter buffer will begin emitting packets, having met it's minimum threshold.


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pion/interceptor/pull/230?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 `6.25000%` with `30 lines` in your changes are missing coverage. Please review.
> Project coverage is 77.10%. Comparing base [(`5574fda`)](https://app.codecov.io/gh/pion/interceptor/commit/5574fda4dd5cc44b2c2e51fbfe885d747c37f45f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) to head [(`9be703e`)](https://app.codecov.io/gh/pion/interceptor/pull/230?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).

| [Files](https://app.codecov.io/gh/pion/interceptor/pull/230?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | Patch % | Lines |
|---|---|---|
| [pkg/jitterbuffer/jitter\_buffer.go](https://app.codecov.io/gh/pion/interceptor/pull/230?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2ppdHRlcmJ1ZmZlci9qaXR0ZXJfYnVmZmVyLmdv) | 6.25% | [30 Missing :warning: ](https://app.codecov.io/gh/pion/interceptor/pull/230?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     #230      +/-   ##
==========================================
- Coverage   79.63%   77.10%   -2.54%     
==========================================
  Files          70       70              
  Lines        3791     3821      +30     
==========================================
- Hits         3019     2946      -73     
- Misses        637      732      +95     
- Partials      135      143       +8     
```

| [Flag](https://app.codecov.io/gh/pion/interceptor/pull/230/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/230/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/230/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | `77.10% <6.25%> (-0.54%)` | :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/230?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 b66d501..56cb6c9 100644
--- a/pkg/jitterbuffer/jitter_buffer.go
+++ b/pkg/jitterbuffer/jitter_buffer.go
@@ -65,14 +65,15 @@ type (
 // order, and allows removing in either sequence number order or via a
 // provided timestamp
 type JitterBuffer struct {
-	packets      *PriorityQueue
-	lastSequence uint16
-	playoutHead  uint16
-	playoutReady bool
-	state        State
-	stats        Stats
-	listeners    map[Event][]EventListener
-	mutex        sync.Mutex
+	packets       *PriorityQueue
+	minStartCount uint16
+	lastSequence  uint16
+	playoutHead   uint16
+	playoutReady  bool
+	state         State
+	stats         Stats
+	listeners     map[Event][]EventListener
+	mutex         sync.Mutex
 }
 
 // Stats Track interesting statistics for the life of this JitterBuffer
@@ -90,13 +91,21 @@ 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}, 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
 }
 
+// WithMinimumPacketCount will set the required number of packets to be received before
+// any attempt to pop a packet can succeed
+func WithMinimumPacketCount(count uint16) Option {
+	return func(jb *JitterBuffer) {
+		jb.minStartCount = count
+	}
+}
+
 // Listen will register an event listener
 // The jitter buffer may emit events correspnding, interested listerns should
 // look at Event for available events
@@ -142,7 +151,7 @@ func (jb *JitterBuffer) emit(event Event) {
 
 func (jb *JitterBuffer) updateState() {
 	// For now, we only look at the number of packets captured in the play buffer
-	if jb.packets.Length() >= 50 && jb.state == Buffering {
+	if jb.packets.Length() >= jb.minStartCount && jb.state == Buffering {
 		jb.state = Emitting
 		jb.playoutReady = true
 		jb.emit(BeginPlayback)
@@ -186,6 +195,36 @@ func (jb *JitterBuffer) Pop() (*rtp.Packet, error) {
 	return packet, nil
 }
 
+// PopAtSequence will pop an RTP packet from the jitter buffer at the specified Sequence
+func (jb *JitterBuffer) PopAtSequence(sq uint16) (*rtp.Packet, error) {
+	jb.mutex.Lock()
+	defer jb.mutex.Unlock()
+	if jb.state != Emitting {
+		return nil, ErrPopWhileBuffering
+	}
+	packet, err := jb.packets.PopAt(sq)
+	if err != nil {
+		jb.stats.underflowCount++
+		jb.emit(BufferUnderflow)
+		return (*rtp.Packet)(nil), err
+	}
+	jb.playoutHead = (jb.playoutHead + 1) % math.MaxUint16
+	jb.updateState()
+	return packet, nil
+}
+
+// PeekAtSequence will return an RTP packet from the jitter buffer at the specified Sequence
+// without removing it from the buffer
+func (jb *JitterBuffer) PeekAtSequence(sq uint16) (*rtp.Packet, error) {
+	jb.mutex.Lock()
+	defer jb.mutex.Unlock()
+	packet, err := jb.packets.Find(sq)
+	if err != nil {
+		return (*rtp.Packet)(nil), err
+	}
+	return packet, nil
+}
+
 // PopAtTimestamp pops an RTP packet from the jitter buffer with the provided timestamp
 // Call this method repeatedly to drain the buffer at the timestamp
 func (jb *JitterBuffer) PopAtTimestamp(ts uint32) (*rtp.Packet, error) {


Test patch:
diff --git a/pkg/jitterbuffer/jitter_buffer_test.go b/pkg/jitterbuffer/jitter_buffer_test.go
index 5ba9eaa..b326cd2 100644
--- a/pkg/jitterbuffer/jitter_buffer_test.go
+++ b/pkg/jitterbuffer/jitter_buffer_test.go
@@ -29,7 +29,6 @@ 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++ {
@@ -42,6 +41,18 @@ func TestJitterBuffer(t *testing.T) {
 		assert.Equal(head.SequenceNumber, uint16(5012))
 		assert.Equal(err, nil)
 	})
+	t.Run("Appends packets and begins playout", func(*testing.T) {
+		jb := New(WithMinimumPacketCount(1))
+		for i := 0; i < 2; i++ {
+			jb.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: uint16(5012 + i), Timestamp: uint32(512 + i)}, Payload: []byte{0x02}})
+		}
+		assert.Equal(jb.packets.Length(), uint16(2))
+		assert.Equal(jb.state, Emitting)
+		assert.Equal(jb.playoutHead, uint16(5012))
+		head, err := jb.Pop()
+		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++ {
@@ -99,6 +110,20 @@ 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++ {
+			sqnum := uint16((math.MaxUint16 - 32 + i) % math.MaxUint16)
+			jb.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: sqnum, Timestamp: uint32(512 + i)}, Payload: []byte{0x02}})
+		}
+		jb.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: 1019, Timestamp: uint32(9000)}, Payload: []byte{0x02}})
+		jb.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: 1020, Timestamp: uint32(9000)}, Payload: []byte{0x02}})
+		assert.Equal(jb.packets.Length(), uint16(52))
+		assert.Equal(jb.state, Emitting)
+		head, err := jb.PopAtSequence(uint16(9000))
+		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++ {
@@ -120,4 +145,25 @@ 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++ {
+			sqnum := uint16((math.MaxUint16 - 32 + i) % math.MaxUint16)
+			jb.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: sqnum, Timestamp: uint32(512 + i)}, Payload: []byte{0x02}})
+		}
+		jb.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: 1019, Timestamp: uint32(9000)}, Payload: []byte{0x02}})
+		jb.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: 1020, Timestamp: uint32(9000)}, Payload: []byte{0x02}})
+		assert.Equal(jb.packets.Length(), uint16(52))
+		assert.Equal(jb.state, Emitting)
+		head, err := jb.PeekAtSequence(uint16(1019))
+		assert.Equal(head.SequenceNumber, uint16(1019))
+		assert.Equal(err, nil)
+		head, err = jb.PeekAtSequence(uint16(1020))
+		assert.Equal(head.SequenceNumber, uint16(1020))
+		assert.Equal(err, nil)
+
+		head, err = jb.PopAtSequence(uint16(math.MaxUint16 - 32))
+		assert.Equal(head.SequenceNumber, uint16(math.MaxUint16-32))
+		assert.Equal(err, nil)
+	})
 }