add RTP/Opus decoder and encoder (#151)

README.md

@@ -46,6 +46,7 @@ Features:
   * Parse RTSP elements: requests, responses, SDP
   * Parse H264 elements and formats: RTP/H264, Annex-B, AVCC, anti-competition, DTS
   * Parse MPEG4-audio (AAC) elements and formats: RTP/MPEG4-audio, ADTS, MPEG4-audio configurations
+  * Parse Opus elements: RTP/Opus
 
 ## Table of contents
 
@@ -61,6 +62,7 @@ Features:
 * [client-read-codec-h264-convert-to-jpeg](examples/client-read-codec-h264-convert-to-jpeg/main.go)
 * [client-read-codec-h264-save-to-disk](examples/client-read-codec-h264-save-to-disk/main.go)
 * [client-read-codec-mpeg4audio](examples/client-read-codec-mpeg4audio/main.go)
+* [client-read-codec-opus](examples/client-read-codec-opus/main.go)
 * [client-read-partial](examples/client-read-partial/main.go)
 * [client-read-options](examples/client-read-options/main.go)
 * [client-read-pause](examples/client-read-pause/main.go)

examples/client-read-codec-opus/main.go

@@ -0,0 +1,81 @@
+package main
+
+import (
+	"log"
+
+	"github.com/aler9/gortsplib"
+	"github.com/aler9/gortsplib/pkg/rtpopus"
+	"github.com/aler9/gortsplib/pkg/url"
+)
+
+// This example shows how to
+// 1. connect to a RTSP server and read all tracks on a path
+// 2. check if there's an Opus track
+// 3. get Opus packets of that track
+
+func main() {
+	c := gortsplib.Client{}
+
+	// parse URL
+	u, err := url.Parse("rtsp://localhost:8554/mystream")
+	if err != nil {
+		panic(err)
+	}
+
+	// connect to the server
+	err = c.Start(u.Scheme, u.Host)
+	if err != nil {
+		panic(err)
+	}
+	defer c.Close()
+
+	// find published tracks
+	tracks, baseURL, _, err := c.Describe(u)
+	if err != nil {
+		panic(err)
+	}
+
+	// find the Opus track
+	opusTrack, opusTrackID := func() (*gortsplib.TrackOpus, int) {
+		for i, track := range tracks {
+			if tt, ok := track.(*gortsplib.TrackOpus); ok {
+				return tt, i
+			}
+		}
+		return nil, -1
+	}()
+	if opusTrack == nil {
+		panic("Opus track not found")
+	}
+
+	// setup decoder
+	dec := &rtpopus.Decoder{
+		SampleRate: opusTrack.SampleRate,
+	}
+	dec.Init()
+
+	// called when a RTP packet arrives
+	c.OnPacketRTP = func(ctx *gortsplib.ClientOnPacketRTPCtx) {
+		if ctx.TrackID != opusTrackID {
+			return
+		}
+
+		// decode an Opus packet from the RTP packet
+		op, _, err := dec.Decode(ctx.Packet)
+		if err != nil {
+			return
+		}
+
+		// print
+		log.Printf("received Opus packet of size %d\n", len(op))
+	}
+
+	// setup and read all tracks
+	err = c.SetupAndPlay(tracks, baseURL)
+	if err != nil {
+		panic(err)
+	}
+
+	// wait until a fatal error
+	panic(c.Wait())
+}

pkg/rtpopus/decoder.go

@@ -0,0 +1,31 @@
+package rtpopus
+
+import (
+	"time"
+
+	"github.com/pion/rtp"
+	"github.com/pion/rtp/codecs"
+
+	"github.com/aler9/gortsplib/pkg/rtptimedec"
+)
+
+// Decoder is a RTP/Opus decoder.
+type Decoder struct {
+	// sample rate of input packets.
+	SampleRate int
+
+	cop         codecs.OpusPacket
+	timeDecoder *rtptimedec.Decoder
+}
+
+// Init initializes the decoder.
+func (d *Decoder) Init() {
+	d.timeDecoder = rtptimedec.New(d.SampleRate)
+}
+
+// Decode decodes a Opus packet from a RTP/Opus packet.
+// It returns the Opus packet and its PTS.
+func (d *Decoder) Decode(pkt *rtp.Packet) ([]byte, time.Duration, error) {
+	_, err := d.cop.Unmarshal(pkt.Payload)
+	return d.cop.Payload, d.timeDecoder.Decode(pkt.Timestamp), err
+}

pkg/rtpopus/decoder_test.go

@@ -0,0 +1,67 @@
+package rtpopus
+
+import (
+	"testing"
+	"time"
+
+	"github.com/pion/rtp"
+	"github.com/stretchr/testify/require"
+)
+
+var cases = []struct {
+	name string
+	op   []byte
+	pts  time.Duration
+	pkt  *rtp.Packet
+}{
+	{
+		"a",
+		[]byte{0x01, 0x02, 0x03, 0x04},
+		20 * time.Millisecond,
+		&rtp.Packet{
+			Header: rtp.Header{
+				Version:        2,
+				Marker:         true,
+				PayloadType:    96,
+				SequenceNumber: 17645,
+				Timestamp:      2289527317,
+				SSRC:           0x9dbb7812,
+			},
+			Payload: []byte{0x01, 0x02, 0x03, 0x04},
+		},
+	},
+}
+
+func TestDecode(t *testing.T) {
+	for _, ca := range cases {
+		t.Run(ca.name, func(t *testing.T) {
+			d := &Decoder{
+				SampleRate: 48000,
+			}
+			d.Init()
+
+			// send an initial packet downstream
+			// in order to compute the right timestamp,
+			// that is relative to the initial packet
+			pkt := rtp.Packet{
+				Header: rtp.Header{
+					Version:        2,
+					Marker:         true,
+					PayloadType:    96,
+					SequenceNumber: 17645,
+					Timestamp:      2289526357,
+					SSRC:           0x9dbb7812,
+				},
+				Payload: []byte{0x00},
+			}
+
+			_, _, err := d.Decode(&pkt)
+			require.NoError(t, err)
+
+			op, pts, err := d.Decode(ca.pkt)
+			require.NoError(t, err)
+			require.Equal(t, ca.op, op)
+			require.Equal(t, ca.pts, pts)
+		})
+	}
+}

pkg/rtpopus/encoder.go

@@ -0,0 +1,96 @@
+package rtpopus
+
+import (
+	"crypto/rand"
+	"fmt"
+	"time"
+
+	"github.com/pion/rtp"
+	"github.com/pion/rtp/codecs"
+)
+
+const (
+	rtpVersion = 2
+)
+
+func randUint32() uint32 {
+	var b [4]byte
+	rand.Read(b[:])
+	return uint32(b[0])<<24 | uint32(b[1])<<16 | uint32(b[2])<<8 | uint32(b[3])
+}
+
+// Encoder is a RTP/Opus encoder.
+type Encoder struct {
+	// payload type of packets.
+	PayloadType uint8
+
+	// SSRC of packets (optional).
+	// It defaults to a random value.
+	SSRC *uint32
+
+	// initial sequence number of packets (optional).
+	// It defaults to a random value.
+	InitialSequenceNumber *uint16
+
+	// initial timestamp of packets (optional).
+	// It defaults to a random value.
+	InitialTimestamp *uint32
+
+	// maximum size of packet payloads (optional).
+	// It defaults to 1460.
+	PayloadMaxSize int
+
+	// sample rate of packets.
+	SampleRate int
+
+	sequenceNumber uint16
+	op             codecs.OpusPayloader
+}
+
+// Init initializes the encoder.
+func (e *Encoder) Init() {
+	if e.SSRC == nil {
+		v := randUint32()
+		e.SSRC = &v
+	}
+	if e.InitialSequenceNumber == nil {
+		v := uint16(randUint32())
+		e.InitialSequenceNumber = &v
+	}
+	if e.InitialTimestamp == nil {
+		v := randUint32()
+		e.InitialTimestamp = &v
+	}
+	if e.PayloadMaxSize == 0 {
+		e.PayloadMaxSize = 1460 // 1500 (UDP MTU) - 20 (IP header) - 8 (UDP header) - 12 (RTP header)
+	}
+
+	e.sequenceNumber = *e.InitialSequenceNumber
+}
+
+func (e *Encoder) encodeTimestamp(ts time.Duration) uint32 {
+	return *e.InitialTimestamp + uint32(ts.Seconds()*float64(e.SampleRate))
+}
+
+// Encode encodes an Opus packet into a RTP/Opus packet.
+func (e *Encoder) Encode(op []byte, pts time.Duration) (*rtp.Packet, error) {
+	if len(op) > e.PayloadMaxSize {
+		return nil, fmt.Errorf("packet size exceeds maximum size")
+	}
+
+	pkt := &rtp.Packet{
+		Header: rtp.Header{
+			Version:        rtpVersion,
+			PayloadType:    e.PayloadType,
+			SequenceNumber: e.sequenceNumber,
+			Timestamp:      e.encodeTimestamp(pts),
+			SSRC:           *e.SSRC,
+			Marker:         true,
+		},
+		Payload: e.op.Payload(0, op)[0],
+	}
+
+	e.sequenceNumber++
+
+	return pkt, nil
+}

pkg/rtpopus/encoder_test.go

@@ -0,0 +1,46 @@
+package rtpopus
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestEncode(t *testing.T) {
+	for _, ca := range cases {
+		t.Run(ca.name, func(t *testing.T) {
+			e := &Encoder{
+				PayloadType: 96,
+				SampleRate:  48000,
+				SSRC: func() *uint32 {
+					v := uint32(0x9dbb7812)
+					return &v
+				}(),
+				InitialSequenceNumber: func() *uint16 {
+					v := uint16(0x44ed)
+					return &v
+				}(),
+				InitialTimestamp: func() *uint32 {
+					v := uint32(0x88776655)
+					return &v
+				}(),
+			}
+			e.Init()
+
+			pkt, err := e.Encode(ca.op, ca.pts)
+			require.NoError(t, err)
+			require.Equal(t, ca.pkt, pkt)
+		})
+	}
+}
+
+func TestEncodeRandomInitialState(t *testing.T) {
+	e := &Encoder{
+		PayloadType: 96,
+		SampleRate:  48000,
+	}
+	e.Init()
+	require.NotEqual(t, nil, e.SSRC)
+	require.NotEqual(t, nil, e.InitialSequenceNumber)
+	require.NotEqual(t, nil, e.InitialTimestamp)
+}

pkg/rtpopus/rtpopus.go

@@ -0,0 +1,2 @@
+// Package rtpopus contains a RTP/Opus decoder and encoder.
+package rtpopus