JitterBuffer: Improve performance for SampleBuilder use

internal/test/mock_stream.go

@@ -92,7 +92,6 @@ func NewMockStream(info *interceptor.StreamInfo, i interceptor.Interceptor) *Moc
 		if !ok {
 			return 0, nil, io.EOF
 		}
-
 		marshaled, err := p.Marshal()
 		if err != nil {
 			return 0, nil, io.EOF

pkg/jitterbuffer/jitter_buffer.go

@@ -64,7 +64,7 @@
 // order, and allows removing in either sequence number order or via a
 // provided timestamp
 type JitterBuffer struct {
-	packets       *PriorityQueue
+	packets       *RBTree
 	minStartCount uint16
 	lastSequence  uint16
 	playoutHead   uint16
@@ -128,6 +128,12 @@
 	return jb.playoutHead
 }
 
+func (jb *JitterBuffer) Length() uint16 {
+	jb.mutex.Lock()
+	defer jb.mutex.Unlock()
+	return jb.packets.Length()
+}
+
 // 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) {
@@ -155,7 +161,7 @@
 	if jb.packets.Length() == 0 {
 		jb.emit(StartBuffering)
 	}
-	if jb.packets.Length() > 100 {
+	if jb.packets.Length() > 2*jb.minStartCount {
 		jb.stats.overflowCount++
 		jb.emit(BufferOverflow)
 	}
@@ -240,8 +246,6 @@
 // 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 nil, err

pkg/jitterbuffer/jitter_buffer_test.go

@@ -112,6 +112,7 @@ func TestJitterBuffer(t *testing.T) {
 		assert.Equal(jb.packets.Length(), uint16(100))
 		assert.Equal(jb.state, Emitting)
 		head, err := jb.PopAtTimestamp(uint32(513))
+		assert.NotNil(head)
 		assert.Equal(head.SequenceNumber, uint16(math.MaxUint16-32+1))
 		assert.Equal(err, nil)
 		head, err = jb.PopAtTimestamp(uint32(513))

pkg/jitterbuffer/option.go

@@ -17,3 +17,10 @@ func Log(log logging.LeveledLogger) ReceiverInterceptorOption {
 		return nil
 	}
 }
+
+func WithSkipMissingPackets() ReceiverInterceptorOption {
+	return func(d *ReceiverInterceptor) error {
+		d.skipMissingPackets = true
+		return nil
+	}
+}