Export receive time of packet. (#22)

* Export receive time of packet.

* staticcheck

* handler close

* writer closer

* not the correct interface

Author: boks1971

jitter/buffer.go

@@ -25,6 +25,11 @@ import (
 	"github.com/livekit/protocol/logger"
 )
 
+type ExtPacket struct {
+	ReceivedAt time.Time
+	*rtp.Packet
+}
+
 type Buffer struct {
 	depacketizer rtp.Depacketizer
 	latency      time.Duration
@@ -58,7 +63,7 @@ type BufferStats struct {
 	SamplesPopped  uint64 // samples sent to handler
 }
 
-type PacketFunc func(packets []*rtp.Packet)
+type PacketFunc func(packets []ExtPacket)
 
 func NewBuffer(
 	depacketizer rtp.Depacketizer,
@@ -117,7 +122,7 @@ func (b *Buffer) UpdateLatency(latency time.Duration) {
 
 	b.latency = latency
 	if b.head != nil {
-		b.timer.Reset(time.Until(b.head.received.Add(latency)))
+		b.timer.Reset(time.Until(b.head.extPacket.ReceivedAt.Add(latency)))
 	}
 }
 
@@ -199,10 +204,10 @@ func (b *Buffer) push(pkt *rtp.Packet) {
 		return
 	}
 
-	beforeHead := before(pkt.SequenceNumber, b.head.packet.SequenceNumber)
-	afterTail := !before(pkt.SequenceNumber, b.tail.packet.SequenceNumber)
-	withinHeadRange := withinRange(pkt.SequenceNumber, b.head.packet.SequenceNumber)
-	withinTailRange := withinRange(pkt.SequenceNumber, b.tail.packet.SequenceNumber)
+	beforeHead := before(pkt.SequenceNumber, b.head.extPacket.SequenceNumber)
+	afterTail := !before(pkt.SequenceNumber, b.tail.extPacket.SequenceNumber)
+	withinHeadRange := withinRange(pkt.SequenceNumber, b.head.extPacket.SequenceNumber)
+	withinTailRange := withinRange(pkt.SequenceNumber, b.tail.extPacket.SequenceNumber)
 
 	switch {
 	case beforeHead && withinHeadRange:
@@ -221,8 +226,8 @@ func (b *Buffer) push(pkt *rtp.Packet) {
 	case withinTailRange:
 		// insert, search from tail
 		for c := b.tail.prev; c != nil; c = c.prev {
-			discont = !withinRange(pkt.SequenceNumber, c.packet.SequenceNumber)
-			if !before(pkt.SequenceNumber, c.packet.SequenceNumber) || discont {
+			discont = !withinRange(pkt.SequenceNumber, c.extPacket.SequenceNumber)
+			if !before(pkt.SequenceNumber, c.extPacket.SequenceNumber) || discont {
 				// insert after c
 				p.discont = discont && p.start
 				p.prev = c
@@ -236,8 +241,8 @@ func (b *Buffer) push(pkt *rtp.Packet) {
 	case withinHeadRange:
 		// insert, search from head
 		for c := b.head.next; c != nil; c = c.next {
-			discont = !withinRange(pkt.SequenceNumber, c.packet.SequenceNumber)
-			if before(pkt.SequenceNumber, c.packet.SequenceNumber) || discont {
+			discont = !withinRange(pkt.SequenceNumber, c.extPacket.SequenceNumber)
+			if before(pkt.SequenceNumber, c.extPacket.SequenceNumber) || discont {
 				// insert before c
 				p.prev = c.prev
 				p.next = c
@@ -266,12 +271,12 @@ func (b *Buffer) popReady() {
 	for b.head != nil &&
 		b.head.isComplete() {
 
-		if b.head.packet.SequenceNumber == b.prevSN+1 || b.head.discont || !b.initialized {
+		if b.head.extPacket.SequenceNumber == b.prevSN+1 || b.head.discont || !b.initialized {
 			// normal
-		} else if b.head.received.Before(expiry) {
+		} else if b.head.extPacket.ReceivedAt.Before(expiry) {
 			// max latency reached
 			loss = true
-			b.stats.PacketsLost += uint64(b.head.packet.SequenceNumber - b.prevSN - 1)
+			b.stats.PacketsLost += uint64(b.head.extPacket.SequenceNumber - b.prevSN - 1)
 		} else {
 			break
 		}
@@ -286,17 +291,17 @@ func (b *Buffer) popReady() {
 	}
 
 	if b.head != nil {
-		b.timer.Reset(time.Until(b.head.received.Add(b.latency)))
+		b.timer.Reset(time.Until(b.head.extPacket.ReceivedAt.Add(b.latency)))
 	}
 }
 
 // dropIncompleteExpired drops incomplete expired packets
 func (b *Buffer) dropIncompleteExpired(expiry time.Time) {
 	dropped := false
 
-	for b.head != nil && !b.head.isComplete() && b.head.received.Before(expiry) {
+	for b.head != nil && !b.head.isComplete() && b.head.extPacket.ReceivedAt.Before(expiry) {
 		if b.initialized && !b.head.discont {
-			b.stats.PacketsLost += uint64(b.head.packet.SequenceNumber - b.prevSN - 1)
+			b.stats.PacketsLost += uint64(b.head.extPacket.SequenceNumber - b.prevSN - 1)
 		}
 
 		b.free(b.popHead())
@@ -310,15 +315,15 @@ func (b *Buffer) dropIncompleteExpired(expiry time.Time) {
 	}
 }
 
-func (b *Buffer) popSample() []*rtp.Packet {
-	sample := make([]*rtp.Packet, 0, b.size)
+func (b *Buffer) popSample() []ExtPacket {
+	sample := make([]ExtPacket, 0, b.size)
 	end := false
 	for !end {
 		c := b.popHead()
 		end = c.end
 
-		if !c.packet.Padding {
-			sample = append(sample, c.packet)
+		if !c.extPacket.Padding {
+			sample = append(sample, c.extPacket)
 		}
 
 		b.stats.PacketsPopped++
@@ -333,7 +338,7 @@ func (b *Buffer) popSample() []*rtp.Packet {
 
 func (b *Buffer) popHead() *packet {
 	c := b.head
-	b.prevSN = c.packet.SequenceNumber
+	b.prevSN = c.extPacket.SequenceNumber
 	b.head = c.next
 	if b.head == nil {
 		b.tail = nil

jitter/buffer_test.go

@@ -26,8 +26,8 @@ import (
 
 const testBufferLatency = 800 * time.Millisecond
 
-func chanFunc(t testing.TB, out chan<- []*rtp.Packet) PacketFunc {
-	return func(packets []*rtp.Packet) {
+func chanFunc(t testing.TB, out chan<- []ExtPacket) PacketFunc {
+	return func(packets []ExtPacket) {
 		select {
 		case out <- packets:
 		default:
@@ -37,7 +37,7 @@ func chanFunc(t testing.TB, out chan<- []*rtp.Packet) PacketFunc {
 }
 
 func TestJitterBuffer(t *testing.T) {
-	out := make(chan []*rtp.Packet, 100)
+	out := make(chan []ExtPacket, 100)
 	b := NewBuffer(&testDepacketizer{}, testBufferLatency, chanFunc(t, out))
 	s := newTestStream()
 
@@ -57,7 +57,7 @@ func TestJitterBuffer(t *testing.T) {
 }
 
 func TestSamples(t *testing.T) {
-	out := make(chan []*rtp.Packet, 100)
+	out := make(chan []ExtPacket, 100)
 	b := NewBuffer(&testDepacketizer{}, testBufferLatency, chanFunc(t, out))
 	s := newTestStream()
 
@@ -80,7 +80,7 @@ func TestSamples(t *testing.T) {
 }
 
 func TestJitter(t *testing.T) {
-	out := make(chan []*rtp.Packet, 100)
+	out := make(chan []ExtPacket, 100)
 	b := NewBuffer(&testDepacketizer{}, testBufferLatency, chanFunc(t, out))
 	s := newTestStream()
 
@@ -114,7 +114,7 @@ func TestJitter(t *testing.T) {
 }
 
 func TestDiscontinuity(t *testing.T) {
-	out := make(chan []*rtp.Packet, 100)
+	out := make(chan []ExtPacket, 100)
 	b := NewBuffer(&testDepacketizer{}, testBufferLatency, chanFunc(t, out))
 	s := newTestStream()
 
@@ -139,7 +139,7 @@ func TestDiscontinuity(t *testing.T) {
 }
 
 func TestLostPackets(t *testing.T) {
-	out := make(chan []*rtp.Packet, 100)
+	out := make(chan []ExtPacket, 100)
 	b := NewBuffer(&testDepacketizer{}, testBufferLatency, chanFunc(t, out))
 	s := newTestStream()
 
@@ -173,7 +173,7 @@ func TestLostPackets(t *testing.T) {
 }
 
 func TestDroppedPackets(t *testing.T) {
-	out := make(chan []*rtp.Packet, 100)
+	out := make(chan []ExtPacket, 100)
 	b := NewBuffer(&testDepacketizer{}, testBufferLatency, chanFunc(t, out))
 	s := newTestStream()
 
@@ -232,7 +232,7 @@ func TestDroppedPackets(t *testing.T) {
 	})
 }
 
-func checkSample(t *testing.T, out chan []*rtp.Packet, expected int) {
+func checkSample(t *testing.T, out chan []ExtPacket, expected int) {
 	select {
 	case sample := <-out:
 		if expected == 0 {

jitter/packet.go

@@ -21,11 +21,10 @@ import (
 )
 
 type packet struct {
-	received   time.Time
+	extPacket  ExtPacket
 	prev, next *packet
 	start, end bool
 	discont    bool
-	packet     *rtp.Packet
 }
 
 func (b *Buffer) newPacket(pkt *rtp.Packet) *packet {
@@ -37,12 +36,11 @@ func (b *Buffer) newPacket(pkt *rtp.Packet) *packet {
 	}
 	b.pool = p.next
 
-	p.received = time.Now()
 	p.prev = nil
 	p.next = nil
 	p.start = b.depacketizer.IsPartitionHead(pkt.Payload)
 	p.end = b.depacketizer.IsPartitionTail(pkt.Marker, pkt.Payload)
-	p.packet = pkt
+	p.extPacket = ExtPacket{time.Now(), pkt}
 
 	return p
 }
@@ -57,7 +55,7 @@ func (p *packet) isComplete() bool {
 	}
 
 	for c := p; c.next != nil; c = c.next {
-		if c.next.packet.SequenceNumber != c.packet.SequenceNumber+1 {
+		if c.next.extPacket.SequenceNumber != c.extPacket.SequenceNumber+1 {
 			return false
 		}
 		if c.next.end {
@@ -72,7 +70,7 @@ func (b *Buffer) free(pkt *packet) {
 	b.size--
 
 	pkt.prev = nil
-	pkt.packet = nil
+	pkt.extPacket = ExtPacket{}
 	pkt.next = b.pool
 
 	b.pool = pkt

rtp/jitter.go

@@ -33,9 +33,9 @@ func HandleJitter(h HandlerCloser) HandlerCloser {
 	}
 	// Jitter buffer expects to be closed (to stop the timer), but handler interface doesn't allow it.
 	// This should be fine, because GC can now collect timers and goroutines blocked on them if they are not referenced.
-	handler.buf = jitter.NewBuffer(audioDepacketizer{}, jitterMaxLatency, func(packets []*rtp.Packet) {
+	handler.buf = jitter.NewBuffer(audioDepacketizer{}, jitterMaxLatency, func(packets []jitter.ExtPacket) {
 		for _, p := range packets {
-			handler.handleRTP(p)
+			handler.handleRTP(p.Packet)
 		}
 	})
 	return handler