Add video RTP reader

Generated by Renovate Bot

codec.go

@@ -58,17 +58,16 @@ func (selector *CodecSelector) Populate(setting *webrtc.MediaEngine) {
 	}
 }
 
-func (selector *CodecSelector) selectVideoCodec(wantCodecs []*webrtc.RTPCodec, reader video.Reader, inputProp prop.Media) (codec.ReadCloser, *codec.RTPCodec, error) {
+func (selector *CodecSelector) selectVideoCodecByNames(reader video.Reader, inputProp prop.Media, codecNames ...string) (codec.ReadCloser, *codec.RTPCodec, error) {
 	var selectedEncoder codec.VideoEncoderBuilder
 	var encodedReader codec.ReadCloser
 	var errReasons []string
 	var err error
 
 outer:
-	for _, wantCodec := range wantCodecs {
-		name := wantCodec.Name
+	for _, wantCodec := range codecNames {
 		for _, encoder := range selector.videoEncoders {
-			if encoder.RTPCodec().Name == name {
+			if encoder.RTPCodec().Name == wantCodec {
 				encodedReader, err = encoder.BuildVideoEncoder(reader, inputProp)
 				if err == nil {
 					selectedEncoder = encoder
@@ -87,17 +86,26 @@ outer:
 	return encodedReader, selectedEncoder.RTPCodec(), nil
 }
 
-func (selector *CodecSelector) selectAudioCodec(wantCodecs []*webrtc.RTPCodec, reader audio.Reader, inputProp prop.Media) (codec.ReadCloser, *codec.RTPCodec, error) {
+func (selector *CodecSelector) selectVideoCodec(reader video.Reader, inputProp prop.Media, codecs ...*webrtc.RTPCodec) (codec.ReadCloser, *codec.RTPCodec, error) {
+	var codecNames []string
+
+	for _, codec := range codecs {
+		codecNames = append(codecNames, codec.Name)
+	}
+
+	return selector.selectVideoCodecByNames(reader, inputProp, codecNames...)
+}
+
+func (selector *CodecSelector) selectAudioCodecByNames(reader audio.Reader, inputProp prop.Media, codecNames ...string) (codec.ReadCloser, *codec.RTPCodec, error) {
 	var selectedEncoder codec.AudioEncoderBuilder
 	var encodedReader codec.ReadCloser
 	var errReasons []string
 	var err error
 
 outer:
-	for _, wantCodec := range wantCodecs {
-		name := wantCodec.Name
+	for _, wantCodec := range codecNames {
 		for _, encoder := range selector.audioEncoders {
-			if encoder.RTPCodec().Name == name {
+			if encoder.RTPCodec().Name == wantCodec {
 				encodedReader, err = encoder.BuildAudioEncoder(reader, inputProp)
 				if err == nil {
 					selectedEncoder = encoder
@@ -115,3 +123,13 @@ outer:
 
 	return encodedReader, selectedEncoder.RTPCodec(), nil
 }
+
+func (selector *CodecSelector) selectAudioCodec(reader audio.Reader, inputProp prop.Media, codecs ...*webrtc.RTPCodec) (codec.ReadCloser, *codec.RTPCodec, error) {
+	var codecNames []string
+
+	for _, codec := range codecs {
+		codecNames = append(codecNames, codec.Name)
+	}
+
+	return selector.selectAudioCodecByNames(reader, inputProp, codecNames...)
+}

examples/rtp/README.md

@@ -0,0 +1,30 @@
+## Instructions
+
+### Download rtp-send example
+
+```
+go get github.com/pion/mediadevices/examples/rtp-send
+```
+
+### Listen RTP
+
+Install GStreamer and run:
+```
+gst-launch-1.0 udpsrc port=5000 caps=application/x-rtp,encode-name=VP8 \
+    ! rtpvp8depay ! vp8dec ! videoconvert ! autovideosink
+```
+
+Or run VLC media plyer:
+```
+vlc ./vp8.sdp
+```
+
+### Run rtp-send
+
+Run `rtp-send localhost:5000`
+
+A video should start playing in your GStreamer or VLC window.
+It's not WebRTC, but pure RTP.
+
+Congrats, you have used pion-MediaDevices! Now start building something cool
+

examples/rtp/main.go

@@ -0,0 +1,76 @@
+package main
+
+import (
+	"fmt"
+	"net"
+	"os"
+
+	"github.com/pion/mediadevices"
+	"github.com/pion/mediadevices/pkg/codec/vpx"       // This is required to use VP8/VP9 video encoder
+	_ "github.com/pion/mediadevices/pkg/driver/camera" // This is required to register camera adapter
+	"github.com/pion/mediadevices/pkg/frame"
+	"github.com/pion/mediadevices/pkg/prop"
+)
+
+const (
+	mtu = 1000
+)
+
+func must(err error) {
+	if err != nil {
+		panic(err)
+	}
+}
+
+func main() {
+	if len(os.Args) != 2 {
+		fmt.Printf("usage: %s host:port\n", os.Args[0])
+		return
+	}
+	dest := os.Args[1]
+
+	vp8Params, err := vpx.NewVP8Params()
+	must(err)
+	vp8Params.BitRate = 100000 // 100kbps
+
+	codecSelector := mediadevices.NewCodecSelector(
+		mediadevices.WithVideoEncoders(&vp8Params),
+	)
+
+	mediaStream, err := mediadevices.GetUserMedia(mediadevices.MediaStreamConstraints{
+		Video: func(c *mediadevices.MediaTrackConstraints) {
+			c.FrameFormat = prop.FrameFormat(frame.FormatYUY2)
+			c.Width = prop.Int(640)
+			c.Height = prop.Int(480)
+		},
+		Codec: codecSelector,
+	})
+	must(err)
+
+	videoTrack := mediaStream.GetVideoTracks()[0].(*mediadevices.VideoTrack)
+	defer videoTrack.Close()
+
+	rtpReader, err := videoTrack.NewRTPReader(vp8Params.RTPCodec().Name, mtu)
+	must(err)
+
+	addr, err := net.ResolveUDPAddr("udp", dest)
+	must(err)
+	conn, err := net.DialUDP("udp", nil, addr)
+	must(err)
+
+	buff := make([]byte, mtu)
+	for {
+		pkts, release, err := rtpReader.Read()
+		must(err)
+
+		for _, pkt := range pkts {
+			n, err := pkt.MarshalTo(buff)
+			must(err)
+
+			_, err = conn.Write(buff[:n])
+			must(err)
+		}
+
+		release()
+	}
+}

examples/rtp/vp8.sdp

@@ -0,0 +1,9 @@
+v=0
+o=- 1234567890 1234567890 IN IP4 0.0.0.0
+s=RTP-Send Example
+i=Example
+c=IN IP4 0.0.0.0
+t=0 0
+a=recvonly
+m=video 5000 RTP/AVP 100
+a=rtpmap:100 VP8/90000

go.mod

@@ -7,8 +7,9 @@ require (
 	github.com/jfreymuth/pulse v0.0.0-20201014123913-1e525c426c93
 	github.com/lherman-cs/opus v0.0.2
 	github.com/pion/logging v0.2.2
+	github.com/pion/rtp v1.6.0
 	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
-	golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f
+	golang.org/x/sys v0.0.0-20201029080932-201ba4db2418
 )

go.sum

@@ -109,8 +109,8 @@ golang.org/x/sys v0.0.0-20190228124157-a34e9553db1e/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200724161237-0e2f3a69832c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f h1:+Nyd8tzPX9R7BWHguqsrbFdRx3WQ/1ib8I44HXV5yTA=
-golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201029080932-201ba4db2418 h1:HlFl4V6pEMziuLXyRkm5BIYq1y1GAbb02pRlWvI54OM=
+golang.org/x/sys v0.0.0-20201029080932-201ba4db2418/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=

rtpreader.go

@@ -0,0 +1,21 @@
+package mediadevices
+
+import "github.com/pion/rtp"
+
+type RTPReader interface {
+	Read() (pkts []*rtp.Packet, release func(), err error)
+	Close() error
+}
+
+type rtpReaderImpl struct {
+	readFn  func() ([]*rtp.Packet, func(), error)
+	closeFn func() error
+}
+
+func (r *rtpReaderImpl) Read() ([]*rtp.Packet, func(), error) {
+	return r.readFn()
+}
+
+func (r *rtpReaderImpl) Close() error {
+	return r.closeFn()
+}

sampler.go

@@ -3,34 +3,30 @@ package mediadevices
 import (
 	"math"
 	"time"
-
-	"github.com/pion/webrtc/v2"
-	"github.com/pion/webrtc/v2/pkg/media"
 )
 
-type samplerFunc func(b []byte) error
+type samplerFunc func() uint32
 
 // newVideoSampler creates a video sampler that uses the actual video frame rate and
 // the codec's clock rate to come up with a duration for each sample.
-func newVideoSampler(t *webrtc.Track) samplerFunc {
-	clockRate := float64(t.Codec().ClockRate)
+func newVideoSampler(clockRate uint32) samplerFunc {
+	clockRateFloat := float64(clockRate)
 	lastTimestamp := time.Now()
 
-	return samplerFunc(func(b []byte) error {
+	return samplerFunc(func() uint32 {
 		now := time.Now()
 		duration := now.Sub(lastTimestamp).Seconds()
-		samples := uint32(math.Round(clockRate * duration))
+		samples := uint32(math.Round(clockRateFloat * duration))
 		lastTimestamp = now
-
-		return t.WriteSample(media.Sample{Data: b, Samples: samples})
+		return samples
 	})
 }
 
 // newAudioSampler creates a audio sampler that uses a fixed latency and
 // the codec's clock rate to come up with a duration for each sample.
-func newAudioSampler(t *webrtc.Track, latency time.Duration) samplerFunc {
-	samples := uint32(math.Round(float64(t.Codec().ClockRate) * latency.Seconds()))
-	return samplerFunc(func(b []byte) error {
-		return t.WriteSample(media.Sample{Data: b, Samples: samples})
+func newAudioSampler(clockRate uint32, latency time.Duration) samplerFunc {
+	samples := uint32(math.Round(float64(clockRate) * latency.Seconds()))
+	return samplerFunc(func() uint32 {
+		return samples
 	})
 }

track.go

@@ -11,7 +11,9 @@ import (
 	"github.com/pion/mediadevices/pkg/io/audio"
 	"github.com/pion/mediadevices/pkg/io/video"
 	"github.com/pion/mediadevices/pkg/wave"
+	"github.com/pion/rtp"
 	"github.com/pion/webrtc/v2"
+	"github.com/pion/webrtc/v2/pkg/media"
 )
 
 var (
@@ -109,7 +111,7 @@ func (track *baseTrack) onError(err error) {
 	}
 }
 
-func (track *baseTrack) bind(pc *webrtc.PeerConnection, encodedReader codec.ReadCloser, selectedCodec *codec.RTPCodec, sampler func(*webrtc.Track) samplerFunc) (*webrtc.Track, error) {
+func (track *baseTrack) bind(pc *webrtc.PeerConnection, encodedReader codec.ReadCloser, selectedCodec *codec.RTPCodec, sample samplerFunc) (*webrtc.Track, error) {
 	track.mu.Lock()
 	defer track.mu.Unlock()
 
@@ -118,7 +120,6 @@ func (track *baseTrack) bind(pc *webrtc.PeerConnection, encodedReader codec.Read
 		return nil, err
 	}
 
-	sample := sampler(webrtcTrack)
 	signalCh := make(chan chan<- struct{})
 	track.activePeerConnections[pc] = signalCh
 
@@ -147,7 +148,12 @@ func (track *baseTrack) bind(pc *webrtc.PeerConnection, encodedReader codec.Read
 				return
 			}
 
-			if err := sample(buff); err != nil {
+			sampleCount := sample()
+			err = webrtcTrack.WriteSample(media.Sample{
+				Data:    buff,
+				Samples: sampleCount,
+			})
+			if err != nil {
 				track.onError(err)
 				return
 			}
@@ -242,18 +248,53 @@ func (track *VideoTrack) Bind(pc *webrtc.PeerConnection) (*webrtc.Track, error)
 	}
 
 	wantCodecs := pc.GetRegisteredRTPCodecs(webrtc.RTPCodecTypeVideo)
-	encodedReader, selectedCodec, err := track.selector.selectVideoCodec(wantCodecs, reader, inputProp)
+	encodedReader, selectedCodec, err := track.selector.selectVideoCodec(reader, inputProp, wantCodecs...)
 	if err != nil {
 		return nil, err
 	}
 
-	return track.bind(pc, encodedReader, selectedCodec, newVideoSampler)
+	return track.bind(pc, encodedReader, selectedCodec, newVideoSampler(selectedCodec.ClockRate))
 }
 
 func (track *VideoTrack) Unbind(pc *webrtc.PeerConnection) error {
 	return track.unbind(pc)
 }
 
+func (track *VideoTrack) NewRTPReader(codecName string, mtu int) (RTPReader, error) {
+	reader := track.NewReader(false)
+	inputProp, err := detectCurrentVideoProp(track.Broadcaster)
+	if err != nil {
+		return nil, err
+	}
+
+	encodedReader, selectedCodec, err := track.selector.selectVideoCodecByNames(reader, inputProp, codecName)
+	if err != nil {
+		return nil, err
+	}
+
+	sample := newVideoSampler(selectedCodec.ClockRate)
+
+	// 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)
+
+	return &rtpReaderImpl{
+		readFn: func() ([]*rtp.Packet, func(), error) {
+			encoded, release, err := encodedReader.Read()
+			if err != nil {
+				encodedReader.Close()
+				track.onError(err)
+				return nil, func() {}, err
+			}
+			defer release()
+
+			samples := sample()
+			pkts := packetizer.Packetize(encoded, samples)
+			return pkts, release, err
+		},
+		closeFn: encodedReader.Close,
+	}, nil
+}
+
 // AudioTrack is a specific track type that contains audio source which allows multiple readers to access, and
 // manipulate.
 type AudioTrack struct {
@@ -312,12 +353,12 @@ func (track *AudioTrack) Bind(pc *webrtc.PeerConnection) (*webrtc.Track, error)
 	}
 
 	wantCodecs := pc.GetRegisteredRTPCodecs(webrtc.RTPCodecTypeAudio)
-	encodedReader, selectedCodec, err := track.selector.selectAudioCodec(wantCodecs, reader, inputProp)
+	encodedReader, selectedCodec, err := track.selector.selectAudioCodec(reader, inputProp, wantCodecs...)
 	if err != nil {
 		return nil, err
 	}
 
-	return track.bind(pc, encodedReader, selectedCodec, func(t *webrtc.Track) samplerFunc { return newAudioSampler(t, inputProp.Latency) })
+	return track.bind(pc, encodedReader, selectedCodec, newAudioSampler(selectedCodec.ClockRate, inputProp.Latency))
 }
 
 func (track *AudioTrack) Unbind(pc *webrtc.PeerConnection) error {