diff --git a/pkg/twcc/sender_interceptor.go b/pkg/twcc/sender_interceptor.go index 782dd1b9..8895e446 100644 --- a/pkg/twcc/sender_interceptor.go +++ b/pkg/twcc/sender_interceptor.go @@ -27,7 +27,10 @@ func (s *SenderInterceptorFactory) NewInterceptor(_ string) (interceptor.Interce packetChan: make(chan packet), close: make(chan struct{}), interval: 100 * time.Millisecond, - startTime: time.Now(), + now: time.Now, + newTicker: func(d time.Duration) Ticker { + return &timeTicker{time.NewTicker(d)} + }, } for _, opt := range s.opts { @@ -37,6 +40,8 @@ func (s *SenderInterceptorFactory) NewInterceptor(_ string) (interceptor.Interce } } + senderInterceptor.startTime = senderInterceptor.now() + if senderInterceptor.loggerFactory == nil { senderInterceptor.loggerFactory = logging.NewDefaultLoggerFactory() } @@ -66,6 +71,8 @@ type SenderInterceptor struct { interval time.Duration startTime time.Time + now func() time.Time + newTicker TickerFactory recorder *Recorder packetChan chan packet @@ -93,6 +100,24 @@ func WithLoggerFactory(loggerFactory logging.LoggerFactory) Option { } } +// SendNow sets an alternative for the time.Now function. +func SendNow(f func() time.Time) Option { + return func(s *SenderInterceptor) error { + s.now = f + + return nil + } +} + +// SendTicker sets an alternative for the time.NewTicker function. +func SendTicker(f TickerFactory) Option { + return func(s *SenderInterceptor) error { + s.newTicker = f + + return nil + } +} + // BindRTCPWriter lets you modify any outgoing RTCP packets. It is called once per PeerConnection. The returned method // will be called once per packet batch. func (s *SenderInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter { @@ -163,7 +188,7 @@ func (s *SenderInterceptor) BindRemoteStream( p := packet{ hdr: header, sequenceNumber: tccExt.TransportSequence, - arrivalTime: time.Since(s.startTime).Microseconds(), + arrivalTime: s.now().Sub(s.startTime).Microseconds(), ssrc: info.SSRC, } select { @@ -210,7 +235,7 @@ func (s *SenderInterceptor) loop(writer interceptor.RTCPWriter) { s.recorder.Record(p.ssrc, p.sequenceNumber, p.arrivalTime) } - ticker := time.NewTicker(s.interval) + ticker := s.newTicker(s.interval) for { select { case <-s.close: @@ -220,7 +245,7 @@ func (s *SenderInterceptor) loop(writer interceptor.RTCPWriter) { case p := <-s.packetChan: s.recorder.Record(p.ssrc, p.sequenceNumber, p.arrivalTime) - case <-ticker.C: + case <-ticker.Ch(): // build and send twcc pkts := s.recorder.BuildFeedbackPacket() if len(pkts) == 0 { diff --git a/pkg/twcc/sender_interceptor_test.go b/pkg/twcc/sender_interceptor_test.go index 3f951332..e5f4713b 100644 --- a/pkg/twcc/sender_interceptor_test.go +++ b/pkg/twcc/sender_interceptor_test.go @@ -44,7 +44,13 @@ func TestSenderInterceptor(t *testing.T) { }) t.Run("after RTP packets", func(t *testing.T) { - f, err := NewSenderInterceptor() + mt := &test.MockTime{} + mt.SetNow(time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)) + mTick := &test.MockTicker{C: make(chan time.Time)} + f, err := NewSenderInterceptor( + SendNow(mt.Now), + SendTicker(func(time.Duration) Ticker { return mTick }), + ) assert.NoError(t, err) i, err := f.NewInterceptor("") @@ -70,6 +76,13 @@ func TestSenderInterceptor(t *testing.T) { stream.ReceiveRTP(&rtp.Packet{Header: hdr}) } + // Reading a packet means the interceptor has handed it to the send + // loop, so the tick below happens after all packets are recorded. + for range 10 { + assert.NoError(t, (<-stream.ReadRTP()).Err) + } + mTick.Tick(mt.Now()) + pkts := <-stream.WrittenRTCP() assert.Equal(t, 1, len(pkts)) cc, ok := pkts[0].(*rtcp.TransportLayerCC) diff --git a/pkg/twcc/ticker.go b/pkg/twcc/ticker.go new file mode 100644 index 00000000..420c221e --- /dev/null +++ b/pkg/twcc/ticker.go @@ -0,0 +1,23 @@ +// SPDX-FileCopyrightText: 2026 The Pion community +// SPDX-License-Identifier: MIT + +package twcc + +import "time" + +// Ticker is an interface for *time.Ticker for use with the SendTicker option. +type Ticker interface { + Ch() <-chan time.Time + Stop() +} + +// TickerFactory is a factory to create new tickers. +type TickerFactory func(d time.Duration) Ticker + +type timeTicker struct { + *time.Ticker +} + +func (t *timeTicker) Ch() <-chan time.Time { + return t.C +}