Repository: pion/webrtc (go)

Issue:
interceptor.UnbindLocalStream is never called
https://github.com/pion/webrtc/search?q=UnbindLocalStream

Resources which are released in `interceptor.UnbindLocalStream()` won't be freed.
It should be called on `PeerConnection.Close()`.

Pull request:
Unbind interceptors on stop
Make resources used by interceptor bindings freed.

### Reference issue
Fixes #1712 
Need https://github.com/pion/interceptor/pull/30


Patch:
diff --git a/go.mod b/go.mod
index c7f17d7b..858299df 100644
--- a/go.mod
+++ b/go.mod
@@ -8,7 +8,7 @@ require (
 	github.com/pion/datachannel v1.4.21
 	github.com/pion/dtls/v2 v2.0.8
 	github.com/pion/ice/v2 v2.0.16
-	github.com/pion/interceptor v0.0.11
+	github.com/pion/interceptor v0.0.12
 	github.com/pion/logging v0.2.2
 	github.com/pion/randutil v0.1.0
 	github.com/pion/rtcp v1.2.6
diff --git a/go.sum b/go.sum
index 3d855d69..d4cafd95 100644
--- a/go.sum
+++ b/go.sum
@@ -40,8 +40,8 @@ github.com/pion/dtls/v2 v2.0.8 h1:reGe8rNIMfO/UAeFLqO61tl64t154Qfkr4U3Gzu1tsg=
 github.com/pion/dtls/v2 v2.0.8/go.mod h1:QuDII+8FVvk9Dp5t5vYIMTo7hh7uBkra+8QIm7QGm10=
 github.com/pion/ice/v2 v2.0.16 h1:K6bzD8ef9vMKbGMTHaUweHXEyuNGnvr2zdqKoLKZPn0=
 github.com/pion/ice/v2 v2.0.16/go.mod h1:SJNJzC27gDZoOW0UoxIoC8Hf2PDxG28hQyNdSexDu38=
-github.com/pion/interceptor v0.0.11 h1:YIEMghiTZYb88q6awk3N/8WUU5P+aVyKMuU3YZfTSmI=
-github.com/pion/interceptor v0.0.11/go.mod h1:qzeuWuD/ZXvPqOnxNcnhWfkCZ2e1kwwslicyyPnhoK4=
+github.com/pion/interceptor v0.0.12 h1:eC1iVneBIAQJEfaNAfDqAncJWhMDAnaXPRCJsltdokE=
+github.com/pion/interceptor v0.0.12/go.mod h1:qzeuWuD/ZXvPqOnxNcnhWfkCZ2e1kwwslicyyPnhoK4=
 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.5 h1:Q2oj/JB3NqfzY9xGZ1fPzZzK7sDSD8rZPOvcIQ10BCw=
diff --git a/rtpreceiver.go b/rtpreceiver.go
index 8d558e05..6c44ed6c 100644
--- a/rtpreceiver.go
+++ b/rtpreceiver.go
@@ -19,6 +19,8 @@ import (
 type trackStreams struct {
 	track *TrackRemote
 
+	streamInfo interceptor.StreamInfo
+
 	rtpReadStream  *srtp.ReadStreamSRTP
 	rtpInterceptor interceptor.RTPReader
 
@@ -123,9 +125,9 @@ func (r *RTPReceiver) Receive(parameters RTPReceiveParameters) error {
 			codec = globalParams.Codecs[0].RTPCodecCapability
 		}
 
-		streamInfo := createStreamInfo("", parameters.Encodings[0].SSRC, 0, codec, globalParams.HeaderExtensions)
+		t.streamInfo = createStreamInfo("", parameters.Encodings[0].SSRC, 0, codec, globalParams.HeaderExtensions)
 		var err error
-		if t.rtpReadStream, t.rtpInterceptor, t.rtcpReadStream, t.rtcpInterceptor, err = r.streamsForSSRC(parameters.Encodings[0].SSRC, streamInfo); err != nil {
+		if t.rtpReadStream, t.rtpInterceptor, t.rtcpReadStream, t.rtcpInterceptor, err = r.streamsForSSRC(parameters.Encodings[0].SSRC, t.streamInfo); err != nil {
 			return err
 		}
 
@@ -235,6 +237,7 @@ func (r *RTPReceiver) Stop() error {
 			}
 
 			err = util.FlattenErrs(errs)
+			r.api.interceptor.UnbindRemoteStream(&r.tracks[i].streamInfo)
 		}
 	default:
 	}
@@ -275,11 +278,11 @@ func (r *RTPReceiver) receiveForRid(rid string, params RTPParameters, ssrc SSRC)
 			r.tracks[i].track.codec = params.Codecs[0]
 			r.tracks[i].track.params = params
 			r.tracks[i].track.ssrc = ssrc
-			streamInfo := createStreamInfo("", ssrc, params.Codecs[0].PayloadType, params.Codecs[0].RTPCodecCapability, params.HeaderExtensions)
+			r.tracks[i].streamInfo = createStreamInfo("", ssrc, params.Codecs[0].PayloadType, params.Codecs[0].RTPCodecCapability, params.HeaderExtensions)
 			r.tracks[i].track.mu.Unlock()
 
 			var err error
-			if r.tracks[i].rtpReadStream, r.tracks[i].rtpInterceptor, r.tracks[i].rtcpReadStream, r.tracks[i].rtcpInterceptor, err = r.streamsForSSRC(ssrc, streamInfo); err != nil {
+			if r.tracks[i].rtpReadStream, r.tracks[i].rtpInterceptor, r.tracks[i].rtcpReadStream, r.tracks[i].rtcpInterceptor, err = r.streamsForSSRC(ssrc, r.tracks[i].streamInfo); err != nil {
 				return nil, err
 			}
 
diff --git a/rtpsender.go b/rtpsender.go
index f3d34c78..b9ba82ec 100644
--- a/rtpsender.go
+++ b/rtpsender.go
@@ -19,6 +19,7 @@ type RTPSender struct {
 
 	srtpStream      *srtpWriterFuture
 	rtcpInterceptor interceptor.RTCPReader
+	streamInfo      interceptor.StreamInfo
 
 	context TrackLocalContext
 
@@ -175,8 +176,8 @@ func (r *RTPSender) Send(parameters RTPSendParameters) error {
 	}
 	r.context.params.Codecs = []RTPCodecParameters{codec}
 
-	streamInfo := createStreamInfo(r.id, parameters.Encodings[0].SSRC, codec.PayloadType, codec.RTPCodecCapability, parameters.HeaderExtensions)
-	rtpInterceptor := r.api.interceptor.BindLocalStream(&streamInfo, interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
+	r.streamInfo = createStreamInfo(r.id, parameters.Encodings[0].SSRC, codec.PayloadType, codec.RTPCodecCapability, parameters.HeaderExtensions)
+	rtpInterceptor := r.api.interceptor.BindLocalStream(&r.streamInfo, interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
 		return r.srtpStream.WriteRTP(header, payload)
 	}))
 	writeStream.interceptor.Store(rtpInterceptor)
@@ -205,6 +206,8 @@ func (r *RTPSender) Stop() error {
 		return err
 	}
 
+	r.api.interceptor.UnbindLocalStream(&r.streamInfo)
+
 	return r.srtpStream.Close()
 }
 


Test patch:
diff --git a/interceptor_test.go b/interceptor_test.go
index c0ed41b9..f7b9ce69 100644
--- a/interceptor_test.go
+++ b/interceptor_test.go
@@ -5,44 +5,18 @@ package webrtc
 //
 import (
 	"context"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/pion/interceptor"
+	mock_interceptor "github.com/pion/interceptor/pkg/mock"
 	"github.com/pion/rtp"
 	"github.com/pion/transport/test"
 	"github.com/pion/webrtc/v3/pkg/media"
 	"github.com/stretchr/testify/assert"
 )
 
-type testInterceptor struct {
-	interceptor.NoOp
-
-	t *testing.T
-}
-
-func (t *testInterceptor) BindLocalStream(_ *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter {
-	return interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
-		// set extension on outgoing packet
-		header.Extension = true
-		header.ExtensionProfile = 0xBEDE
-		assert.NoError(t.t, header.SetExtension(2, []byte("foo")))
-
-		return writer.Write(header, payload, attributes)
-	})
-}
-
-func (t *testInterceptor) BindRemoteStream(_ *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader {
-	return interceptor.RTPReaderFunc(func(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) {
-		if a == nil {
-			a = interceptor.Attributes{}
-		}
-
-		a.Set("attribute", "value")
-		return reader.Read(b, a)
-	})
-}
-
 // E2E test of the features of Interceptors
 // * Assert an extension can be set on an outbound packet
 // * Assert an extension can be read on an outbound packet
@@ -59,7 +33,28 @@ func TestPeerConnection_Interceptor(t *testing.T) {
 		assert.NoError(t, m.RegisterDefaultCodecs())
 
 		ir := &interceptor.Registry{}
-		ir.Add(&testInterceptor{t: t})
+		ir.Add(&mock_interceptor.Interceptor{
+			BindLocalStreamFn: func(_ *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter {
+				return interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
+					// set extension on outgoing packet
+					header.Extension = true
+					header.ExtensionProfile = 0xBEDE
+					assert.NoError(t, header.SetExtension(2, []byte("foo")))
+
+					return writer.Write(header, payload, attributes)
+				})
+			},
+			BindRemoteStreamFn: func(_ *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader {
+				return interceptor.RTPReaderFunc(func(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) {
+					if a == nil {
+						a = interceptor.Attributes{}
+					}
+
+					a.Set("attribute", "value")
+					return reader.Read(b, a)
+				})
+			},
+		})
 
 		pc, err := NewAPI(WithMediaEngine(m), WithInterceptorRegistry(ir)).NewPeerConnection(Configuration{})
 		assert.NoError(t, err)
@@ -104,3 +99,113 @@ func TestPeerConnection_Interceptor(t *testing.T) {
 
 	closePairNow(t, offerer, answerer)
 }
+
+func Test_Interceptor_BindUnbind(t *testing.T) {
+	lim := test.TimeOut(time.Second * 10)
+	defer lim.Stop()
+
+	report := test.CheckRoutines(t)
+	defer report()
+
+	m := &MediaEngine{}
+	assert.NoError(t, m.RegisterDefaultCodecs())
+
+	var (
+		cntBindRTCPReader     uint32
+		cntBindRTCPWriter     uint32
+		cntBindLocalStream    uint32
+		cntUnbindLocalStream  uint32
+		cntBindRemoteStream   uint32
+		cntUnbindRemoteStream uint32
+		cntClose              uint32
+	)
+	mockInterceptor := &mock_interceptor.Interceptor{
+		BindRTCPReaderFn: func(reader interceptor.RTCPReader) interceptor.RTCPReader {
+			atomic.AddUint32(&cntBindRTCPReader, 1)
+			return reader
+		},
+		BindRTCPWriterFn: func(writer interceptor.RTCPWriter) interceptor.RTCPWriter {
+			atomic.AddUint32(&cntBindRTCPWriter, 1)
+			return writer
+		},
+		BindLocalStreamFn: func(i *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter {
+			atomic.AddUint32(&cntBindLocalStream, 1)
+			return writer
+		},
+		UnbindLocalStreamFn: func(i *interceptor.StreamInfo) {
+			atomic.AddUint32(&cntUnbindLocalStream, 1)
+		},
+		BindRemoteStreamFn: func(i *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader {
+			atomic.AddUint32(&cntBindRemoteStream, 1)
+			return reader
+		},
+		UnbindRemoteStreamFn: func(i *interceptor.StreamInfo) {
+			atomic.AddUint32(&cntUnbindRemoteStream, 1)
+		},
+		CloseFn: func() error {
+			atomic.AddUint32(&cntClose, 1)
+			return nil
+		},
+	}
+	ir := &interceptor.Registry{}
+	ir.Add(mockInterceptor)
+
+	sender, receiver, err := NewAPI(WithMediaEngine(m), WithInterceptorRegistry(ir)).newPair(Configuration{})
+	assert.NoError(t, err)
+
+	track, err := NewTrackLocalStaticSample(RTPCodecCapability{MimeType: "video/vp8"}, "video", "pion")
+	assert.NoError(t, err)
+
+	_, err = sender.AddTrack(track)
+	assert.NoError(t, err)
+
+	receiverReady, receiverReadyFn := context.WithCancel(context.Background())
+	receiver.OnTrack(func(track *TrackRemote, _ *RTPReceiver) {
+		_, _, readErr := track.ReadRTP()
+		assert.NoError(t, readErr)
+		receiverReadyFn()
+	})
+
+	assert.NoError(t, signalPair(sender, receiver))
+
+	ticker := time.NewTicker(time.Millisecond * 20)
+	defer ticker.Stop()
+	func() {
+		for {
+			select {
+			case <-receiverReady.Done():
+				return
+			case <-ticker.C:
+				// Send packet to make receiver track actual creates RTPReceiver.
+				assert.NoError(t, track.WriteSample(media.Sample{Data: []byte{0xAA}, Duration: time.Second}))
+			}
+		}
+	}()
+
+	closePairNow(t, sender, receiver)
+
+	// Bind/UnbindLocal/RemoteStream should be called from one side.
+	if cnt := atomic.LoadUint32(&cntBindLocalStream); cnt != 1 {
+		t.Errorf("BindLocalStreamFn is expected to be called once, but called %d times", cnt)
+	}
+	if cnt := atomic.LoadUint32(&cntUnbindLocalStream); cnt != 1 {
+		t.Errorf("UnbindLocalStreamFn is expected to be called once, but called %d times", cnt)
+	}
+	if cnt := atomic.LoadUint32(&cntBindRemoteStream); cnt != 1 {
+		t.Errorf("BindRemoteStreamFn is expected to be called once, but called %d times", cnt)
+	}
+	if cnt := atomic.LoadUint32(&cntUnbindRemoteStream); cnt != 1 {
+		t.Errorf("UnbindRemoteStreamFn is expected to be called once, but called %d times", cnt)
+	}
+
+	// BindRTCPWriter/Reader and Close should be called from both side.
+	if cnt := atomic.LoadUint32(&cntBindRTCPWriter); cnt != 2 {
+		t.Errorf("BindRTCPWriterFn is expected to be called twice, but called %d times", cnt)
+	}
+	if cnt := atomic.LoadUint32(&cntBindRTCPReader); cnt != 2 {
+		t.Errorf("BindRTCPReaderFn is expected to be called twice, but called %d times", cnt)
+	}
+	if cnt := atomic.LoadUint32(&cntClose); cnt != 2 {
+		t.Errorf("CloseFn is expected to be called twice, but called %d times", cnt)
+	}
+}