interceptor POC

api.go

@@ -3,6 +3,7 @@
 package webrtc
 
 import (
+	"github.com/pion/interceptor"
 	"github.com/pion/logging"
 )
 
@@ -13,6 +14,7 @@ import (
 type API struct {
 	settingEngine *SettingEngine
 	mediaEngine   *MediaEngine
+	interceptor   interceptor.Interceptor
 }
 
 // NewAPI Creates a new API object for keeping semi-global settings to WebRTC objects
@@ -35,6 +37,10 @@ func NewAPI(options ...func(*API)) *API {
 		a.mediaEngine = &MediaEngine{}
 	}
 
+	if a.interceptor == nil {
+		a.interceptor = &interceptor.NoOp{}
+	}
+
 	return a
 }
 
@@ -57,3 +63,11 @@ func WithSettingEngine(s SettingEngine) func(a *API) {
 		a.settingEngine = &s
 	}
 }
+
+// WithInterceptorRegistry allows providing Interceptors to the API.
+// Settings should not be changed after passing the registry to an API.
+func WithInterceptorRegistry(interceptorRegistry *interceptor.Registry) func(a *API) {
+	return func(a *API) {
+		a.interceptor = interceptorRegistry.Build()
+	}
+}

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/pion/datachannel v1.4.21
 	github.com/pion/dtls/v2 v2.0.3
 	github.com/pion/ice/v2 v2.0.11
+	github.com/pion/interceptor v0.0.3
 	github.com/pion/logging v0.2.2
 	github.com/pion/quic v0.1.4
 	github.com/pion/randutil v0.1.0

go.sum

@@ -106,6 +106,8 @@ github.com/pion/dtls/v2 v2.0.3 h1:3qQ0s4+TXD00rsllL8g8KQcxAs+Y/Z6oz618RXX6p14=
 github.com/pion/dtls/v2 v2.0.3/go.mod h1:TUjyL8bf8LH95h81Xj7kATmzMRt29F/4lxpIPj2Xe4Y=
 github.com/pion/ice/v2 v2.0.11 h1:XAKZPtglESY/w18eHJR8YLWx3sS8ms7LbWvtOpEM1Dg=
 github.com/pion/ice/v2 v2.0.11/go.mod h1:Sqdo0oy3ZkaOCsK7Ai9ksLpJkREG03R3fHMJ0PXmHO8=
+github.com/pion/interceptor v0.0.3 h1:VQtmPts/2IgYQtb9sZLTp6B0kIdHE5zBMQ6tCcgdJcM=
+github.com/pion/interceptor v0.0.3/go.mod h1:lPVrf5xfosI989ZcmgPS4WwwRhd+XAyTFaYI2wHf7nU=
 github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY=
 github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms=
 github.com/pion/mdns v0.0.4 h1:O4vvVqr4DGX63vzmO6Fw9vpy3lfztVWHGCQfyw0ZLSY=

interceptor.go

@@ -0,0 +1,26 @@
+// +build !js
+
+package webrtc
+
+import (
+	"github.com/pion/interceptor"
+)
+
+// RegisterDefaultInterceptors will register some useful interceptors. If you want to customize which interceptors are loaded,
+// you should copy the code from this method and remove unwanted interceptors.
+func RegisterDefaultInterceptors(mediaEngine *MediaEngine, interceptorRegistry *interceptor.Registry) error {
+	err := ConfigureNack(mediaEngine, interceptorRegistry)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// ConfigureNack will setup everything necessary for handling generating/responding to nack messages.
+func ConfigureNack(mediaEngine *MediaEngine, interceptorRegistry *interceptor.Registry) error {
+	mediaEngine.RegisterFeedback(RTCPFeedback{Type: "nack"}, RTPCodecTypeVideo)
+	mediaEngine.RegisterFeedback(RTCPFeedback{Type: "nack", Parameter: "pli"}, RTPCodecTypeVideo)
+	interceptorRegistry.Add(&interceptor.NACK{})
+	return nil
+}

interceptor_test.go

@@ -0,0 +1,188 @@
+// +build !js
+
+package webrtc
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/pion/interceptor"
+	"github.com/pion/rtcp"
+	"github.com/pion/rtp"
+	"github.com/pion/transport/test"
+	"github.com/pion/webrtc/v3/pkg/media"
+	"github.com/stretchr/testify/assert"
+)
+
+type testInterceptor struct {
+	t           *testing.T
+	extensionID uint8
+	rtcpWriter  atomic.Value
+	lastRTCP    atomic.Value
+	interceptor.NoOp
+}
+
+func (t *testInterceptor) BindLocalStream(_ *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter {
+	return interceptor.RTPWriterFunc(func(p *rtp.Packet, attributes interceptor.Attributes) (int, error) {
+		// set extension on outgoing packet
+		p.Header.Extension = true
+		p.Header.ExtensionProfile = 0xBEDE
+		assert.NoError(t.t, p.Header.SetExtension(t.extensionID, []byte("write")))
+
+		return writer.Write(p, attributes)
+	})
+}
+
+func (t *testInterceptor) BindRemoteStream(info *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader {
+	return interceptor.RTPReaderFunc(func() (*rtp.Packet, interceptor.Attributes, error) {
+		p, attributes, err := reader.Read()
+		if err != nil {
+			return nil, nil, err
+		}
+		// set extension on incoming packet
+		p.Header.Extension = true
+		p.Header.ExtensionProfile = 0xBEDE
+		assert.NoError(t.t, p.Header.SetExtension(t.extensionID, []byte("read")))
+
+		// write back a pli
+		rtcpWriter := t.rtcpWriter.Load().(interceptor.RTCPWriter)
+		pli := &rtcp.PictureLossIndication{SenderSSRC: info.SSRC, MediaSSRC: info.SSRC}
+		_, err = rtcpWriter.Write([]rtcp.Packet{pli}, make(interceptor.Attributes))
+		assert.NoError(t.t, err)
+
+		return p, attributes, nil
+	})
+}
+
+func (t *testInterceptor) BindRTCPReader(reader interceptor.RTCPReader) interceptor.RTCPReader {
+	return interceptor.RTCPReaderFunc(func() ([]rtcp.Packet, interceptor.Attributes, error) {
+		pkts, attributes, err := reader.Read()
+		if err != nil {
+			return nil, nil, err
+		}
+
+		t.lastRTCP.Store(pkts[0])
+
+		return pkts, attributes, nil
+	})
+}
+
+func (t *testInterceptor) lastReadRTCP() rtcp.Packet {
+	p, _ := t.lastRTCP.Load().(rtcp.Packet)
+	return p
+}
+
+func (t *testInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter {
+	t.rtcpWriter.Store(writer)
+	return writer
+}
+
+func TestPeerConnection_Interceptor(t *testing.T) {
+	to := test.TimeOut(time.Second * 20)
+	defer to.Stop()
+
+	report := test.CheckRoutines(t)
+	defer report()
+
+	createPC := func(i interceptor.Interceptor) *PeerConnection {
+		m := &MediaEngine{}
+		err := m.RegisterDefaultCodecs()
+		if err != nil {
+			t.Fatal(err)
+		}
+		ir := &interceptor.Registry{}
+		ir.Add(i)
+		pc, err := NewAPI(WithMediaEngine(m), WithInterceptorRegistry(ir)).NewPeerConnection(Configuration{})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		return pc
+	}
+
+	sendInterceptor := &testInterceptor{t: t, extensionID: 1}
+	senderPC := createPC(sendInterceptor)
+	receiverPC := createPC(&testInterceptor{t: t, extensionID: 2})
+
+	track, err := NewTrackLocalStaticSample(RTPCodecCapability{MimeType: "video/vp8"}, "video", "pion")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sender, err := senderPC.AddTrack(track)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pending := new(int32)
+	wg := &sync.WaitGroup{}
+
+	wg.Add(1)
+	*pending++
+	receiverPC.OnTrack(func(track *TrackRemote, receiver *RTPReceiver) {
+		p, readErr := track.ReadRTP()
+		if readErr != nil {
+			t.Fatal(readErr)
+		}
+		assert.Equal(t, p.Extension, true)
+		assert.Equal(t, "write", string(p.GetExtension(1)))
+		assert.Equal(t, "read", string(p.GetExtension(2)))
+		atomic.AddInt32(pending, -1)
+		wg.Done()
+
+		for {
+			_, readErr = track.ReadRTP()
+			if readErr != nil {
+				return
+			}
+		}
+	})
+
+	wg.Add(1)
+	*pending++
+	go func() {
+		_, readErr := sender.ReadRTCP()
+		assert.NoError(t, readErr)
+		atomic.AddInt32(pending, -1)
+		wg.Done()
+
+		for {
+			_, readErr = sender.ReadRTCP()
+			if readErr != nil {
+				return
+			}
+		}
+	}()
+
+	err = signalPair(senderPC, receiverPC)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			time.Sleep(time.Millisecond * 100)
+			if routineErr := track.WriteSample(media.Sample{Data: []byte{0x00}, Duration: time.Second}); routineErr != nil {
+				t.Error(routineErr)
+				return
+			}
+
+			if atomic.LoadInt32(pending) == 0 {
+				return
+			}
+		}
+	}()
+
+	wg.Wait()
+	assert.NoError(t, senderPC.Close())
+	assert.NoError(t, receiverPC.Close())
+
+	pli, _ := sendInterceptor.lastReadRTCP().(*rtcp.PictureLossIndication)
+	if pli == nil || pli.SenderSSRC == 0 {
+		t.Errorf("pli not found by send interceptor")
+	}
+}

interceptor_track_local.go

@@ -0,0 +1,29 @@
+// +build !js
+
+package webrtc
+
+import (
+	"sync/atomic"
+
+	"github.com/pion/interceptor"
+	"github.com/pion/rtp"
+)
+
+type interceptorTrackLocalWriter struct {
+	TrackLocalWriter
+	rtpWriter atomic.Value
+}
+
+func (i *interceptorTrackLocalWriter) setRTPWriter(writer interceptor.RTPWriter) {
+	i.rtpWriter.Store(writer)
+}
+
+func (i *interceptorTrackLocalWriter) WriteRTP(header *rtp.Header, payload []byte) (int, error) {
+	writer := i.rtpWriter.Load().(interceptor.RTPWriter)
+
+	if writer == nil {
+		return 0, nil
+	}
+
+	return writer.Write(&rtp.Packet{Header: *header, Payload: payload}, make(interceptor.Attributes))
+}