Buffer packets when no RTCP writer is set

pkg/rfc8888/interceptor.go

@@ -15,6 +15,10 @@ import (
 	"github.com/pion/rtcp"
 )
 
+// packetChanBufferSize is the number of packets buffered for the report loop
+// before new packets are dropped.
+const packetChanBufferSize = 1000
+
 var errClosed = errors.New("interceptor is closed")
 
 // TickerFactory is a factory to create new tickers.
@@ -34,7 +38,7 @@ func (s *SenderInterceptorFactory) NewInterceptor(_ string) (interceptor.Interce
 		recorder:      NewRecorder(),
 		interval:      100 * time.Millisecond,
 		maxReportSize: 1200,
-		packetChan:    make(chan packet),
+		packetChan:    make(chan packet, packetChanBufferSize),
 		newTicker: func(d time.Duration) ticker {
 			return &timeTicker{time.NewTicker(d)}
 		},
@@ -128,10 +132,13 @@ func (s *SenderInterceptor) BindRemoteStream(
 			sequenceNumber: header.SequenceNumber,
 			ecn:            0, // ECN is not supported (yet).
 		}
-		select {
-		case <-s.close:
+		if s.isClosed() {
 			return 0, nil, errClosed
+		}
+		select {
 		case s.packetChan <- p:
+		default:
+			s.log.Trace("packet channel is full, dropping packet")
 		}
 
 		return i, attr, nil
@@ -161,6 +168,17 @@ func (s *SenderInterceptor) isClosed() bool {
 	}
 }
 
+func (s *SenderInterceptor) recordQueuedPackets() {
+	for range len(s.packetChan) {
+		select {
+		case pkt := <-s.packetChan:
+			s.recorder.AddPacket(pkt.arrival, pkt.ssrc, pkt.sequenceNumber, pkt.ecn)
+		default:
+			return
+		}
+	}
+}
+
 func (s *SenderInterceptor) loop(writer interceptor.RTCPWriter) {
 	defer s.wg.Done()
 
@@ -186,6 +204,7 @@ func (s *SenderInterceptor) loop(writer interceptor.RTCPWriter) {
 			s.recorder.AddPacket(pkt.arrival, pkt.ssrc, pkt.sequenceNumber, pkt.ecn)
 
 		case <-t.Ch():
+			s.recordQueuedPackets()
 			now := s.now()
 			s.log.Tracef("report triggered at %v", now)
 			if writer == nil {

pkg/rfc8888/interceptor_test.go

@@ -348,3 +348,41 @@ func TestConcurrentClose(t *testing.T) {
 	}
 	wg.Wait()
 }
+
+func TestReadWithoutRTCPWriter(t *testing.T) {
+	f, err := NewSenderInterceptor()
+	assert.NoError(t, err)
+
+	intcp, err := f.NewInterceptor("")
+	assert.NoError(t, err)
+	defer func() {
+		assert.NoError(t, intcp.Close())
+	}()
+
+	raw, err := (&rtp.Packet{
+		Header:  rtp.Header{Version: 2, SequenceNumber: 1, SSRC: 123456},
+		Payload: []byte{},
+	}).Marshal()
+	assert.NoError(t, err)
+
+	reader := intcp.BindRemoteStream(&interceptor.StreamInfo{SSRC: 123456}, interceptor.RTPReaderFunc(
+		func(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) {
+			return copy(b, raw), a, nil
+		},
+	))
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for range 2 * packetChanBufferSize {
+			_, _, err := reader.Read(make([]byte, 1500), nil)
+			assert.NoError(t, err)
+		}
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		assert.Fail(t, "read blocked without a bound RTCP writer")
+	}
+}

pkg/twcc/sender_interceptor.go

@@ -19,12 +19,16 @@ type SenderInterceptorFactory struct {
 	opts []Option
 }
 
+// packetChanBufferSize is the number of packets buffered for the report loop
+// before new packets are dropped.
+const packetChanBufferSize = 1000
+
 var errClosed = errors.New("interceptor is closed")
 
 // NewInterceptor constructs a new SenderInterceptor.
 func (s *SenderInterceptorFactory) NewInterceptor(_ string) (interceptor.Interceptor, error) {
 	senderInterceptor := &SenderInterceptor{
-		packetChan: make(chan packet),
+		packetChan: make(chan packet, packetChanBufferSize),
 		close:      make(chan struct{}),
 		interval:   100 * time.Millisecond,
 		startTime:  time.Now(),
@@ -166,10 +170,13 @@ func (s *SenderInterceptor) BindRemoteStream(
 					arrivalTime:    time.Since(s.startTime).Microseconds(),
 					ssrc:           info.SSRC,
 				}
-				select {
-				case <-s.close:
+				if s.isClosed() {
 					return 0, nil, errClosed
+				}
+				select {
 				case s.packetChan <- p:
+				default:
+					s.log.Trace("packet channel is full, dropping packet")
 				}
 			}
 
@@ -200,6 +207,17 @@ func (s *SenderInterceptor) isClosed() bool {
 	}
 }
 
+func (s *SenderInterceptor) recordQueuedPackets() {
+	for range len(s.packetChan) {
+		select {
+		case p := <-s.packetChan:
+			s.recorder.Record(p.ssrc, p.sequenceNumber, p.arrivalTime)
+		default:
+			return
+		}
+	}
+}
+
 func (s *SenderInterceptor) loop(writer interceptor.RTCPWriter) {
 	defer s.wg.Done()
 
@@ -221,6 +239,7 @@ func (s *SenderInterceptor) loop(writer interceptor.RTCPWriter) {
 			s.recorder.Record(p.ssrc, p.sequenceNumber, p.arrivalTime)
 
 		case <-ticker.C:
+			s.recordQueuedPackets()
 			// build and send twcc
 			pkts := s.recorder.BuildFeedbackPacket()
 			if len(pkts) == 0 {

pkg/twcc/sender_interceptor_test.go

@@ -310,3 +310,47 @@ func TestSenderInterceptor_Leak(t *testing.T) {
 		stream.ReceiveRTP(&rtp.Packet{Header: hdr})
 	}
 }
+
+func TestSenderInterceptorReadWithoutRTCPWriter(t *testing.T) {
+	f, err := NewSenderInterceptor()
+	assert.NoError(t, err)
+
+	testInterceptor, err := f.NewInterceptor("")
+	assert.NoError(t, err)
+	defer func() {
+		assert.NoError(t, testInterceptor.Close())
+	}()
+
+	hdr := rtp.Header{Version: 2}
+	tcc, err := (&rtp.TransportCCExtension{TransportSequence: 1}).Marshal()
+	assert.NoError(t, err)
+	assert.NoError(t, hdr.SetExtension(1, tcc))
+	raw, err := (&rtp.Packet{Header: hdr, Payload: []byte{}}).Marshal()
+	assert.NoError(t, err)
+
+	reader := testInterceptor.BindRemoteStream(&interceptor.StreamInfo{
+		SSRC: 1,
+		RTPHeaderExtensions: []interceptor.RTPHeaderExtension{
+			{URI: transportCCURI, ID: 1},
+		},
+	}, interceptor.RTPReaderFunc(
+		func(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) {
+			return copy(b, raw), a, nil
+		},
+	))
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for range 2 * packetChanBufferSize {
+			_, _, err := reader.Read(make([]byte, 1500), nil)
+			assert.NoError(t, err)
+		}
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		assert.Fail(t, "read blocked without a bound RTCP writer")
+	}
+}