From 7d97c9b5d30b255b464e20497bbf947d2f13d0b9 Mon Sep 17 00:00:00 2001 From: Robin Raymond Date: Thu, 6 May 2021 14:57:25 -0400 Subject: [PATCH] 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 | 1 + examples/custom-logger/main.go | 6 +- go.mod | 2 +- go.sum | 3 +- pkg/media/h264writer/h264writer.go | 12 +- pkg/media/h264writer/h264writer_test.go | 29 +- pkg/media/media.go | 8 +- .../samplebuilder/sampleSequenceLocation.go | 82 +++++ pkg/media/samplebuilder/samplebuilder.go | 332 +++++++++++++----- pkg/media/samplebuilder/samplebuilder_test.go | 223 +++++++----- track_local_static.go | 16 +- 11 files changed, 516 insertions(+), 198 deletions(-) create mode 100644 pkg/media/samplebuilder/sampleSequenceLocation.go diff --git a/README.md b/README.md index b998a200..5883ae9e 100644 --- a/README.md +++ b/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 diff --git a/examples/custom-logger/main.go b/examples/custom-logger/main.go index 0976941c..49c187d8 100644 --- a/examples/custom-logger/main.go +++ b/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) diff --git a/go.mod b/go.mod index 8a77deee..979edd56 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 6dd307fc..d1645f84 100644 --- a/go.sum +++ b/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= diff --git a/pkg/media/h264writer/h264writer.go b/pkg/media/h264writer/h264writer.go index 9637441d..578af017 100644 --- a/pkg/media/h264writer/h264writer.go +++ b/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 - hasKeyFrame bool + writer io.Writer + 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() diff --git a/pkg/media/h264writer/h264writer_test.go b/pkg/media/h264writer/h264writer_test.go index eac3dadc..b414d763 100644 --- a/pkg/media/h264writer/h264writer_test.go +++ b/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 + } }) } } diff --git a/pkg/media/media.go b/pkg/media/media.go index bcd7e338..4b00edbe 100644 --- a/pkg/media/media.go +++ b/pkg/media/media.go @@ -9,9 +9,11 @@ import ( // A Sample contains encoded media and timing information type Sample struct { - Data []byte - Timestamp time.Time - Duration time.Duration + Data []byte + Timestamp time.Time + Duration time.Duration + PacketTimestamp uint32 + PrevDroppedPackets uint16 } // Writer defines an interface to handle diff --git a/pkg/media/samplebuilder/sampleSequenceLocation.go b/pkg/media/samplebuilder/sampleSequenceLocation.go new file mode 100644 index 00000000..56b9b970 --- /dev/null +++ b/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 +} diff --git a/pkg/media/samplebuilder/samplebuilder.go b/pkg/media/samplebuilder/samplebuilder.go index eb4296ed..59861e39 100644 --- a/pkg/media/samplebuilder/samplebuilder.go +++ b/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 - buffer [65536]*rtp.Packet + maxLate uint16 // how many packets to wait until we get a valid Sample + 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. - if int16(p.SequenceNumber-s.lastPush) > 0 { - for i := s.lastPush; i != p.SequenceNumber+1; i++ { - s.releasePacket(i - s.maxLate) - } + switch s.filled.compare(p.SequenceNumber) { + case slCompareVoid: + s.filled.head = p.SequenceNumber + 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.lastPush = p.SequenceNumber + s.purgeBuffers() } const secondToNanoseconds = 1000000000 -// We have a valid collection of RTP Packets -// walk forwards building a sample if everything looks good clear and update buffer+values -func (s *SampleBuilder) buildSample(firstBuffer uint16) (*media.Sample, uint32) { +// buildSample creates a sample from a valid collection of RTP Packets by +// walking forwards building a sample if everything looks good clear and +// 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 i := firstBuffer; s.buffer[i] != nil; i++ { - 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) + for ; consume.head != consume.tail; consume.head++ { + p, err := s.depacketizer.Unmarshal(s.buffer[consume.head].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. -// It returns nil if no valid samples have been found. -func (s *SampleBuilder) Pop() *media.Sample { - sample, _ := s.PopWithTimestamp() - return sample -} - -// 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 - } +// timestampDistance computes the distance between two timestamps +func timestampDistance(x, y uint32) uint32 { + diff := int32(x - y) + if diff < 0 { + return uint32(-diff) } - for ; i != s.lastPush; i++ { - 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 + return uint32(diff) } // 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) + } +} diff --git a/pkg/media/samplebuilder/samplebuilder_test.go b/pkg/media/samplebuilder/samplebuilder_test.go index 91e28758..937f8a98 100644 --- a/pkg/media/samplebuilder/samplebuilder_test.go +++ b/pkg/media/samplebuilder/samplebuilder_test.go @@ -11,22 +11,25 @@ import ( ) type sampleBuilderTest struct { - message string - packets []*rtp.Packet - withHeadChecker bool - headBytes []byte - samples []*media.Sample - timestamps []uint32 - maxLate uint16 + message string + packets []*rtp.Packet + withHeadChecker bool + headBytes []byte + samples []*media.Sample + 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{}, - timestamps: []uint32{}, - maxLate: 50, + samples: []*media.Sample{}, + 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{ - 6, + maxLate: 50, + 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{ - 6, - }, - maxLate: 50, + maxLate: 50, + maxLateTimestamp: 0, }, { 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{}, - timestamps: []uint32{}, - maxLate: 50, + samples: []*media.Sample{}, + 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, - headBytes: []byte{0x02}, - samples: []*media.Sample{ - {Data: []byte{0x02}, Duration: 0}, - }, - timestamps: []uint32{ - 6, - }, - maxLate: 50, + withHeadChecker: true, + headBytes: []byte{0x02}, + samples: []*media.Sample{}, + maxLate: 50, + maxLateTimestamp: 0, }, { 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, - headBytes: []byte{}, - samples: []*media.Sample{}, - timestamps: []uint32{}, - maxLate: 50, + withHeadChecker: true, + headBytes: []byte{}, + samples: []*media.Sample{}, + 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{0x03}, Duration: time.Second}, - {Data: []byte{0x04}, Duration: time.Second}, - {Data: []byte{0x05}, Duration: time.Second}, + {Data: []byte{0x01}, Duration: time.Second, PacketTimestamp: 1}, + {Data: []byte{0x02}, Duration: time.Second, PacketTimestamp: 2}, + {Data: []byte{0x03}, Duration: time.Second, PacketTimestamp: 3}, + {Data: []byte{0x04}, Duration: time.Second, PacketTimestamp: 4}, + {Data: []byte{0x05}, Duration: time.Second, PacketTimestamp: 5}, }, - timestamps: []uint32{ - 2, - 3, - 4, - 5, + maxLate: 50, + maxLateTimestamp: 0, + }, + { + 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(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) + assert.Equal(t.samples, samples, 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") } } diff --git a/track_local_static.go b/track_local_static.go index cf5aa22a..4275eb85 100644 --- a/track_local_static.go +++ b/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 - packets := p.(rtp.Packetizer).Packetize(sample.Data, uint32(samples)) + // skip packets by the number of previously dropped packets + 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 {