Update module pion/rtp to v1.6.2 (#265)

Generated by Renovate Bot

Co-authored-by: Renovate Bot <bot@renovateapp.com>

.github/workflows/ci.yaml

@@ -11,6 +11,7 @@ jobs:
   build-linux:
     runs-on: ubuntu-latest
     strategy:
+      fail-fast: false
       matrix:
         go: [ '1.15', '1.14' ]
     name: Linux Go ${{ matrix.go }}
@@ -48,6 +49,7 @@ jobs:
   build-darwin:
     runs-on: macos-latest
     strategy:
+      fail-fast: false
       matrix:
         go: [ '1.15', '1.14' ]
     name: Darwin Go ${{ matrix.go }}

go.mod

@@ -5,13 +5,13 @@ go 1.13
 require (
 	github.com/BurntSushi/xgb v0.0.0-20201008132610-5f9e7b3c49cd // indirect
 	github.com/blackjack/webcam v0.0.0-20200313125108-10ed912a8539
-	github.com/gen2brain/malgo v0.10.25
+	github.com/gen2brain/malgo v0.10.27
 	github.com/gen2brain/shm v0.0.0-20200228170931-49f9650110c5 // indirect
 	github.com/kbinani/screenshot v0.0.0-20191211154542-3a185f1ce18f
 	github.com/lherman-cs/opus v0.0.2
 	github.com/lxn/win v0.0.0-20201111105847-2a20daff6a55 // indirect
 	github.com/pion/logging v0.2.2
-	github.com/pion/rtp v1.6.1
+	github.com/pion/rtp v1.6.2
 	github.com/pion/webrtc/v2 v2.2.26
 	github.com/satori/go.uuid v1.2.0
 	golang.org/x/image v0.0.0-20200927104501-e162460cd6b5

go.sum

@@ -9,8 +9,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
-github.com/gen2brain/malgo v0.10.25 h1:VRiYTBmBeHTCXD0wCg7XyLi6lJJBqND/XVmSEyrGkGc=
-github.com/gen2brain/malgo v0.10.25/go.mod h1:zHSUNZAXfCeNsZou0RtQ6Zk7gDYLIcKOrUWtAdksnEs=
+github.com/gen2brain/malgo v0.10.27 h1:KlNitZIO8V4W2VnjtTM8AGMy/XBb2pN+fnIB5bEps8E=
+github.com/gen2brain/malgo v0.10.27/go.mod h1:zHSUNZAXfCeNsZou0RtQ6Zk7gDYLIcKOrUWtAdksnEs=
 github.com/gen2brain/shm v0.0.0-20200228170931-49f9650110c5 h1:Y5Q2mEwfzjMt5+3u70Gtw93ZOu2UuPeeeTBDntF7FoY=
 github.com/gen2brain/shm v0.0.0-20200228170931-49f9650110c5/go.mod h1:uF6rMu/1nvu+5DpiRLwusA6xB8zlkNoGzKn8lmYONUo=
 github.com/golang/mock v1.2.0 h1:28o5sBqPkBsMGnC6b4MvE2TzSr5/AT4c/1fLqVGIwlk=
@@ -61,8 +61,8 @@ github.com/pion/rtcp v1.2.3 h1:2wrhKnqgSz91Q5nzYTO07mQXztYPtxL8a0XOss4rJqA=
 github.com/pion/rtcp v1.2.3/go.mod h1:zGhIv0RPRF0Z1Wiij22pUt5W/c9fevqSzT4jje/oK7I=
 github.com/pion/rtp v1.6.0 h1:4Ssnl/T5W2LzxHj9ssYpGVEQh3YYhQFNVmSWO88MMwk=
 github.com/pion/rtp v1.6.0/go.mod h1:QgfogHsMBVE/RFNno467U/KBqfUywEH+HK+0rtnwsdI=
-github.com/pion/rtp v1.6.1 h1:2Y2elcVBrahYnHKN2X7rMHX/r1R4TEBMP1LaVu/wNhk=
-github.com/pion/rtp v1.6.1/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko=
+github.com/pion/rtp v1.6.2 h1:iGBerLX6JiDjB9NXuaPzHyxHFG9JsIEdgwTC0lp5n/U=
+github.com/pion/rtp v1.6.2/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko=
 github.com/pion/sctp v1.7.10 h1:o3p3/hZB5Cx12RMGyWmItevJtZ6o2cpuxaw6GOS4x+8=
 github.com/pion/sctp v1.7.10/go.mod h1:EhpTUQu1/lcK3xI+eriS6/96fWetHGCvBi9MSsnaBN0=
 github.com/pion/sdp/v2 v2.4.0 h1:luUtaETR5x2KNNpvEMv/r4Y+/kzImzbz4Lm1z8eQNQI=

internal/codec/measurement_test.go

@@ -2,12 +2,18 @@ package codec
 
 import (
 	"io"
+	"runtime"
 	"sync"
 	"testing"
 	"time"
 )
 
 func TestMeasureBitRateStatic(t *testing.T) {
+	// https://github.com/pion/mediadevices/issues/198
+	if runtime.GOOS == "darwin" {
+		t.Skip("Skipping because Darwin CI is not reliable for timing related tests.")
+	}
+
 	r, w := io.Pipe()
 	const (
 		dataSize       = 1000
@@ -54,6 +60,11 @@ func TestMeasureBitRateStatic(t *testing.T) {
 }
 
 func TestMeasureBitRateDynamic(t *testing.T) {
+	// https://github.com/pion/mediadevices/issues/198
+	if runtime.GOOS == "darwin" {
+		t.Skip("Skipping because Darwin CI is not reliable for timing related tests.")
+	}
+
 	r, w := io.Pipe()
 	const (
 		dataSize       = 1000

ioreader.go

@@ -1,14 +1,61 @@
 package mediadevices
 
+type EncodedBuffer struct {
+	Data    []byte
+	Samples uint32
+}
+
+type EncodedReadCloser interface {
+	Read() (EncodedBuffer, func(), error)
+	Close() error
+}
+
 type encodedReadCloserImpl struct {
-	readFn  func([]byte) (int, error)
+	readFn  func() (EncodedBuffer, func(), error)
 	closeFn func() error
 }
 
-func (r *encodedReadCloserImpl) Read(b []byte) (int, error) {
-	return r.readFn(b)
+func (r *encodedReadCloserImpl) Read() (EncodedBuffer, func(), error) {
+	return r.readFn()
 }
 
 func (r *encodedReadCloserImpl) Close() error {
 	return r.closeFn()
 }
+
+type encodedIOReadCloserImpl struct {
+	readFn  func([]byte) (int, error)
+	closeFn func() error
+}
+
+func newEncodedIOReadCloserImpl(reader EncodedReadCloser) *encodedIOReadCloserImpl {
+	var encoded EncodedBuffer
+	release := func() {}
+	return &encodedIOReadCloserImpl{
+		readFn: func(b []byte) (int, error) {
+			var err error
+
+			if len(encoded.Data) == 0 {
+				release()
+				encoded, release, err = reader.Read()
+				if err != nil {
+					reader.Close()
+					return 0, err
+				}
+			}
+
+			n := copy(b, encoded.Data)
+			encoded.Data = encoded.Data[n:]
+			return n, nil
+		},
+		closeFn: reader.Close,
+	}
+}
+
+func (r *encodedIOReadCloserImpl) Read(b []byte) (int, error) {
+	return r.readFn(b)
+}
+
+func (r *encodedIOReadCloserImpl) Close() error {
+	return r.closeFn()
+}

mediastream_test.go

@@ -38,7 +38,11 @@ func (track *mockMediaStreamTrack) NewRTPReader(codecName string, mtu int) (RTPR
 	return nil, nil
 }
 
-func (track *mockMediaStreamTrack) NewEncodedReader(codecName string) (io.ReadCloser, error) {
+func (track *mockMediaStreamTrack) NewEncodedReader(codecName string) (EncodedReadCloser, error) {
+	return nil, nil
+}
+
+func (track *mockMediaStreamTrack) NewEncodedIOReader(codecName string) (io.ReadCloser, error) {
 	return nil, nil
 }
 

meta.go

@@ -14,7 +14,7 @@ func detectCurrentVideoProp(broadcaster *video.Broadcaster) (prop.Media, error)
 	// buffered frame or a new frame from the source. This also implies that no frame will be lost
 	// in any case.
 	metaReader := broadcaster.NewReader(false)
-	metaReader = video.DetectChanges(0, func(p prop.Media) { currentProp = p })(metaReader)
+	metaReader = video.DetectChanges(0, 0, func(p prop.Media) { currentProp = p })(metaReader)
 	_, _, err := metaReader.Read()
 
 	return currentProp, err

pkg/io/video/detect.go

@@ -2,6 +2,7 @@ package video
 
 import (
 	"image"
+	"math"
 	"time"
 
 	"github.com/pion/mediadevices/pkg/prop"
@@ -9,7 +10,7 @@ import (
 
 // DetectChanges will detect frame and video property changes. For video property detection,
 // since it's time related, interval will be used to determine the sample rate.
-func DetectChanges(interval time.Duration, onChange func(prop.Media)) TransformFunc {
+func DetectChanges(interval time.Duration, fpsDiffTolerance float64, onChange func(prop.Media)) TransformFunc {
 	return func(r Reader) Reader {
 		var currentProp prop.Media
 		var lastTaken time.Time
@@ -40,11 +41,12 @@ func DetectChanges(interval time.Duration, onChange func(prop.Media)) TransformF
 			elapsed := now.Sub(lastTaken)
 			if elapsed >= interval {
 				fps := float32(float64(frames) / elapsed.Seconds())
-				// TODO: maybe add some epsilon so that small changes will not mark as dirty
-				currentProp.FrameRate = fps
 				frames = 0
 				lastTaken = now
-				dirty = true
+				if math.Abs(float64(currentProp.FrameRate-fps)) > fpsDiffTolerance {
+					currentProp.FrameRate = fps
+					dirty = true
+				}
 			}
 
 			if dirty {

pkg/io/video/detect_test.go

@@ -28,7 +28,7 @@ func BenchmarkDetectChanges(b *testing.B) {
 		src := src
 		b.Run(fmt.Sprintf("WithDetectChanges%d", n), func(b *testing.B) {
 			for i := 0; i < n; i++ {
-				src = DetectChanges(time.Microsecond, func(p prop.Media) {})(src)
+				src = DetectChanges(time.Microsecond, 0, func(p prop.Media) {})(src)
 			}
 
 			for i := 0; i < b.N; i++ {
@@ -74,6 +74,27 @@ func TestDetectChanges(t *testing.T) {
 		}
 	}
 
+	SlowDownAfterThrottle := func(rate float32, factor float64, after time.Duration) TransformFunc {
+		return func(r Reader) Reader {
+			sleep := float64(time.Second) / float64(rate)
+			start := time.Now()
+			f := 1.0
+			return ReaderFunc(func() (image.Image, func(), error) {
+				for {
+					img, _, err := r.Read()
+					if err != nil {
+						return nil, func() {}, err
+					}
+					if time.Since(start) > after {
+						f = factor
+					}
+					time.Sleep(time.Duration(sleep * f))
+					return img, func() {}, nil
+				}
+			})
+		}
+	}
+
 	t.Run("OnChangeCalledBeforeFirstFrame", func(t *testing.T) {
 		var detectBeforeFirstFrame bool
 		var expected prop.Media
@@ -81,7 +102,7 @@ func TestDetectChanges(t *testing.T) {
 		expected.Width = 1920
 		expected.Height = 1080
 		src, _ := buildSource(expected)
-		src = DetectChanges(time.Second, func(p prop.Media) {
+		src = DetectChanges(time.Second, 0, func(p prop.Media) {
 			actual = p
 			detectBeforeFirstFrame = true
 		})(src)
@@ -104,7 +125,7 @@ func TestDetectChanges(t *testing.T) {
 		expected.Width = 1920
 		expected.Height = 1080
 		src, update := buildSource(expected)
-		src = DetectChanges(time.Second, func(p prop.Media) {
+		src = DetectChanges(time.Second, 0, func(p prop.Media) {
 			actual = p
 		})(src)
 
@@ -137,7 +158,7 @@ func TestDetectChanges(t *testing.T) {
 		expected.FrameRate = 30
 		src, _ := buildSource(expected)
 		src = Throttle(expected.FrameRate)(src)
-		src = DetectChanges(time.Second*5, func(p prop.Media) {
+		src = DetectChanges(time.Second*5, 0, func(p prop.Media) {
 			actual = p
 			count++
 		})(src)
@@ -155,4 +176,31 @@ func TestDetectChanges(t *testing.T) {
 			assertEq(t, actual, expected, frame, checkFrameRate)
 		}
 	})
+
+	t.Run("OnChangeNotCalledForToleratedFrameRateVariation", func(t *testing.T) {
+		// https://github.com/pion/mediadevices/issues/198
+		if runtime.GOOS == "darwin" {
+			t.Skip("Skipping because Darwin CI is not reliable for timing related tests.")
+		}
+
+		var expected prop.Media
+		var count int
+		expected.Width = 1920
+		expected.Height = 1080
+		expected.FrameRate = 30
+		src, _ := buildSource(expected)
+		src = SlowDownAfterThrottle(expected.FrameRate, 1.1, time.Second)(src)
+		src = DetectChanges(time.Second, 5, func(p prop.Media) {
+			count++
+		})(src)
+		for start := time.Now(); time.Since(start) < 3*time.Second; {
+			src.Read()
+		}
+		// onChange is called once before first frame: prop.FrameRate still 0.
+		// onChange is called again after receiving frames during the specified interval: prop.FrameRate is properly calculated
+		// So if the frame rate only changes within the specified tolerance, onChange should no longer be called.
+		if count > 2 {
+			t.Fatalf("onChange was called more than twice.")
+		}
+	})
 }

track.go

@@ -58,8 +58,10 @@ type Track interface {
 	// NewRTPReader creates a new reader from the source. The reader will encode the source, and packetize
 	// the encoded data in RTP format with given mtu size.
 	NewRTPReader(codecName string, mtu int) (RTPReadCloser, error)
+	// NewEncodedReader creates a EncodedReadCloser that reads the encoded data in codecName format
+	NewEncodedReader(codecName string) (EncodedReadCloser, error)
 	// NewEncodedReader creates a new Go standard io.ReadCloser that reads the encoded data in codecName format
-	NewEncodedReader(codecName string) (io.ReadCloser, error)
+	NewEncodedIOReader(codecName string) (io.ReadCloser, error)
 }
 
 type baseTrack struct {
@@ -185,31 +187,6 @@ func (track *baseTrack) unbind(pc *webrtc.PeerConnection) error {
 	return nil
 }
 
-func (track *baseTrack) newEncodedReader(reader codec.ReadCloser) (io.ReadCloser, error) {
-	var encoded []byte
-	release := func() {}
-	return &encodedReadCloserImpl{
-		readFn: func(b []byte) (int, error) {
-			var err error
-
-			if len(encoded) == 0 {
-				release()
-				encoded, release, err = reader.Read()
-				if err != nil {
-					reader.Close()
-					track.onError(err)
-					return 0, err
-				}
-			}
-
-			n := copy(b, encoded)
-			encoded = encoded[n:]
-			return n, nil
-		},
-		closeFn: reader.Close,
-	}, nil
-}
-
 func newTrackFromDriver(d driver.Driver, constraints MediaTrackConstraints, selector *CodecSelector) (Track, error) {
 	if err := d.Open(); err != nil {
 		return nil, err
@@ -291,20 +268,52 @@ func (track *VideoTrack) Unbind(pc *webrtc.PeerConnection) error {
 	return track.unbind(pc)
 }
 
-func (track *VideoTrack) NewRTPReader(codecName string, mtu int) (RTPReadCloser, error) {
+func (track *VideoTrack) newEncodedReader(codecNames ...string) (EncodedReadCloser, *codec.RTPCodec, error) {
 	reader := track.NewReader(false)
 	inputProp, err := detectCurrentVideoProp(track.Broadcaster)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
-	encodedReader, selectedCodec, err := track.selector.selectVideoCodecByNames(reader, inputProp, codecName)
+	encodedReader, selectedCodec, err := track.selector.selectVideoCodecByNames(reader, inputProp, codecNames...)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	sample := newVideoSampler(selectedCodec.ClockRate)
 
+	return &encodedReadCloserImpl{
+		readFn: func() (EncodedBuffer, func(), error) {
+			data, release, err := encodedReader.Read()
+			buffer := EncodedBuffer{
+				Data:    data,
+				Samples: sample(),
+			}
+			return buffer, release, err
+		},
+		closeFn: encodedReader.Close,
+	}, selectedCodec, nil
+}
+
+func (track *VideoTrack) NewEncodedReader(codecName string) (EncodedReadCloser, error) {
+	reader, _, err := track.newEncodedReader(codecName)
+	return reader, err
+}
+
+func (track *VideoTrack) NewEncodedIOReader(codecName string) (io.ReadCloser, error) {
+	encodedReader, _, err := track.newEncodedReader(codecName)
+	if err != nil {
+		return nil, err
+	}
+	return newEncodedIOReadCloserImpl(encodedReader), nil
+}
+
+func (track *VideoTrack) NewRTPReader(codecName string, mtu int) (RTPReadCloser, error) {
+	encodedReader, selectedCodec, err := track.newEncodedReader(codecName)
+	if err != nil {
+		return nil, err
+	}
+
 	// FIXME: not sure the best way to get unique ssrc. We probably should have a global keeper that can generate a random ID and does book keeping?
 	packetizer := rtp.NewPacketizer(mtu, selectedCodec.PayloadType, rand.Uint32(), selectedCodec.Payloader, rtp.NewRandomSequencer(), selectedCodec.ClockRate)
 
@@ -318,29 +327,13 @@ func (track *VideoTrack) NewRTPReader(codecName string, mtu int) (RTPReadCloser,
 			}
 			defer release()
 
-			samples := sample()
-			pkts := packetizer.Packetize(encoded, samples)
+			pkts := packetizer.Packetize(encoded.Data, encoded.Samples)
 			return pkts, release, err
 		},
 		closeFn: encodedReader.Close,
 	}, nil
 }
 
-func (track *VideoTrack) NewEncodedReader(codecName string) (io.ReadCloser, error) {
-	reader := track.NewReader(false)
-	inputProp, err := detectCurrentVideoProp(track.Broadcaster)
-	if err != nil {
-		return nil, err
-	}
-
-	encodedReader, _, err := track.selector.selectVideoCodecByNames(reader, inputProp, codecName)
-	if err != nil {
-		return nil, err
-	}
-
-	return track.newEncodedReader(encodedReader)
-}
-
 // AudioTrack is a specific track type that contains audio source which allows multiple readers to access, and
 // manipulate.
 type AudioTrack struct {
@@ -408,20 +401,52 @@ func (track *AudioTrack) Unbind(pc *webrtc.PeerConnection) error {
 	return track.unbind(pc)
 }
 
-func (track *AudioTrack) NewRTPReader(codecName string, mtu int) (RTPReadCloser, error) {
+func (track *AudioTrack) newEncodedReader(codecNames ...string) (EncodedReadCloser, *codec.RTPCodec, error) {
 	reader := track.NewReader(false)
 	inputProp, err := detectCurrentAudioProp(track.Broadcaster)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
-	encodedReader, selectedCodec, err := track.selector.selectAudioCodecByNames(reader, inputProp, codecName)
+	encodedReader, selectedCodec, err := track.selector.selectAudioCodecByNames(reader, inputProp, codecNames...)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	sample := newAudioSampler(selectedCodec.ClockRate, inputProp.Latency)
 
+	return &encodedReadCloserImpl{
+		readFn: func() (EncodedBuffer, func(), error) {
+			data, release, err := encodedReader.Read()
+			buffer := EncodedBuffer{
+				Data:    data,
+				Samples: sample(),
+			}
+			return buffer, release, err
+		},
+		closeFn: encodedReader.Close,
+	}, selectedCodec, nil
+}
+
+func (track *AudioTrack) NewEncodedReader(codecName string) (EncodedReadCloser, error) {
+	reader, _, err := track.newEncodedReader(codecName)
+	return reader, err
+}
+
+func (track *AudioTrack) NewEncodedIOReader(codecName string) (io.ReadCloser, error) {
+	encodedReader, _, err := track.newEncodedReader(codecName)
+	if err != nil {
+		return nil, err
+	}
+	return newEncodedIOReadCloserImpl(encodedReader), nil
+}
+
+func (track *AudioTrack) NewRTPReader(codecName string, mtu int) (RTPReadCloser, error) {
+	encodedReader, selectedCodec, err := track.newEncodedReader(codecName)
+	if err != nil {
+		return nil, err
+	}
+
 	// FIXME: not sure the best way to get unique ssrc. We probably should have a global keeper that can generate a random ID and does book keeping?
 	packetizer := rtp.NewPacketizer(mtu, selectedCodec.PayloadType, rand.Uint32(), selectedCodec.Payloader, rtp.NewRandomSequencer(), selectedCodec.ClockRate)
 
@@ -435,25 +460,9 @@ func (track *AudioTrack) NewRTPReader(codecName string, mtu int) (RTPReadCloser,
 			}
 			defer release()
 
-			samples := sample()
-			pkts := packetizer.Packetize(encoded, samples)
+			pkts := packetizer.Packetize(encoded.Data, encoded.Samples)
 			return pkts, release, err
 		},
 		closeFn: encodedReader.Close,
 	}, nil
 }
-
-func (track *AudioTrack) NewEncodedReader(codecName string) (io.ReadCloser, error) {
-	reader := track.NewReader(false)
-	inputProp, err := detectCurrentAudioProp(track.Broadcaster)
-	if err != nil {
-		return nil, err
-	}
-
-	encodedReader, _, err := track.selector.selectAudioCodecByNames(reader, inputProp, codecName)
-	if err != nil {
-		return nil, err
-	}
-
-	return track.newEncodedReader(encodedReader)
-}