Jitter Buffer: Enable setting a minimum length

Add additional helper method to set the minimum packet count
before the jitterbuffer will emit packets

pkg/jitterbuffer/jitter_buffer.go

@@ -66,6 +66,7 @@ type (
 // provided timestamp
 type JitterBuffer struct {
 	packets      *PriorityQueue
+  minStartCount uint16
 	lastSequence uint16
 	playoutHead  uint16
 	playoutReady bool
@@ -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,40 @@ 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()
+	if jb.state != Emitting {
+		return nil, ErrPopWhileBuffering
+	}
+	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) {

pkg/jitterbuffer/jitter_buffer_test.go

@@ -120,4 +120,26 @@ 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(t *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)
+	})
+
 }