feat: Add playout delay header interceptor

AUTHORS.txt

@@ -15,5 +15,6 @@ boks1971 <raja.gobi@tutanota.com>
 David Zhao <david@davidzhao.com>
 Jonathan Müller <jonathan@fotokite.com>
 Kevin Caffrey <kcaffrey@gmail.com>
+Kevin Wang <kevmo314@gmail.com>
 Mathis Engelbart <mathis.engelbart@gmail.com>
 Sean DuBois <sean@siobud.com>

pkg/playoutdelay/header_extension_interceptor.go

@@ -0,0 +1,65 @@
+package playoutdelay
+
+import (
+	"time"
+
+	"github.com/pion/interceptor"
+	"github.com/pion/rtp"
+)
+
+// HeaderExtensionInterceptorFactory is a interceptor.Factory for a HeaderExtensionInterceptor
+type HeaderExtensionInterceptorFactory struct{}
+
+const (
+	playoutDelayMaxMs = 40950
+)
+
+// NewInterceptor constructs a new HeaderExtensionInterceptor
+func (h *HeaderExtensionInterceptorFactory) NewInterceptor(id string, minDelay, maxDelay time.Duration) (interceptor.Interceptor, error) {
+	if minDelay.Milliseconds() < 0 || minDelay.Milliseconds() > playoutDelayMaxMs || maxDelay.Milliseconds() < 0 || maxDelay.Milliseconds() > playoutDelayMaxMs {
+		return nil, errPlayoutDelayInvalidValue
+	}
+	return &HeaderExtensionInterceptor{minDelay: uint16(minDelay.Milliseconds() / 10), maxDelay: uint16(maxDelay.Milliseconds() / 10)}, nil
+}
+
+// NewHeaderExtensionInterceptor returns a HeaderExtensionInterceptorFactory
+func NewHeaderExtensionInterceptor() (*HeaderExtensionInterceptorFactory, error) {
+	return &HeaderExtensionInterceptorFactory{}, nil
+}
+
+// HeaderExtensionInterceptor adds transport wide sequence numbers as header extension to each RTP packet
+type HeaderExtensionInterceptor struct {
+	interceptor.NoOp
+	minDelay, maxDelay uint16
+}
+
+const playoutDelayURI = "http://www.webrtc.org/experiments/rtp-hdrext/playout-delay"
+
+// BindLocalStream returns a writer that adds a rtp.PlayoutDelayExtension
+// header with increasing sequence numbers to each outgoing packet.
+func (h *HeaderExtensionInterceptor) BindLocalStream(info *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter {
+	var hdrExtID uint8
+	for _, e := range info.RTPHeaderExtensions {
+		if e.URI == playoutDelayURI {
+			hdrExtID = uint8(e.ID)
+			break
+		}
+	}
+	if hdrExtID == 0 { // Don't add header extension if ID is 0, because 0 is an invalid extension ID
+		return writer
+	}
+	return interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
+		ext, err := (&rtp.PlayoutDelayExtension{
+			minDelay: h.minDelay,
+			maxDelay: h.maxDelay,
+		}).Marshal()
+		if err != nil {
+			return 0, err
+		}
+		err = header.SetExtension(hdrExtID, ext)
+		if err != nil {
+			return 0, err
+		}
+		return writer.Write(header, payload, attributes)
+	})
+}

pkg/playoutdelay/header_extension_interceptor_test.go

@@ -0,0 +1,66 @@
+package playoutdelay
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/pion/interceptor"
+	"github.com/pion/interceptor/internal/test"
+	"github.com/pion/rtp"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestHeaderExtensionInterceptor(t *testing.T) {
+	t.Run("add playout delay to each packet", func(t *testing.T) {
+		factory, err := NewHeaderExtensionInterceptor()
+		assert.NoError(t, err)
+
+		inter, err := factory.NewInterceptor("", 10*time.Millisecond, 20*time.Millisecond)
+		assert.NoError(t, err)
+
+		pChan := make(chan *rtp.Packet, 10*5)
+		go func() {
+			// start some parallel streams using the same interceptor to test for race conditions
+			var wg sync.WaitGroup
+			num := 10
+			wg.Add(num)
+			for i := 0; i < num; i++ {
+				go func(ch chan *rtp.Packet, id uint16) {
+					stream := test.NewMockStream(&interceptor.StreamInfo{RTPHeaderExtensions: []interceptor.RTPHeaderExtension{
+						{
+							URI: playoutDelayURI,
+							ID:  1,
+						},
+					}}, inter)
+					defer func() {
+						wg.Done()
+						assert.NoError(t, stream.Close())
+					}()
+
+					for _, seqNum := range []uint16{id * 1, id * 2, id * 3, id * 4, id * 5} {
+						assert.NoError(t, stream.WriteRTP(&rtp.Packet{Header: rtp.Header{SequenceNumber: seqNum}}))
+						select {
+						case p := <-stream.WrittenRTP():
+							assert.Equal(t, seqNum, p.SequenceNumber)
+							ch <- p
+						case <-time.After(10 * time.Millisecond):
+							panic("written rtp packet not found")
+						}
+					}
+				}(pChan, uint16(i+1))
+			}
+			wg.Wait()
+			close(pChan)
+		}()
+
+		for p := range pChan {
+			extensionHeader := p.GetExtension(1)
+			ext := &rtp.PlayoutDelayExtension{}
+			err = ext.Unmarshal(extensionHeader)
+			assert.NoError(t, err)
+			assert.Equal(t, uint16(1), ext.minDelay)
+			assert.Equal(t, uint16(2), ext.maxDelay)
+		}
+	})
+}

pkg/playoutdelay/playout_delay.go

@@ -0,0 +1,10 @@
+// Package playoutdelay implements the playout delay header extension.
+package playoutdelay
+
+import (
+	"errors"
+)
+
+var (
+	errPlayoutDelayInvalidValue = errors.New("invalid playout delay value")
+)