From 7744acf7e5c34bd7acd887b847bf4f0b72a0fd8a Mon Sep 17 00:00:00 2001 From: Adam Kiss Date: Fri, 20 Nov 2020 14:05:33 +0100 Subject: [PATCH] Implement Interceptors Provide API so that handling around RTP can be easily defined by the user. See the design doc here[0] [0] https://github.com/pion/webrtc-v3-design/issues/34 --- api.go | 14 +++ go.mod | 1 + go.sum | 2 + interceptor.go | 26 +++++ interceptor_test.go | 188 +++++++++++++++++++++++++++++++++++++ interceptor_track_local.go | 29 ++++++ mediaengine.go | 63 +++++++++++-- mediaengine_test.go | 10 +- peerconnection.go | 38 +++++--- rtpcodec.go | 16 ++++ rtpreceiver.go | 56 +++++++---- rtpsender.go | 89 +++++++++++++----- track_local.go | 13 ++- track_remote.go | 54 ++++++++++- 14 files changed, 528 insertions(+), 71 deletions(-) create mode 100644 interceptor.go create mode 100644 interceptor_test.go create mode 100644 interceptor_track_local.go diff --git a/api.go b/api.go index c6010eb526f..60ac72809d9 100644 --- a/api.go +++ b/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() + } +} diff --git a/go.mod b/go.mod index 1a2d4828878..d6d6cb6c95a 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 5b4070a9949..1105cd3c9fb 100644 --- a/go.sum +++ b/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= diff --git a/interceptor.go b/interceptor.go new file mode 100644 index 00000000000..fd886707ca4 --- /dev/null +++ b/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 +} diff --git a/interceptor_test.go b/interceptor_test.go new file mode 100644 index 00000000000..47e853a1afd --- /dev/null +++ b/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") + } +} diff --git a/interceptor_track_local.go b/interceptor_track_local.go new file mode 100644 index 00000000000..914a3c92d07 --- /dev/null +++ b/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)) +} diff --git a/mediaengine.go b/mediaengine.go index 45d678493f7..5dac7cb5d59 100644 --- a/mediaengine.go +++ b/mediaengine.go @@ -233,6 +233,22 @@ func (m *MediaEngine) RegisterHeaderExtension(extension RTPHeaderExtensionCapabi return nil } +// RegisterFeedback adds feedback mechanism to already registered codecs. +func (m *MediaEngine) RegisterFeedback(feedback RTCPFeedback, typ RTPCodecType) { + switch typ { + case RTPCodecTypeVideo: + for i, v := range m.videoCodecs { + v.RTCPFeedback = append(v.RTCPFeedback, feedback) + m.videoCodecs[i] = v + } + case RTPCodecTypeAudio: + for i, v := range m.audioCodecs { + v.RTCPFeedback = append(v.RTCPFeedback, feedback) + m.audioCodecs[i] = v + } + } +} + // GetHeaderExtensionID returns the negotiated ID for a header extension. // If the Header Extension isn't enabled ok will be false func (m *MediaEngine) GetHeaderExtensionID(extension RTPHeaderExtensionCapability) (val int, audioNegotiated, videoNegotiated bool) { @@ -249,19 +265,19 @@ func (m *MediaEngine) GetHeaderExtensionID(extension RTPHeaderExtensionCapabilit return } -func (m *MediaEngine) getCodecByPayload(payloadType PayloadType) (RTPCodecParameters, error) { +func (m *MediaEngine) getCodecByPayload(payloadType PayloadType) (RTPCodecParameters, RTPCodecType, error) { for _, codec := range m.negotiatedVideoCodecs { if codec.PayloadType == payloadType { - return codec, nil + return codec, RTPCodecTypeVideo, nil } } for _, codec := range m.negotiatedAudioCodecs { if codec.PayloadType == payloadType { - return codec, nil + return codec, RTPCodecTypeAudio, nil } } - return RTPCodecParameters{}, ErrCodecNotFound + return RTPCodecParameters{}, 0, ErrCodecNotFound } func (m *MediaEngine) collectStats(collector *statsReportCollector) { @@ -309,7 +325,7 @@ func (m *MediaEngine) updateCodecParameters(remoteCodec RTPCodecParameters, typ return err } - if _, err = m.getCodecByPayload(PayloadType(payloadType)); err != nil { + if _, _, err = m.getCodecByPayload(PayloadType(payloadType)); err != nil { return nil // not an error, we just ignore this codec we don't support } } @@ -378,8 +394,8 @@ func (m *MediaEngine) updateFromRemoteDescription(desc sdp.SessionDescription) e return err } - for id, extension := range extensions { - if err = m.updateHeaderExtension(extension, id, typ); err != nil { + for extension, id := range extensions { + if err = m.updateHeaderExtension(id, extension, typ); err != nil { return err } } @@ -405,6 +421,39 @@ func (m *MediaEngine) getCodecsByKind(typ RTPCodecType) []RTPCodecParameters { return nil } +func (m *MediaEngine) getRTPParametersByKind(typ RTPCodecType) RTPParameters { + headerExtensions := make([]RTPHeaderExtensionParameter, 0) + for id, e := range m.negotiatedHeaderExtensions { + if e.isAudio && typ == RTPCodecTypeAudio || e.isVideo && typ == RTPCodecTypeVideo { + headerExtensions = append(headerExtensions, RTPHeaderExtensionParameter{ID: id, URI: e.uri}) + } + } + + return RTPParameters{ + HeaderExtensions: headerExtensions, + Codecs: m.getCodecsByKind(typ), + } +} + +func (m *MediaEngine) getRTPParametersByPayloadType(payloadType PayloadType) (RTPParameters, error) { + codec, typ, err := m.getCodecByPayload(payloadType) + if err != nil { + return RTPParameters{}, err + } + + headerExtensions := make([]RTPHeaderExtensionParameter, 0) + for id, e := range m.negotiatedHeaderExtensions { + if e.isAudio && typ == RTPCodecTypeAudio || e.isVideo && typ == RTPCodecTypeVideo { + headerExtensions = append(headerExtensions, RTPHeaderExtensionParameter{ID: id, URI: e.uri}) + } + } + + return RTPParameters{ + HeaderExtensions: headerExtensions, + Codecs: []RTPCodecParameters{codec}, + }, nil +} + func (m *MediaEngine) negotiatedHeaderExtensionsForType(typ RTPCodecType) map[int]mediaEngineHeaderExtension { headerExtensions := map[int]mediaEngineHeaderExtension{} for id, e := range m.negotiatedHeaderExtensions { diff --git a/mediaengine_test.go b/mediaengine_test.go index f3312e876be..b857b2e7549 100644 --- a/mediaengine_test.go +++ b/mediaengine_test.go @@ -63,7 +63,7 @@ a=fmtp:111 minptime=10; useinbandfec=1 assert.False(t, m.negotiatedVideo) assert.True(t, m.negotiatedAudio) - opusCodec, err := m.getCodecByPayload(111) + opusCodec, _, err := m.getCodecByPayload(111) assert.NoError(t, err) assert.Equal(t, opusCodec.MimeType, mimeTypeOpus) }) @@ -85,10 +85,10 @@ a=fmtp:112 minptime=10; useinbandfec=1 assert.False(t, m.negotiatedVideo) assert.True(t, m.negotiatedAudio) - _, err := m.getCodecByPayload(111) + _, _, err := m.getCodecByPayload(111) assert.Error(t, err) - opusCodec, err := m.getCodecByPayload(112) + opusCodec, _, err := m.getCodecByPayload(112) assert.NoError(t, err) assert.Equal(t, opusCodec.MimeType, mimeTypeOpus) }) @@ -110,7 +110,7 @@ a=fmtp:111 minptime=10; useinbandfec=1 assert.False(t, m.negotiatedVideo) assert.True(t, m.negotiatedAudio) - opusCodec, err := m.getCodecByPayload(111) + opusCodec, _, err := m.getCodecByPayload(111) assert.NoError(t, err) assert.Equal(t, opusCodec.MimeType, "audio/OPUS") }) @@ -131,7 +131,7 @@ a=rtpmap:111 opus/48000/2 assert.False(t, m.negotiatedVideo) assert.True(t, m.negotiatedAudio) - opusCodec, err := m.getCodecByPayload(111) + opusCodec, _, err := m.getCodecByPayload(111) assert.NoError(t, err) assert.Equal(t, opusCodec.MimeType, mimeTypeOpus) }) diff --git a/peerconnection.go b/peerconnection.go index aee1360a4b7..eb89abb7eb9 100644 --- a/peerconnection.go +++ b/peerconnection.go @@ -15,6 +15,7 @@ import ( "time" "github.com/pion/ice/v2" + "github.com/pion/interceptor" "github.com/pion/logging" "github.com/pion/rtcp" "github.com/pion/sdp/v3" @@ -76,6 +77,8 @@ type PeerConnection struct { // A reference to the associated API state used by this connection api *API log logging.LeveledLogger + + interceptorRTCPWriter interceptor.RTCPWriter } // NewPeerConnection creates a peerconnection with the default @@ -119,6 +122,8 @@ func (api *API) NewPeerConnection(configuration Configuration) (*PeerConnection, log: api.settingEngine.LoggerFactory.NewLogger("pc"), } + pc.interceptorRTCPWriter = api.interceptor.BindRTCPWriter(interceptor.RTCPWriterFunc(pc.writeRTCP)) + var err error if err = pc.initConfiguration(configuration); err != nil { return nil, err @@ -1125,7 +1130,7 @@ func (pc *PeerConnection) startReceiver(incoming trackDetails, receiver *RTPRece return } - codec, err := pc.api.mediaEngine.getCodecByPayload(receiver.Track().PayloadType()) + params, err := pc.api.mediaEngine.getRTPParametersByPayloadType(receiver.Track().PayloadType()) if err != nil { pc.log.Warnf("no codec could be found for payloadType %d", receiver.Track().PayloadType()) return @@ -1133,7 +1138,9 @@ func (pc *PeerConnection) startReceiver(incoming trackDetails, receiver *RTPRece receiver.Track().mu.Lock() receiver.Track().kind = receiver.kind - receiver.Track().codec = codec + receiver.Track().codec = params.Codecs[0] + receiver.Track().params = params + receiver.Track().bindInterceptor() receiver.Track().mu.Unlock() pc.onTrack(receiver.Track(), receiver) @@ -1335,7 +1342,7 @@ func (pc *PeerConnection) handleUndeclaredSSRC(rtpStream io.Reader, ssrc SSRC) e continue } - codec, err := pc.api.mediaEngine.getCodecByPayload(payloadType) + params, err := pc.api.mediaEngine.getRTPParametersByPayloadType(payloadType) if err != nil { return err } @@ -1345,7 +1352,7 @@ func (pc *PeerConnection) handleUndeclaredSSRC(rtpStream io.Reader, ssrc SSRC) e continue } - track, err := t.Receiver().receiveForRid(rid, codec, ssrc) + track, err := t.Receiver().receiveForRid(rid, params, ssrc) if err != nil { return err } @@ -1730,28 +1737,33 @@ func (pc *PeerConnection) SetIdentityProvider(provider string) error { return errPeerConnSetIdentityProviderNotImplemented } -// WriteRTCP sends a user provided RTCP packet to the connected peer -// If no peer is connected the packet is discarded +// WriteRTCP sends a user provided RTCP packet to the connected peer. If no peer is connected the +// packet is discarded. It also runs any configured interceptors. func (pc *PeerConnection) WriteRTCP(pkts []rtcp.Packet) error { + _, err := pc.interceptorRTCPWriter.Write(pkts, make(interceptor.Attributes)) + return err +} + +func (pc *PeerConnection) writeRTCP(pkts []rtcp.Packet, _ interceptor.Attributes) (int, error) { raw, err := rtcp.Marshal(pkts) if err != nil { - return err + return 0, err } srtcpSession, err := pc.dtlsTransport.getSRTCPSession() if err != nil { - return nil + return 0, nil } writeStream, err := srtcpSession.OpenWriteStream() if err != nil { - return fmt.Errorf("%w: %v", errPeerConnWriteRTCPOpenWriteStream, err) + return 0, fmt.Errorf("%w: %v", errPeerConnWriteRTCPOpenWriteStream, err) } - if _, err := writeStream.Write(raw); err != nil { - return err + if n, err := writeStream.Write(raw); err != nil { + return n, err } - return nil + return 0, nil } // Close ends the PeerConnection @@ -1775,6 +1787,8 @@ func (pc *PeerConnection) Close() error { // continue the chain the Mux has to be closed. closeErrs := make([]error, 4) + closeErrs = append(closeErrs, pc.api.interceptor.Close()) + // https://www.w3.org/TR/webrtc/#dom-rtcpeerconnection-close (step #4) for _, t := range pc.GetTransceivers() { if !t.stopped { diff --git a/rtpcodec.go b/rtpcodec.go index c03abd13d86..2d9b6d7600a 100644 --- a/rtpcodec.go +++ b/rtpcodec.go @@ -57,6 +57,14 @@ type RTPHeaderExtensionCapability struct { URI string } +// RTPHeaderExtensionParameter represents a negotiated RFC5285 RTP header extension. +// +// https://w3c.github.io/webrtc-pc/#dictionary-rtcrtpheaderextensionparameters-members +type RTPHeaderExtensionParameter struct { + URI string + ID int +} + // RTPCodecParameters is a sequence containing the media codecs that an RtpSender // will choose from, as well as entries for RTX, RED and FEC mechanisms. This also // includes the PayloadType that has been negotiated @@ -77,6 +85,14 @@ type RTCRtpCapabilities struct { Codecs []RTPCodecCapability } +// RTPParameters is a list of negotiated codecs and header extensions +// +// https://w3c.github.io/webrtc-pc/#dictionary-rtcrtpparameters-members +type RTPParameters struct { + HeaderExtensions []RTPHeaderExtensionParameter + Codecs []RTPCodecParameters +} + // Do a fuzzy find for a codec in the list of codecs // Used for lookup up a codec in an existing list to find a match func codecParametersFuzzySearch(needle RTPCodecParameters, haystack []RTPCodecParameters) (RTPCodecParameters, error) { diff --git a/rtpreceiver.go b/rtpreceiver.go index 1b37f624aa8..8ae5469d909 100644 --- a/rtpreceiver.go +++ b/rtpreceiver.go @@ -7,6 +7,7 @@ import ( "io" "sync" + "github.com/pion/interceptor" "github.com/pion/rtcp" "github.com/pion/srtp" ) @@ -31,6 +32,8 @@ type RTPReceiver struct { // A reference to the associated api object api *API + + interceptorRTCPReader interceptor.RTCPReader } // NewRTPReceiver constructs a new RTPReceiver @@ -39,14 +42,17 @@ func (api *API) NewRTPReceiver(kind RTPCodecType, transport *DTLSTransport) (*RT return nil, errRTPReceiverDTLSTransportNil } - return &RTPReceiver{ + r := &RTPReceiver{ kind: kind, transport: transport, api: api, closed: make(chan interface{}), received: make(chan interface{}), tracks: []trackStreams{}, - }, nil + } + r.interceptorRTCPReader = api.interceptor.BindRTCPReader(interceptor.RTCPReaderFunc(r.readRTCP)) + + return r, nil } // Transport returns the currently-configured *DTLSTransport or nil @@ -94,11 +100,12 @@ func (r *RTPReceiver) Receive(parameters RTPReceiveParameters) error { if len(parameters.Encodings) == 1 && parameters.Encodings[0].SSRC != 0 { t := trackStreams{ - track: &TrackRemote{ - kind: r.kind, - ssrc: parameters.Encodings[0].SSRC, - receiver: r, - }, + track: newTrackRemote( + r.kind, + parameters.Encodings[0].SSRC, + "", + r, + ), } var err error @@ -111,11 +118,12 @@ func (r *RTPReceiver) Receive(parameters RTPReceiveParameters) error { } else { for _, encoding := range parameters.Encodings { r.tracks = append(r.tracks, trackStreams{ - track: &TrackRemote{ - kind: r.kind, - rid: encoding.RID, - receiver: r, - }, + track: newTrackRemote( + r.kind, + 0, + encoding.RID, + r, + ), }) } } @@ -148,15 +156,27 @@ func (r *RTPReceiver) ReadSimulcast(b []byte, rid string) (n int, err error) { } } -// ReadRTCP is a convenience method that wraps Read and unmarshal for you +// ReadRTCP is a convenience method that wraps Read and unmarshal for you. +// It also runs any configured interceptors. func (r *RTPReceiver) ReadRTCP() ([]rtcp.Packet, error) { + pkts, _, err := r.interceptorRTCPReader.Read() + return pkts, err +} + +// ReadRTCP is a convenience method that wraps Read and unmarshal for you +func (r *RTPReceiver) readRTCP() ([]rtcp.Packet, interceptor.Attributes, error) { b := make([]byte, receiveMTU) i, err := r.Read(b) if err != nil { - return nil, err + return nil, nil, err } - return rtcp.Unmarshal(b[:i]) + pkts, err := rtcp.Unmarshal(b[:i]) + if err != nil { + return nil, nil, err + } + + return pkts, make(interceptor.Attributes), nil } // ReadSimulcastRTCP is a convenience method that wraps ReadSimulcast and unmarshal for you @@ -232,7 +252,7 @@ func (r *RTPReceiver) readRTP(b []byte, reader *TrackRemote) (n int, err error) // receiveForRid is the sibling of Receive expect for RIDs instead of SSRCs // It populates all the internal state for the given RID -func (r *RTPReceiver) receiveForRid(rid string, codec RTPCodecParameters, ssrc SSRC) (*TrackRemote, error) { +func (r *RTPReceiver) receiveForRid(rid string, params RTPParameters, ssrc SSRC) (*TrackRemote, error) { r.mu.Lock() defer r.mu.Unlock() @@ -240,8 +260,10 @@ func (r *RTPReceiver) receiveForRid(rid string, codec RTPCodecParameters, ssrc S if r.tracks[i].track.RID() == rid { r.tracks[i].track.mu.Lock() r.tracks[i].track.kind = r.kind - r.tracks[i].track.codec = codec + r.tracks[i].track.codec = params.Codecs[0] + r.tracks[i].track.params = params r.tracks[i].track.ssrc = ssrc + r.tracks[i].track.bindInterceptor() r.tracks[i].track.mu.Unlock() var err error diff --git a/rtpsender.go b/rtpsender.go index 8cfe97d30a1..e2aaf1aee70 100644 --- a/rtpsender.go +++ b/rtpsender.go @@ -6,8 +6,10 @@ import ( "io" "sync" + "github.com/pion/interceptor" "github.com/pion/randutil" "github.com/pion/rtcp" + "github.com/pion/rtp" "github.com/pion/srtp" ) @@ -16,13 +18,12 @@ type RTPSender struct { track TrackLocal rtcpReadStream *srtp.ReadStreamSRTCP - rtpWriteStream *srtp.WriteStreamSRTP + context TrackLocalContext transport *DTLSTransport payloadType PayloadType ssrc SSRC - codec RTPCodecParameters // nolint:godox // TODO(sgotti) remove this when in future we'll avoid replacing @@ -36,6 +37,8 @@ type RTPSender struct { mu sync.RWMutex sendCalled, stopCalled chan interface{} + + interceptorRTCPReader interceptor.RTCPReader } // NewRTPSender constructs a new RTPSender @@ -51,7 +54,7 @@ func (api *API) NewRTPSender(track TrackLocal, transport *DTLSTransport) (*RTPSe return nil, err } - return &RTPSender{ + r := &RTPSender{ track: track, transport: transport, api: api, @@ -59,7 +62,10 @@ func (api *API) NewRTPSender(track TrackLocal, transport *DTLSTransport) (*RTPSe stopCalled: make(chan interface{}), ssrc: SSRC(randutil.NewMathRandomGenerator().Uint32()), id: id, - }, nil + } + r.interceptorRTCPReader = api.interceptor.BindRTCPReader(interceptor.RTCPReaderFunc(r.readRTCP)) + + return r, nil } func (r *RTPSender) isNegotiated() bool { @@ -97,11 +103,7 @@ func (r *RTPSender) ReplaceTrack(track TrackLocal) error { defer r.mu.Unlock() if r.hasSent() { - if err := r.track.Unbind(TrackLocalContext{ - id: r.id, - ssrc: r.ssrc, - writeStream: r.rtpWriteStream, - }); err != nil { + if err := r.track.Unbind(r.context); err != nil { return err } } @@ -111,12 +113,7 @@ func (r *RTPSender) ReplaceTrack(track TrackLocal) error { return nil } - if _, err := track.Bind(TrackLocalContext{ - id: r.id, - codecs: []RTPCodecParameters{r.codec}, - ssrc: r.ssrc, - writeStream: r.rtpWriteStream, - }); err != nil { + if _, err := track.Bind(r.context); err != nil { return err } @@ -148,18 +145,53 @@ func (r *RTPSender) Send(parameters RTPSendParameters) error { return err } - if r.rtpWriteStream, err = srtpSession.OpenWriteStream(); err != nil { + rtpWriteStream, err := srtpSession.OpenWriteStream() + if err != nil { return err } - if r.codec, err = r.track.Bind(TrackLocalContext{ + writeStream := &interceptorTrackLocalWriter{TrackLocalWriter: rtpWriteStream} + + r.context = TrackLocalContext{ id: r.id, - codecs: r.api.mediaEngine.getCodecsByKind(r.track.Kind()), + params: r.api.mediaEngine.getRTPParametersByKind(r.track.Kind()), ssrc: parameters.Encodings.SSRC, - writeStream: r.rtpWriteStream, - }); err != nil { + writeStream: writeStream, + } + + codec, err := r.track.Bind(r.context) + if err != nil { return err } + r.context.params.Codecs = []RTPCodecParameters{codec} + + headerExtensions := make([]interceptor.RTPHeaderExtension, 0, len(r.context.params.HeaderExtensions)) + for _, h := range r.context.params.HeaderExtensions { + headerExtensions = append(headerExtensions, interceptor.RTPHeaderExtension{ID: h.ID, URI: h.URI}) + } + feedbacks := make([]interceptor.RTCPFeedback, 0, len(codec.RTCPFeedback)) + for _, f := range codec.RTCPFeedback { + feedbacks = append(feedbacks, interceptor.RTCPFeedback{Type: f.Type, Parameter: f.Parameter}) + } + info := &interceptor.StreamInfo{ + ID: r.context.id, + Attributes: interceptor.Attributes{}, + SSRC: uint32(r.context.ssrc), + PayloadType: uint8(codec.PayloadType), + RTPHeaderExtensions: headerExtensions, + MimeType: codec.MimeType, + ClockRate: codec.ClockRate, + Channels: codec.Channels, + SDPFmtpLine: codec.SDPFmtpLine, + RTCPFeedback: feedbacks, + } + writeStream.setRTPWriter( + r.api.interceptor.BindLocalStream( + info, + interceptor.RTPWriterFunc(func(p *rtp.Packet, attributes interceptor.Attributes) (int, error) { + return rtpWriteStream.WriteRTP(&p.Header, p.Payload) + }), + )) close(r.sendCalled) return nil @@ -194,15 +226,26 @@ func (r *RTPSender) Read(b []byte) (n int, err error) { } } -// ReadRTCP is a convenience method that wraps Read and unmarshals for you +// ReadRTCP is a convenience method that wraps Read and unmarshals for you. +// It also runs any configured interceptors. func (r *RTPSender) ReadRTCP() ([]rtcp.Packet, error) { + pkts, _, err := r.interceptorRTCPReader.Read() + return pkts, err +} + +func (r *RTPSender) readRTCP() ([]rtcp.Packet, interceptor.Attributes, error) { b := make([]byte, receiveMTU) i, err := r.Read(b) if err != nil { - return nil, err + return nil, nil, err + } + + pkts, err := rtcp.Unmarshal(b[:i]) + if err != nil { + return nil, nil, err } - return rtcp.Unmarshal(b[:i]) + return pkts, make(interceptor.Attributes), nil } // hasSent tells if data has been ever sent for this instance diff --git a/track_local.go b/track_local.go index 1b232b9517e..e6e1da1f490 100644 --- a/track_local.go +++ b/track_local.go @@ -11,10 +11,11 @@ type TrackLocalWriter interface { Write(b []byte) (int, error) } -// TrackLocalContext is the Context passed when a TrackLocal has been Binded/Unbinded from a PeerConnection +// TrackLocalContext is the Context passed when a TrackLocal has been Binded/Unbinded from a PeerConnection, and used +// in Interceptors. type TrackLocalContext struct { id string - codecs []RTPCodecParameters + params RTPParameters ssrc SSRC writeStream TrackLocalWriter } @@ -22,7 +23,13 @@ type TrackLocalContext struct { // CodecParameters returns the negotiated RTPCodecParameters. These are the codecs supported by both // PeerConnections and the SSRC/PayloadTypes func (t *TrackLocalContext) CodecParameters() []RTPCodecParameters { - return t.codecs + return t.params.Codecs +} + +// HeaderExtensions returns the negotiated RTPHeaderExtensionParameters. These are the header extensions supported by +// both PeerConnections and the SSRC/PayloadTypes +func (t *TrackLocalContext) HeaderExtensions() []RTPHeaderExtensionParameter { + return t.params.HeaderExtensions } // SSRC requires the negotiated SSRC of this track diff --git a/track_remote.go b/track_remote.go index dd48e3a8451..e195c637839 100644 --- a/track_remote.go +++ b/track_remote.go @@ -5,6 +5,7 @@ package webrtc import ( "sync" + "github.com/pion/interceptor" "github.com/pion/rtp" ) @@ -19,10 +20,49 @@ type TrackRemote struct { kind RTPCodecType ssrc SSRC codec RTPCodecParameters + params RTPParameters rid string receiver *RTPReceiver peeked []byte + + interceptorRTPReader interceptor.RTPReader +} + +func newTrackRemote(kind RTPCodecType, ssrc SSRC, rid string, receiver *RTPReceiver) *TrackRemote { + t := &TrackRemote{ + kind: kind, + ssrc: ssrc, + rid: rid, + receiver: receiver, + } + t.interceptorRTPReader = interceptor.RTPReaderFunc(t.readRTP) + + return t +} + +func (t *TrackRemote) bindInterceptor() { + headerExtensions := make([]interceptor.RTPHeaderExtension, 0, len(t.params.HeaderExtensions)) + for _, h := range t.params.HeaderExtensions { + headerExtensions = append(headerExtensions, interceptor.RTPHeaderExtension{ID: h.ID, URI: h.URI}) + } + feedbacks := make([]interceptor.RTCPFeedback, 0, len(t.codec.RTCPFeedback)) + for _, f := range t.codec.RTCPFeedback { + feedbacks = append(feedbacks, interceptor.RTCPFeedback{Type: f.Type, Parameter: f.Parameter}) + } + info := &interceptor.StreamInfo{ + ID: t.id, + Attributes: interceptor.Attributes{}, + SSRC: uint32(t.ssrc), + PayloadType: uint8(t.payloadType), + RTPHeaderExtensions: headerExtensions, + MimeType: t.codec.MimeType, + ClockRate: t.codec.ClockRate, + Channels: t.codec.Channels, + SDPFmtpLine: t.codec.SDPFmtpLine, + RTCPFeedback: feedbacks, + } + t.interceptorRTPReader = t.receiver.api.interceptor.BindRemoteStream(info, interceptor.RTPReaderFunc(t.readRTP)) } // ID is the unique identifier for this Track. This should be unique for the @@ -125,19 +165,25 @@ func (t *TrackRemote) peek(b []byte) (n int, err error) { return } -// ReadRTP is a convenience method that wraps Read and unmarshals for you +// ReadRTP is a convenience method that wraps Read and unmarshals for you. +// It also runs any configured interceptors. func (t *TrackRemote) ReadRTP() (*rtp.Packet, error) { + p, _, err := t.interceptorRTPReader.Read() + return p, err +} + +func (t *TrackRemote) readRTP() (*rtp.Packet, interceptor.Attributes, error) { b := make([]byte, receiveMTU) i, err := t.Read(b) if err != nil { - return nil, err + return nil, nil, err } r := &rtp.Packet{} if err := r.Unmarshal(b[:i]); err != nil { - return nil, err + return nil, nil, err } - return r, nil + return r, interceptor.Attributes{}, nil } // determinePayloadType blocks and reads a single packet to determine the PayloadType for this Track