Method deliveryLoop

cmd/test-streammanager/deliverypipe.go:203–241

(
	getPending func() *[]taggedPacket,
	deliver func(taggedPacket),
)

}

func (dp *DeliveryPipe) deliveryLoop(
	getPending func() *[]taggedPacket,
	deliver func(taggedPacket),
) {
	for {
		dp.lock.Lock()
		if dp.closed {
			dp.lock.Unlock()
			return
		}

		pending := getPending()
		now := time.Now()

		// Find all packets ready for delivery (deliveryTime <= now)
		readyCount := 0
		for _, pkt := range *pending {
			if pkt.deliveryTime.After(now) {
				break
			}
			readyCount++
		}

		// Extract ready packets
		ready := make([]taggedPacket, readyCount)
		copy(ready, (*pending)[:readyCount])
		*pending = (*pending)[readyCount:]

		dp.lock.Unlock()

		// Deliver all ready packets (outside lock)
		for _, pkt := range ready {
			deliver(pkt)
		}

		// Always sleep 1ms - simple busy loop
		time.Sleep(1 * time.Millisecond)
	}
}

func (dp *DeliveryPipe) Close() {
	dp.lock.Lock()