Refactored samplebuilder logic

Many corner cases would cause samplebuilder to fail and return invalid
results. This refactoring is more reliable in all cases.

Fixed bug in H264 writer by reusing the packet object in  H264 writer.

README.md

@@ -275,6 +275,7 @@ Check out the **[contributing wiki](https://github.com/pion/webrtc/wiki/Contribu
 * [Jin Gong](https://github.com/cgojin)
 * [yusuke](https://github.com/yusukem99)
 * [Patryk Rogalski](https://github.com/digitalix)
+* [Robin Raymond](https://github.com/robin-raymond)
 
 ### License
 MIT License - see [LICENSE](LICENSE) for full text

examples/custom-logger/main.go

@@ -14,8 +14,7 @@ import (
 // customLogger satisfies the interface logging.LeveledLogger
 // a logger is created per subsystem in Pion, so you can have custom
 // behavior per subsystem (ICE, DTLS, SCTP...)
-type customLogger struct {
+type customLogger struct{}
-}
 
 // Print all messages except trace
 func (c customLogger) Trace(msg string)                          {}
@@ -41,8 +40,7 @@ func (c customLogger) Errorf(format string, args ...interface{}) {
 // customLoggerFactory satisfies the interface logging.LoggerFactory
 // This allows us to create different loggers per subsystem. So we can
 // add custom behavior
-type customLoggerFactory struct {
+type customLoggerFactory struct{}
-}
 
 func (c customLoggerFactory) NewLogger(subsystem string) logging.LeveledLogger {
 	fmt.Printf("Creating logger for %s \n", subsystem)

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/pion/logging v0.2.2
 	github.com/pion/randutil v0.1.0
 	github.com/pion/rtcp v1.2.6
-	github.com/pion/rtp v1.6.2
+	github.com/pion/rtp v1.6.5
 	github.com/pion/sctp v1.7.12
 	github.com/pion/sdp/v3 v3.0.4
 	github.com/pion/srtp/v2 v2.0.2

go.sum

@@ -53,8 +53,9 @@ github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA=
 github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8=
 github.com/pion/rtcp v1.2.6 h1:1zvwBbyd0TeEuuWftrd/4d++m+/kZSeiguxU61LFWpo=
 github.com/pion/rtcp v1.2.6/go.mod h1:52rMNPWFsjr39z9B9MhnkqhPLoeHTv1aN63o/42bWE0=
-github.com/pion/rtp v1.6.2 h1:iGBerLX6JiDjB9NXuaPzHyxHFG9JsIEdgwTC0lp5n/U=
 github.com/pion/rtp v1.6.2/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko=
+github.com/pion/rtp v1.6.5 h1:o2cZf8OascA5HF/b0PAbTxRKvOWxTQxWYt7SlToxFGI=
+github.com/pion/rtp v1.6.5/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko=
 github.com/pion/sctp v1.7.10/go.mod h1:EhpTUQu1/lcK3xI+eriS6/96fWetHGCvBi9MSsnaBN0=
 github.com/pion/sctp v1.7.12 h1:GsatLufywVruXbZZT1CKg+Jr8ZTkwiPnmUC/oO9+uuY=
 github.com/pion/sctp v1.7.12/go.mod h1:xFe9cLMZ5Vj6eOzpyiKjT9SwGM4KpK/8Jbw5//jc+0s=

pkg/media/h264writer/h264writer.go

@@ -18,8 +18,9 @@ type (
 	// Therefore, only 1-23, 24 (STAP-A), 28 (FU-A) NAL types are allowed.
 	// https://tools.ietf.org/html/rfc6184#section-5.2
 	H264Writer struct {
-		writer      io.Writer
+		writer       io.Writer
-		hasKeyFrame bool
+		hasKeyFrame  bool
+		cachedPacket *codecs.H264Packet
 	}
 )
 
@@ -53,7 +54,11 @@ func (h *H264Writer) WriteRTP(packet *rtp.Packet) error {
 		}
 	}
 
-	data, err := (&codecs.H264Packet{}).Unmarshal(packet.Payload)
+	if h.cachedPacket == nil {
+		h.cachedPacket = &codecs.H264Packet{}
+	}
+
+	data, err := h.cachedPacket.Unmarshal(packet.Payload)
 	if err != nil {
 		return err
 	}
@@ -65,6 +70,7 @@ func (h *H264Writer) WriteRTP(packet *rtp.Packet) error {
 
 // Close closes the underlying writer
 func (h *H264Writer) Close() error {
+	h.cachedPacket = nil
 	if h.writer != nil {
 		if closer, ok := h.writer.(io.Closer); ok {
 			return closer.Close()

pkg/media/h264writer/h264writer_test.go

@@ -59,6 +59,7 @@ func TestWriteRTP(t *testing.T) {
 		hasKeyFrame bool
 		wantBytes   []byte
 		wantErr     error
+		reuseWriter bool
 	}{
 		{
 			"When given an empty payload; it should return nil",
@@ -66,6 +67,7 @@ func TestWriteRTP(t *testing.T) {
 			false,
 			[]byte{},
 			nil,
+			false,
 		},
 		{
 			"When no keyframe is defined; it should discard the packet",
@@ -73,6 +75,7 @@ func TestWriteRTP(t *testing.T) {
 			false,
 			[]byte{},
 			nil,
+			false,
 		},
 		{
 			"When a valid Single NAL Unit packet is given; it should unpack it without error",
@@ -80,6 +83,7 @@ func TestWriteRTP(t *testing.T) {
 			true,
 			[]byte{0x00, 0x00, 0x00, 0x01, 0x27, 0x90, 0x90},
 			nil,
+			false,
 		},
 		{
 			"When a valid STAP-A packet is given; it should unpack it without error",
@@ -87,23 +91,29 @@ func TestWriteRTP(t *testing.T) {
 			true,
 			[]byte{0x00, 0x00, 0x00, 0x01, 0x27, 0x90, 0x90, 0x00, 0x00, 0x00, 0x01, 0x28, 0x90, 0x90, 0x90, 0x90},
 			nil,
+			false,
 		},
 		{
 			"When a valid FU-A start packet is given; it should unpack it without error",
 			[]byte{0x3C, 0x85, 0x90, 0x90, 0x90},
 			true,
-			[]byte{0x00, 0x00, 0x00, 0x01, 0x25, 0x90, 0x90, 0x90},
+			[]byte{},
 			nil,
+			true,
 		},
 		{
 			"When a valid FU-A end packet is given; it should unpack it without error",
 			[]byte{0x3C, 0x45, 0x90, 0x90, 0x90},
 			true,
-			[]byte{0x90, 0x90, 0x90},
+			[]byte{0x00, 0x00, 0x00, 0x01, 0x25, 0x90, 0x90, 0x90, 0x90, 0x90, 0x90},
 			nil,
+			false,
 		},
 	}
 
+	var reuseWriter *bytes.Buffer
+	var reuseH264Writer *H264Writer
+
 	for _, tt := range tests {
 		tt := tt
 		t.Run(tt.name, func(t *testing.T) {
@@ -112,6 +122,12 @@ func TestWriteRTP(t *testing.T) {
 				hasKeyFrame: tt.hasKeyFrame,
 				writer:      writer,
 			}
+			if reuseWriter != nil {
+				writer = reuseWriter
+			}
+			if reuseH264Writer != nil {
+				h264Writer = reuseH264Writer
+			}
 			packet := &rtp.Packet{
 				Payload: tt.payload,
 			}
@@ -120,7 +136,14 @@ func TestWriteRTP(t *testing.T) {
 
 			assert.Equal(t, tt.wantErr, err)
 			assert.True(t, bytes.Equal(tt.wantBytes, writer.Bytes()))
-			assert.Nil(t, h264Writer.Close())
+			if !tt.reuseWriter {
+				assert.Nil(t, h264Writer.Close())
+				reuseWriter = nil
+				reuseH264Writer = nil
+			} else {
+				reuseWriter = writer
+				reuseH264Writer = h264Writer
+			}
 		})
 	}
 }

pkg/media/media.go

@@ -9,9 +9,11 @@ import (
 
 // A Sample contains encoded media and timing information
 type Sample struct {
-	Data      []byte
+	Data               []byte
-	Timestamp time.Time
+	Timestamp          time.Time
-	Duration  time.Duration
+	Duration           time.Duration
+	PacketTimestamp    uint32
+	PrevDroppedPackets uint16
 }
 
 // Writer defines an interface to handle

pkg/media/samplebuilder/sampleSequenceLocation.go

@@ -0,0 +1,82 @@
+// Package samplebuilder provides functionality to reconstruct media frames from RTP packets.
+package samplebuilder
+
+import "math"
+
+type sampleSequenceLocation struct {
+	// head is the first packet in a sequence
+	head uint16
+	// tail is always set to one after the final sequence number,
+	// so if head == tail then the sequence is empty
+	tail uint16
+}
+
+func (l sampleSequenceLocation) empty() bool {
+	return l.head == l.tail
+}
+
+func (l sampleSequenceLocation) hasData() bool {
+	return l.head != l.tail
+}
+
+func (l sampleSequenceLocation) count() uint16 {
+	return seqnumDistance(l.head, l.tail)
+}
+
+const (
+	slCompareVoid = iota
+	slCompareBefore
+	slCompareInside
+	slCompareAfter
+)
+
+func minUint32(x, y uint32) uint32 {
+	if x < y {
+		return x
+	}
+	return y
+}
+
+// Distance between two seqnums
+func seqnumDistance32(x, y uint32) uint32 {
+	diff := int32(x - y)
+	if diff < 0 {
+		return uint32(-diff)
+	}
+
+	return uint32(diff)
+}
+
+func (l sampleSequenceLocation) compare(pos uint16) int {
+	if l.empty() {
+		return slCompareVoid
+	}
+
+	head32 := uint32(l.head)
+	count32 := uint32(l.count())
+	tail32 := head32 + count32
+
+	// pos32 is possibly two values, the normal value or a wrap
+	// around the start value, figure out which it is...
+
+	pos32Normal := uint32(pos)
+	pos32Wrap := uint32(pos) + math.MaxUint16 + 1
+
+	distNormal := minUint32(seqnumDistance32(head32, pos32Normal), seqnumDistance32(tail32, pos32Normal))
+	distWrap := minUint32(seqnumDistance32(head32, pos32Wrap), seqnumDistance32(tail32, pos32Wrap))
+
+	pos32 := pos32Normal
+	if distWrap < distNormal {
+		pos32 = pos32Wrap
+	}
+
+	if pos32 < head32 {
+		return slCompareBefore
+	}
+
+	if pos32 >= tail32 {
+		return slCompareAfter
+	}
+
+	return slCompareInside
+}

pkg/media/samplebuilder/samplebuilder.go

@@ -2,6 +2,7 @@
 package samplebuilder
 
 import (
+	"math"
 	"time"
 
 	"github.com/pion/rtp"
@@ -10,8 +11,10 @@ import (
 
 // SampleBuilder buffers packets until media frames are complete.
 type SampleBuilder struct {
-	maxLate uint16 // how many packets to wait until we get a valid Sample
+	maxLate          uint16 // how many packets to wait until we get a valid Sample
-	buffer  [65536]*rtp.Packet
+	maxLateTimestamp uint32 // max timestamp between old and new timestamps before dropping packets
+	buffer           [math.MaxUint16 + 1]*rtp.Packet
+	preparedSamples  [math.MaxUint16 + 1]*media.Sample
 
 	// Interface that allows us to take RTP packets to samples
 	depacketizer rtp.Depacketizer
@@ -19,22 +22,24 @@ type SampleBuilder struct {
 	// sampleRate allows us to compute duration of media.SamplecA
 	sampleRate uint32
 
-	// Last seqnum that has been added to buffer
-	lastPush uint16
-
-	// Last seqnum that has been successfully popped
-	// isContiguous is false when we start or when we have a gap
-	// that is older then maxLate
-	isContiguous     bool
-	lastPopSeq       uint16
-	lastPopTimestamp uint32
-
 	// Interface that checks whether the packet is the first fragment of the frame or not
 	partitionHeadChecker rtp.PartitionHeadChecker
 
 	// the handler to be called when the builder is about to remove the
 	// reference to some packet.
 	packetReleaseHandler func(*rtp.Packet)
+
+	// filled contains the head/tail of the packets inserted into the buffer
+	filled sampleSequenceLocation
+
+	// active contains the active head/tail of the timestamp being actively processed
+	active sampleSequenceLocation
+
+	// prepared contains the samples that have been processed to date
+	prepared sampleSequenceLocation
+
+	// number of packets forced to be dropped
+	droppedPackets uint16
 }
 
 // New constructs a new SampleBuilder.
@@ -51,6 +56,51 @@ func New(maxLate uint16, depacketizer rtp.Depacketizer, sampleRate uint32, opts
 	return s
 }
 
+func (s *SampleBuilder) tooOld(location sampleSequenceLocation) bool {
+	if s.maxLateTimestamp == 0 {
+		return false
+	}
+
+	var foundHead *rtp.Packet
+	var foundTail *rtp.Packet
+
+	for i := location.head; i != location.tail; i++ {
+		if packet := s.buffer[i]; packet != nil {
+			foundHead = packet
+			break
+		}
+	}
+
+	if foundHead == nil {
+		return false
+	}
+
+	for i := location.tail - 1; i != location.head; i-- {
+		if packet := s.buffer[i]; packet != nil {
+			foundTail = packet
+			break
+		}
+	}
+
+	if foundTail == nil {
+		return false
+	}
+
+	return timestampDistance(foundHead.Timestamp, foundTail.Timestamp) > s.maxLateTimestamp
+}
+
+// fetchTimestamp returns the timestamp associated with a given sample location
+func (s *SampleBuilder) fetchTimestamp(location sampleSequenceLocation) (timestamp uint32, hasData bool) {
+	if location.empty() {
+		return 0, false
+	}
+	packet := s.buffer[location.head]
+	if packet == nil {
+		return 0, false
+	}
+	return packet.Timestamp, true
+}
+
 func (s *SampleBuilder) releasePacket(i uint16) {
 	var p *rtp.Packet
 	p, s.buffer[i] = s.buffer[i], nil
@@ -59,6 +109,43 @@ func (s *SampleBuilder) releasePacket(i uint16) {
 	}
 }
 
+// purgeConsumedBuffers clears all buffers that have already been consumed by
+// popping.
+func (s *SampleBuilder) purgeConsumedBuffers() {
+	for s.active.compare(s.filled.head) == slCompareBefore && s.filled.hasData() {
+		s.releasePacket(s.filled.head)
+		s.filled.head++
+	}
+}
+
+// purgeBuffers flushes all buffers that are already consumed or those buffers
+// that are too late to consume.
+func (s *SampleBuilder) purgeBuffers() {
+	s.purgeConsumedBuffers()
+
+	for (s.tooOld(s.filled) || (s.filled.count() > s.maxLate)) && s.filled.hasData() {
+		if s.active.empty() {
+			// refill the active based on the filled packets
+			s.active = s.filled
+		}
+
+		if s.active.hasData() && (s.active.head == s.filled.head) {
+			// attempt to force the active packet to be consumed even though
+			// outstanding data may be pending arrival
+			if s.buildSample(true) != nil {
+				continue
+			}
+
+			// could not build the sample so drop it
+			s.active.head++
+			s.droppedPackets++
+		}
+
+		s.releasePacket(s.filled.head)
+		s.filled.head++
+	}
+}
+
 // Push adds an RTP Packet to s's buffer.
 //
 // Push does not copy the input. If you wish to reuse
@@ -66,58 +153,139 @@ func (s *SampleBuilder) releasePacket(i uint16) {
 func (s *SampleBuilder) Push(p *rtp.Packet) {
 	s.buffer[p.SequenceNumber] = p
 
-	// Remove outdated references if SequenceNumber is increased.
+	switch s.filled.compare(p.SequenceNumber) {
-	if int16(p.SequenceNumber-s.lastPush) > 0 {
+	case slCompareVoid:
-		for i := s.lastPush; i != p.SequenceNumber+1; i++ {
+		s.filled.head = p.SequenceNumber
-			s.releasePacket(i - s.maxLate)
+		s.filled.tail = p.SequenceNumber + 1
-		}
+	case slCompareBefore:
+		s.filled.head = p.SequenceNumber
+	case slCompareAfter:
+		s.filled.tail = p.SequenceNumber + 1
+	case slCompareInside:
+		break
 	}
-
+	s.purgeBuffers()
-	s.lastPush = p.SequenceNumber
 }
 
 const secondToNanoseconds = 1000000000
 
-// We have a valid collection of RTP Packets
+// buildSample creates a sample from a valid collection of RTP Packets by
-// walk forwards building a sample if everything looks good clear and update buffer+values
+// walking forwards building a sample if everything looks good clear and
-func (s *SampleBuilder) buildSample(firstBuffer uint16) (*media.Sample, uint32) {
+// update buffer+values
+func (s *SampleBuilder) buildSample(purgingBuffers bool) *media.Sample {
+	if s.active.empty() {
+		s.active = s.filled
+	}
+
+	if s.active.empty() {
+		return nil
+	}
+
+	if s.filled.compare(s.active.tail) == slCompareInside {
+		s.active.tail = s.filled.tail
+	}
+
+	var consume sampleSequenceLocation
+
+	for i := s.active.head; s.buffer[i] != nil && i < s.active.tail; i++ {
+		if s.depacketizer.IsDetectedFinalPacketInSequence(s.buffer[i].Marker) {
+			consume.head = s.active.head
+			consume.tail = i + 1
+			break
+		}
+		headTimestamp, hasData := s.fetchTimestamp(s.active)
+		if hasData && s.buffer[i].Timestamp != headTimestamp {
+			consume.head = s.active.head
+			consume.tail = i
+			break
+		}
+	}
+
+	if consume.empty() {
+		return nil
+	}
+
+	if !purgingBuffers && s.buffer[consume.tail] == nil {
+		// wait for the next packet after this set of packets to arrive
+		// to ensure at least one post sample timestamp is known
+		// (unless we have to release right now)
+		return nil
+	}
+
+	sampleTimestamp, _ := s.fetchTimestamp(s.active)
+	afterTimestamp := sampleTimestamp
+
+	// scan for any packet after the current and use that time stamp as the diff point
+	for i := consume.tail; i < s.active.tail; i++ {
+		if s.buffer[i] != nil {
+			afterTimestamp = s.buffer[i].Timestamp
+			break
+		}
+	}
+
+	// the head set of packets is now fully consumed
+	s.active.head = consume.tail
+
+	// prior to decoding all the packets, check if this packet
+	// would end being disposed anyway
+	if s.partitionHeadChecker != nil {
+		if !s.partitionHeadChecker.IsPartitionHead(s.buffer[consume.head].Payload) {
+			s.droppedPackets += consume.count()
+			s.purgeConsumedBuffers()
+			return nil
+		}
+	}
+
+	// merge all the buffers into a sample
 	data := []byte{}
-
+	for ; consume.head != consume.tail; consume.head++ {
-	for i := firstBuffer; s.buffer[i] != nil; i++ {
+		p, err := s.depacketizer.Unmarshal(s.buffer[consume.head].Payload)
-		if s.buffer[i].Timestamp != s.buffer[firstBuffer].Timestamp {
-			lastTimeStamp := s.lastPopTimestamp
-			if !s.isContiguous {
-				if s.buffer[firstBuffer-1] != nil {
-					lastTimeStamp = s.buffer[firstBuffer-1].Timestamp
-				} else {
-					// If PartitionHeadChecker detects that the first packet is a head,
-					// the duration of the packet is not guessable
-					lastTimeStamp = s.buffer[firstBuffer].Timestamp
-				}
-			}
-
-			samples := s.buffer[i-1].Timestamp - lastTimeStamp
-			s.lastPopSeq = i - 1
-			s.isContiguous = true
-			s.lastPopTimestamp = s.buffer[i-1].Timestamp
-			for j := firstBuffer; j < i; j++ {
-				s.releasePacket(j)
-			}
-
-			return &media.Sample{Data: data, Duration: time.Duration((float64(samples)/float64(s.sampleRate))*secondToNanoseconds) * time.Nanosecond}, s.lastPopTimestamp
-		}
-
-		p, err := s.depacketizer.Unmarshal(s.buffer[i].Payload)
 		if err != nil {
-			return nil, 0
+			return nil
 		}
-
 		data = append(data, p...)
 	}
-	return nil, 0
+	samples := afterTimestamp - sampleTimestamp
+
+	sample := &media.Sample{
+		Data:               data,
+		Duration:           time.Duration((float64(samples)/float64(s.sampleRate))*secondToNanoseconds) * time.Nanosecond,
+		PacketTimestamp:    sampleTimestamp,
+		PrevDroppedPackets: s.droppedPackets,
+	}
+
+	s.droppedPackets = 0
+
+	s.preparedSamples[s.prepared.tail] = sample
+	s.prepared.tail++
+
+	s.purgeConsumedBuffers()
+
+	return sample
 }
 
-// Distance between two seqnums
+// Pop compiles pushed RTP packets into media samples and then
+// returns the next valid sample (or nil if no sample is compiled).
+func (s *SampleBuilder) Pop() *media.Sample {
+	_ = s.buildSample(false)
+	if s.prepared.empty() {
+		return nil
+	}
+	var result *media.Sample
+	result, s.preparedSamples[s.prepared.head] = s.preparedSamples[s.prepared.head], nil
+	s.prepared.head++
+	return result
+}
+
+// PopWithTimestamp compiles pushed RTP packets into media samples and then
+// returns the next valid sample with its associated RTP timestamp (or nil, 0 if
+// no sample is compiled).
+func (s *SampleBuilder) PopWithTimestamp() (*media.Sample, uint32) {
+	sample := s.Pop()
+	return sample, sample.PacketTimestamp
+}
+
+// seqnumDistance computes the distance between two sequence numbers
 func seqnumDistance(x, y uint16) uint16 {
 	diff := int16(x - y)
 	if diff < 0 {
@@ -127,53 +295,14 @@ func seqnumDistance(x, y uint16) uint16 {
 	return uint16(diff)
 }
 
-// Pop scans s's buffer for a valid sample.
+// timestampDistance computes the distance between two timestamps
-// It returns nil if no valid samples have been found.
+func timestampDistance(x, y uint32) uint32 {
-func (s *SampleBuilder) Pop() *media.Sample {
+	diff := int32(x - y)
-	sample, _ := s.PopWithTimestamp()
+	if diff < 0 {
-	return sample
+		return uint32(-diff)
-}
-
-// PopWithTimestamp scans s's buffer for a valid sample and its RTP timestamp.
-// It returns nil, 0 when no valid samples have been found.
-func (s *SampleBuilder) PopWithTimestamp() (*media.Sample, uint32) {
-	var i uint16
-	if !s.isContiguous {
-		i = s.lastPush - s.maxLate
-	} else {
-		if seqnumDistance(s.lastPopSeq, s.lastPush) > s.maxLate {
-			i = s.lastPush - s.maxLate
-			s.isContiguous = false
-		} else {
-			i = s.lastPopSeq + 1
-		}
 	}
 
-	for ; i != s.lastPush; i++ {
+	return uint32(diff)
-		curr := s.buffer[i]
-		if curr == nil {
-			continue // we haven't hit a buffer yet, keep moving
-		}
-
-		if !s.isContiguous {
-			if s.partitionHeadChecker == nil {
-				if s.buffer[i-1] == nil {
-					continue // We have never popped a buffer, so we can't assert that the first RTP packet we encounter is valid
-				} else if s.buffer[i-1].Timestamp == curr.Timestamp {
-					continue // We have the same timestamps, so it is data that spans multiple RTP packets
-				}
-			} else {
-				if !s.partitionHeadChecker.IsPartitionHead(curr.Payload) {
-					continue
-				}
-				// We can start using this frame as it is a head of frame partition
-			}
-		}
-
-		// Initial validity checks have passed, walk forward
-		return s.buildSample(i)
-	}
-	return nil, 0
 }
 
 // An Option configures a SampleBuilder.
@@ -194,3 +323,12 @@ func WithPacketReleaseHandler(h func(*rtp.Packet)) Option {
 		o.packetReleaseHandler = h
 	}
 }
+
+// WithMaxTimeDelay ensures that packets that are too old in the buffer get
+// purged based on time rather than building up an extraordinarily long delay.
+func WithMaxTimeDelay(maxLateDuration time.Duration) Option {
+	return func(o *SampleBuilder) {
+		totalMillis := maxLateDuration.Milliseconds()
+		o.maxLateTimestamp = uint32(int64(o.sampleRate) * totalMillis / 1000)
+	}
+}

pkg/media/samplebuilder/samplebuilder_test.go

@@ -11,22 +11,25 @@ import (
 )
 
 type sampleBuilderTest struct {
-	message         string
+	message          string
-	packets         []*rtp.Packet
+	packets          []*rtp.Packet
-	withHeadChecker bool
+	withHeadChecker  bool
-	headBytes       []byte
+	headBytes        []byte
-	samples         []*media.Sample
+	samples          []*media.Sample
-	timestamps      []uint32
+	maxLate          uint16
-	maxLate         uint16
+	maxLateTimestamp uint32
 }
 
-type fakeDepacketizer struct {
+type fakeDepacketizer struct{}
-}
 
 func (f *fakeDepacketizer) Unmarshal(r []byte) ([]byte, error) {
 	return r, nil
 }
 
+func (f *fakeDepacketizer) IsDetectedFinalPacketInSequence(rtpPacketMarketBit bool) bool {
+	return rtpPacketMarketBit
+}
+
 type fakePartitionHeadChecker struct {
 	headBytes []byte
 }
@@ -47,24 +50,82 @@ func TestSampleBuilder(t *testing.T) {
 			packets: []*rtp.Packet{
 				{Header: rtp.Header{SequenceNumber: 5000, Timestamp: 5}, Payload: []byte{0x01}},
 			},
-			samples:    []*media.Sample{},
+			samples:          []*media.Sample{},
-			timestamps: []uint32{},
+			maxLate:          50,
-			maxLate:    50,
+			maxLateTimestamp: 0,
 		},
 		{
-			message: "SampleBuilder should emit one packet, we had three packets with unique timestamps",
+			message: "SampleBuilder shouldn't emit anything if only one RTP packet has been pushed even if the market bit is set",
+			packets: []*rtp.Packet{
+				{Header: rtp.Header{SequenceNumber: 5000, Timestamp: 5, Marker: true}, Payload: []byte{0x01}},
+			},
+			samples:          []*media.Sample{},
+			maxLate:          50,
+			maxLateTimestamp: 0,
+		},
+		{
+			message: "SampleBuilder should emit two packets, we had three packets with unique timestamps",
 			packets: []*rtp.Packet{
 				{Header: rtp.Header{SequenceNumber: 5000, Timestamp: 5}, Payload: []byte{0x01}},
 				{Header: rtp.Header{SequenceNumber: 5001, Timestamp: 6}, Payload: []byte{0x02}},
 				{Header: rtp.Header{SequenceNumber: 5002, Timestamp: 7}, Payload: []byte{0x03}},
 			},
 			samples: []*media.Sample{
-				{Data: []byte{0x02}, Duration: time.Second},
+				{Data: []byte{0x01}, Duration: time.Second, PacketTimestamp: 5},
+				{Data: []byte{0x02}, Duration: time.Second, PacketTimestamp: 6},
 			},
-			timestamps: []uint32{
+			maxLate:          50,
-				6,
+			maxLateTimestamp: 0,
+		},
+		{
+			message: "SampleBuilder should emit one packet, we had a packet end of sequence marker and run out of space",
+			packets: []*rtp.Packet{
+				{Header: rtp.Header{SequenceNumber: 5000, Timestamp: 5, Marker: true}, Payload: []byte{0x01}},
+				{Header: rtp.Header{SequenceNumber: 5002, Timestamp: 7}, Payload: []byte{0x02}},
+				{Header: rtp.Header{SequenceNumber: 5004, Timestamp: 9}, Payload: []byte{0x03}},
+				{Header: rtp.Header{SequenceNumber: 5006, Timestamp: 11}, Payload: []byte{0x04}},
+				{Header: rtp.Header{SequenceNumber: 5008, Timestamp: 13}, Payload: []byte{0x05}},
+				{Header: rtp.Header{SequenceNumber: 5010, Timestamp: 15}, Payload: []byte{0x06}},
+				{Header: rtp.Header{SequenceNumber: 5012, Timestamp: 17}, Payload: []byte{0x07}},
 			},
-			maxLate: 50,
+			samples: []*media.Sample{
+				{Data: []byte{0x01}, Duration: time.Second * 2, PacketTimestamp: 5},
+			},
+			maxLate:          5,
+			maxLateTimestamp: 0,
+		},
+		{
+			message: "SampleBuilder shouldn't emit any packet, we do not have a valid end of sequence and run out of space",
+			packets: []*rtp.Packet{
+				{Header: rtp.Header{SequenceNumber: 5000, Timestamp: 5}, Payload: []byte{0x01}},
+				{Header: rtp.Header{SequenceNumber: 5002, Timestamp: 7}, Payload: []byte{0x02}},
+				{Header: rtp.Header{SequenceNumber: 5004, Timestamp: 9}, Payload: []byte{0x03}},
+				{Header: rtp.Header{SequenceNumber: 5006, Timestamp: 11}, Payload: []byte{0x04}},
+				{Header: rtp.Header{SequenceNumber: 5008, Timestamp: 13}, Payload: []byte{0x05}},
+				{Header: rtp.Header{SequenceNumber: 5010, Timestamp: 15}, Payload: []byte{0x06}},
+				{Header: rtp.Header{SequenceNumber: 5012, Timestamp: 17}, Payload: []byte{0x07}},
+			},
+			samples:          []*media.Sample{},
+			maxLate:          5,
+			maxLateTimestamp: 0,
+		},
+		{
+			message: "SampleBuilder should emit one packet, we had a packet end of sequence marker and run out of space",
+			packets: []*rtp.Packet{
+				{Header: rtp.Header{SequenceNumber: 5000, Timestamp: 5, Marker: true}, Payload: []byte{0x01}},
+				{Header: rtp.Header{SequenceNumber: 5002, Timestamp: 7, Marker: true}, Payload: []byte{0x02}},
+				{Header: rtp.Header{SequenceNumber: 5004, Timestamp: 9}, Payload: []byte{0x03}},
+				{Header: rtp.Header{SequenceNumber: 5006, Timestamp: 11}, Payload: []byte{0x04}},
+				{Header: rtp.Header{SequenceNumber: 5008, Timestamp: 13}, Payload: []byte{0x05}},
+				{Header: rtp.Header{SequenceNumber: 5010, Timestamp: 15}, Payload: []byte{0x06}},
+				{Header: rtp.Header{SequenceNumber: 5012, Timestamp: 17}, Payload: []byte{0x07}},
+			},
+			samples: []*media.Sample{
+				{Data: []byte{0x01}, Duration: time.Second * 2, PacketTimestamp: 5},
+				{Data: []byte{0x02}, Duration: time.Second * 2, PacketTimestamp: 7, PrevDroppedPackets: 1},
+			},
+			maxLate:          5,
+			maxLateTimestamp: 0,
 		},
 		{
 			message: "SampleBuilder should emit one packet, we had two packets but two with duplicate timestamps",
@@ -75,12 +136,11 @@ func TestSampleBuilder(t *testing.T) {
 				{Header: rtp.Header{SequenceNumber: 5003, Timestamp: 7}, Payload: []byte{0x04}},
 			},
 			samples: []*media.Sample{
-				{Data: []byte{0x02, 0x03}, Duration: time.Second},
+				{Data: []byte{0x01}, Duration: time.Second, PacketTimestamp: 5},
+				{Data: []byte{0x02, 0x03}, Duration: time.Second, PacketTimestamp: 6},
 			},
-			timestamps: []uint32{
+			maxLate:          50,
-				6,
+			maxLateTimestamp: 0,
-			},
-			maxLate: 50,
 		},
 		{
 			message: "SampleBuilder shouldn't emit a packet because we have a gap before a valid one",
@@ -89,26 +149,22 @@ func TestSampleBuilder(t *testing.T) {
 				{Header: rtp.Header{SequenceNumber: 5007, Timestamp: 6}, Payload: []byte{0x02}},
 				{Header: rtp.Header{SequenceNumber: 5008, Timestamp: 7}, Payload: []byte{0x03}},
 			},
-			samples:    []*media.Sample{},
+			samples:          []*media.Sample{},
-			timestamps: []uint32{},
+			maxLate:          50,
-			maxLate:    50,
+			maxLateTimestamp: 0,
 		},
 		{
-			message: "SampleBuilder should emit a packet after a gap if PartitionHeadChecker assumes it head",
+			message: "SampleBuilder shouldn't emit a packet after a gap as there are gaps and have not reached maxLate yet",
 			packets: []*rtp.Packet{
 				{Header: rtp.Header{SequenceNumber: 5000, Timestamp: 5}, Payload: []byte{0x01}},
 				{Header: rtp.Header{SequenceNumber: 5007, Timestamp: 6}, Payload: []byte{0x02}},
 				{Header: rtp.Header{SequenceNumber: 5008, Timestamp: 7}, Payload: []byte{0x03}},
 			},
-			withHeadChecker: true,
+			withHeadChecker:  true,
-			headBytes:       []byte{0x02},
+			headBytes:        []byte{0x02},
-			samples: []*media.Sample{
+			samples:          []*media.Sample{},
-				{Data: []byte{0x02}, Duration: 0},
+			maxLate:          50,
-			},
+			maxLateTimestamp: 0,
-			timestamps: []uint32{
-				6,
-			},
-			maxLate: 50,
 		},
 		{
 			message: "SampleBuilder shouldn't emit a packet after a gap if PartitionHeadChecker doesn't assume it head",
@@ -117,11 +173,11 @@ func TestSampleBuilder(t *testing.T) {
 				{Header: rtp.Header{SequenceNumber: 5007, Timestamp: 6}, Payload: []byte{0x02}},
 				{Header: rtp.Header{SequenceNumber: 5008, Timestamp: 7}, Payload: []byte{0x03}},
 			},
-			withHeadChecker: true,
+			withHeadChecker:  true,
-			headBytes:       []byte{},
+			headBytes:        []byte{},
-			samples:         []*media.Sample{},
+			samples:          []*media.Sample{},
-			timestamps:      []uint32{},
+			maxLate:          50,
-			maxLate:         50,
+			maxLateTimestamp: 0,
 		},
 		{
 			message: "SampleBuilder should emit multiple valid packets",
@@ -134,18 +190,34 @@ func TestSampleBuilder(t *testing.T) {
 				{Header: rtp.Header{SequenceNumber: 5005, Timestamp: 6}, Payload: []byte{0x06}},
 			},
 			samples: []*media.Sample{
-				{Data: []byte{0x02}, Duration: time.Second},
+				{Data: []byte{0x01}, Duration: time.Second, PacketTimestamp: 1},
-				{Data: []byte{0x03}, Duration: time.Second},
+				{Data: []byte{0x02}, Duration: time.Second, PacketTimestamp: 2},
-				{Data: []byte{0x04}, Duration: time.Second},
+				{Data: []byte{0x03}, Duration: time.Second, PacketTimestamp: 3},
-				{Data: []byte{0x05}, Duration: time.Second},
+				{Data: []byte{0x04}, Duration: time.Second, PacketTimestamp: 4},
+				{Data: []byte{0x05}, Duration: time.Second, PacketTimestamp: 5},
 			},
-			timestamps: []uint32{
+			maxLate:          50,
-				2,
+			maxLateTimestamp: 0,
-				3,
+		},
-				4,
+		{
-				5,
+			message: "SampleBuilder should skip time stamps too old",
+			packets: []*rtp.Packet{
+				{Header: rtp.Header{SequenceNumber: 5000, Timestamp: 1}, Payload: []byte{0x01}},
+				{Header: rtp.Header{SequenceNumber: 5001, Timestamp: 2}, Payload: []byte{0x02}},
+				{Header: rtp.Header{SequenceNumber: 5002, Timestamp: 3}, Payload: []byte{0x03}},
+				{Header: rtp.Header{SequenceNumber: 5013, Timestamp: 4000}, Payload: []byte{0x04}},
+				{Header: rtp.Header{SequenceNumber: 5014, Timestamp: 4000}, Payload: []byte{0x05}},
+				{Header: rtp.Header{SequenceNumber: 5015, Timestamp: 4002}, Payload: []byte{0x06}},
+				{Header: rtp.Header{SequenceNumber: 5016, Timestamp: 7000}, Payload: []byte{0x04}},
+				{Header: rtp.Header{SequenceNumber: 5017, Timestamp: 7001}, Payload: []byte{0x05}},
 			},
-			maxLate: 50,
+			samples: []*media.Sample{
+				{Data: []byte{0x04, 0x05}, Duration: time.Second * time.Duration(2), PacketTimestamp: 4000, PrevDroppedPackets: 13},
+			},
+			withHeadChecker:  true,
+			headBytes:        []byte{0x04},
+			maxLate:          50,
+			maxLateTimestamp: 2000,
 		},
 	}
 
@@ -159,6 +231,11 @@ func TestSampleBuilder(t *testing.T) {
 					&fakePartitionHeadChecker{headBytes: t.headBytes},
 				))
 			}
+			if t.maxLateTimestamp != 0 {
+				opts = append(opts, WithMaxTimeDelay(
+					time.Millisecond*time.Duration(int64(t.maxLateTimestamp)),
+				))
+			}
 
 			s := New(t.maxLate, &fakeDepacketizer{}, 1, opts...)
 			samples := []*media.Sample{}
@@ -169,35 +246,7 @@ func TestSampleBuilder(t *testing.T) {
 			for sample := s.Pop(); sample != nil; sample = s.Pop() {
 				samples = append(samples, sample)
 			}
-
+			assert.Equal(t.samples, samples, t.message)
-			assert.Equal(samples, t.samples, t.message)
-		}
-	})
-	t.Run("PopWithTimestamp", func(t *testing.T) {
-		assert := assert.New(t)
-
-		for _, t := range testData {
-			var opts []Option
-			if t.withHeadChecker {
-				opts = append(opts, WithPartitionHeadChecker(
-					&fakePartitionHeadChecker{headBytes: t.headBytes},
-				))
-			}
-
-			s := New(t.maxLate, &fakeDepacketizer{}, 1, opts...)
-			samples := []*media.Sample{}
-			timestamps := []uint32{}
-
-			for _, p := range t.packets {
-				s.Push(p)
-			}
-			for sample, timestamp := s.PopWithTimestamp(); sample != nil; sample, timestamp = s.PopWithTimestamp() {
-				samples = append(samples, sample)
-				timestamps = append(timestamps, timestamp)
-			}
-
-			assert.Equal(samples, t.samples, t.message)
-			assert.Equal(timestamps, t.timestamps, t.message)
 		}
 	})
 }
@@ -210,12 +259,18 @@ func TestSampleBuilderMaxLate(t *testing.T) {
 	s.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: 0, Timestamp: 1}, Payload: []byte{0x01}})
 	s.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: 1, Timestamp: 2}, Payload: []byte{0x01}})
 	s.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: 2, Timestamp: 3}, Payload: []byte{0x01}})
-	assert.Equal(s.Pop(), &media.Sample{Data: []byte{0x01}, Duration: time.Second}, "Failed to build samples before gap")
+	assert.Equal(&media.Sample{Data: []byte{0x01}, Duration: time.Second, PacketTimestamp: 1}, s.Pop(), "Failed to build samples before gap")
 
 	s.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: 5000, Timestamp: 500}, Payload: []byte{0x02}})
 	s.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: 5001, Timestamp: 501}, Payload: []byte{0x02}})
 	s.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: 5002, Timestamp: 502}, Payload: []byte{0x02}})
-	assert.Equal(s.Pop(), &media.Sample{Data: []byte{0x02}, Duration: time.Second}, "Failed to build samples after large gap")
+
+	assert.Equal(&media.Sample{Data: []byte{0x01}, Duration: time.Second, PacketTimestamp: 2}, s.Pop(), "Failed to build samples after large gap")
+	assert.Equal((*media.Sample)(nil), s.Pop(), "Failed to build samples after large gap")
+
+	s.Push(&rtp.Packet{Header: rtp.Header{SequenceNumber: 6000, Timestamp: 600}, Payload: []byte{0x03}})
+	assert.Equal(&media.Sample{Data: []byte{0x02}, Duration: time.Second, PacketTimestamp: 500, PrevDroppedPackets: 4998}, s.Pop(), "Failed to build samples after large gap")
+	assert.Equal(&media.Sample{Data: []byte{0x02}, Duration: time.Second, PacketTimestamp: 501}, s.Pop(), "Failed to build samples after large gap")
 }
 
 func TestSeqnumDistance(t *testing.T) {
@@ -285,6 +340,7 @@ func TestSampleBuilderWithPacketReleaseHandler(t *testing.T) {
 		{Header: rtp.Header{SequenceNumber: 11, Timestamp: 120}, Payload: []byte{0x02}},
 		{Header: rtp.Header{SequenceNumber: 12, Timestamp: 121}, Payload: []byte{0x03}},
 		{Header: rtp.Header{SequenceNumber: 13, Timestamp: 122}, Payload: []byte{0x04}},
+		{Header: rtp.Header{SequenceNumber: 21, Timestamp: 200}, Payload: []byte{0x05}},
 	}
 	s := New(10, &fakeDepacketizer{}, 1, WithPacketReleaseHandler(fakePacketReleaseHandler))
 	s.Push(&pkts[0])
@@ -298,13 +354,14 @@ func TestSampleBuilderWithPacketReleaseHandler(t *testing.T) {
 	// Test packets released after samples built.
 	s.Push(&pkts[2])
 	s.Push(&pkts[3])
+	s.Push(&pkts[4])
 	if s.Pop() == nil {
 		t.Errorf("Should have some sample here.")
 	}
-	if len(released) != 2 {
+	if len(released) < 3 {
 		t.Errorf("packet built with sample is not released")
 	}
-	if len(released) >= 2 && released[1].SequenceNumber != pkts[2].SequenceNumber {
+	if len(released) >= 2 && released[2].SequenceNumber != pkts[2].SequenceNumber {
 		t.Errorf("Unexpected packet released by samples built")
 	}
 }

track_local_static.go

@@ -167,6 +167,7 @@ func (s *TrackLocalStaticRTP) Write(b []byte) (n int, err error) {
 // If you wish to send a RTP Packet use TrackLocalStaticRTP
 type TrackLocalStaticSample struct {
 	packetizer rtp.Packetizer
+	sequencer  rtp.Sequencer
 	rtpTrack   *TrackLocalStaticRTP
 	clockRate  float64
 }
@@ -221,12 +222,13 @@ func (s *TrackLocalStaticSample) Bind(t TrackLocalContext) (RTPCodecParameters,
 		return codec, err
 	}
 
+	s.sequencer = rtp.NewRandomSequencer()
 	s.packetizer = rtp.NewPacketizer(
 		rtpOutboundMTU,
 		0, // Value is handled when writing
 		0, // Value is handled when writing
 		payloader,
-		rtp.NewRandomSequencer(),
+		s.sequencer,
 		codec.ClockRate,
 	)
 	s.clockRate = float64(codec.RTPCodecCapability.ClockRate)
@@ -253,8 +255,16 @@ func (s *TrackLocalStaticSample) WriteSample(sample media.Sample) error {
 		return nil
 	}
 
-	samples := sample.Duration.Seconds() * clockRate
+	// skip packets by the number of previously dropped packets
-	packets := p.(rtp.Packetizer).Packetize(sample.Data, uint32(samples))
+	for i := uint16(0); i < sample.PrevDroppedPackets; i++ {
+		s.sequencer.NextSequenceNumber()
+	}
+
+	samples := uint32(sample.Duration.Seconds() * clockRate)
+	if sample.PrevDroppedPackets > 0 {
+		p.(rtp.Packetizer).SkipSamples(samples * uint32(sample.PrevDroppedPackets))
+	}
+	packets := p.(rtp.Packetizer).Packetize(sample.Data, samples)
 
 	writeErrs := []error{}
 	for _, p := range packets {