Merged in rtp-client (pull request #186)

protocol/rtp: RTP client

Approved-by: kortschak <dan@kortschak.io>
Approved-by: Alan Noble <anoble@gmail.com>

protocol/rtp/client.go

@@ -0,0 +1,63 @@
+/*
+NAME
+  client.go
+
+DESCRIPTION
+  client.go provides an RTP client.
+
+AUTHOR
+  Saxon A. Nelson-Milton <saxon@ausocean.org>
+
+LICENSE
+  This is Copyright (C) 2019 the Australian Ocean Lab (AusOcean).
+
+  It is free software: you can redistribute it and/or modify them
+  under the terms of the GNU General Public License as published by the
+  Free Software Foundation, either version 3 of the License, or (at your
+  option) any later version.
+
+  It is distributed in the hope that it will be useful, but WITHOUT
+  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+  FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
+  for more details.
+
+  You should have received a copy of the GNU General Public License
+  in gpl.txt.  If not, see http://www.gnu.org/licenses.
+*/
+
+package rtp
+
+import (
+	"net"
+)
+
+// Client describes an RTP client that can receive an RTP stream and implements
+// io.Reader.
+type Client struct {
+	conn *net.UDPConn
+}
+
+// NewClient returns a pointer to a new Client.
+//
+// addr is the address of form <ip>:<port> that we expect to receive
+// RTP at.
+func NewClient(addr string) (*Client, error) {
+	c := &Client{}
+
+	a, err := net.ResolveUDPAddr("udp", addr)
+	if err != nil {
+		return nil, err
+	}
+
+	c.conn, err = net.ListenUDP("udp", a)
+	if err != nil {
+		return nil, err
+	}
+
+	return c, nil
+}
+
+// Read implements io.Reader. This wraps the Read for the connection.
+func (c *Client) Read(p []byte) (int, error) {
+	return c.conn.Read(p)
+}

protocol/rtp/client_test.go

@@ -0,0 +1,125 @@
+/*
+NAME
+  client_test.go
+
+DESCRIPTION
+  client_test.go provides testing utilities to check RTP client functionality
+  provided in client.go.
+
+AUTHOR
+  Saxon A. Nelson-Milton <saxon@ausocean.org>
+
+LICENSE
+  This is Copyright (C) 2019 the Australian Ocean Lab (AusOcean).
+
+  It is free software: you can redistribute it and/or modify them
+  under the terms of the GNU General Public License as published by the
+  Free Software Foundation, either version 3 of the License, or (at your
+  option) any later version.
+
+  It is distributed in the hope that it will be useful, but WITHOUT
+  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+  FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
+  for more details.
+
+  You should have received a copy of the GNU General Public License
+  in gpl.txt.  If not, see http://www.gnu.org/licenses.
+*/
+
+package rtp
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net"
+	"testing"
+)
+
+// TestReceive checks that the Client can correctly receive RTP packets and
+// perform a specificed operation on the packets before storing in the ringBuffer.
+func TestReceive(t *testing.T) {
+	const (
+		clientAddr    = "localhost:8000"
+		packetsToSend = 20
+	)
+
+	testErr := make(chan error)
+	serverErr := make(chan error)
+	done := make(chan struct{})
+	clientReady := make(chan struct{})
+	var c *Client
+
+	// Start routine to read from client.
+	go func() {
+		// Create and start the client.
+		var err error
+		c, err = NewClient(clientAddr)
+		if err != nil {
+			testErr <- fmt.Errorf("could not create client, failed with error: %v\n", err)
+		}
+		close(clientReady)
+
+		// Read packets using the client and check them with expected.
+		var packetsReceived int
+		buf := make([]byte, 4096)
+		for packetsReceived != packetsToSend {
+			n, err := c.Read(buf)
+			switch err {
+			case nil:
+			case io.EOF:
+				continue
+			default:
+				testErr <- fmt.Errorf("unexpected error from c.Read: %v\n", err)
+			}
+
+			// Create expected data and apply operation if there is one.
+			expect := (&Packet{V: rtpVer, Payload: []byte{byte(packetsReceived)}}).Bytes(nil)
+
+			// Compare.
+			got := buf[:n]
+			if !bytes.Equal(got, expect) {
+				testErr <- fmt.Errorf("did not get expected result. \nGot: %v\n Want: %v\n", got, expect)
+			}
+			packetsReceived++
+		}
+		close(done)
+	}()
+
+	// Start the RTP server.
+	go func() {
+		<-clientReady
+		cAddr, err := net.ResolveUDPAddr("udp", clientAddr)
+		if err != nil {
+			serverErr <- fmt.Errorf("could not resolve server address, failed with err: %v\n", err)
+		}
+
+		conn, err := net.DialUDP("udp", nil, cAddr)
+		if err != nil {
+			serverErr <- fmt.Errorf("could not dial udp, failed with err: %v\n", err)
+		}
+
+		// Send packets to the client.
+		for i := 0; i < packetsToSend; i++ {
+			p := (&Packet{V: rtpVer, Payload: []byte{byte(i)}}).Bytes(nil)
+			_, err := conn.Write(p)
+			if err != nil {
+				serverErr <- fmt.Errorf("could not write packet to conn, failed with err: %v\n", err)
+			}
+		}
+	}()
+
+	<-clientReady
+loop:
+	for {
+		select {
+		case err := <-testErr:
+			t.Fatal(err)
+		case err := <-serverErr:
+			t.Fatal(err)
+		case <-done:
+			break loop
+		default:
+		}
+	}
+}

protocol/rtp/encoder.go

@@ -97,7 +97,7 @@ func min(a, b int) int {
 // Encode takes a nalu unit and encodes it into an rtp packet and
 // writes to the io.Writer given in NewEncoder
 func (e *Encoder) Encode(payload []byte) error {
-	pkt := Pkt{
+	pkt := Packet{
 		V:       rtpVer,           // version
 		X:       false,            // header extension
 		CC:      0,                // CSRC count

protocol/rtp/parse_test.go

@@ -35,7 +35,7 @@ import (
 // TestVersion checks that we can correctly get the version from an RTP packet.
 func TestVersion(t *testing.T) {
 	const expect = 1
-	got := version((&Pkt{V: expect}).Bytes(nil))
+	got := version((&Packet{V: expect}).Bytes(nil))
 	if got != expect {
 		t.Errorf("unexpected version for RTP packet. Got: %v\n Want: %v\n", got, expect)
 	}
@@ -46,7 +46,7 @@ func TestVersion(t *testing.T) {
 func TestCsrcCount(t *testing.T) {
 	const ver, expect = 2, 2
 
-	pkt := (&Pkt{
+	pkt := (&Packet{
 		V:    ver,
 		CC:   expect,
 		CSRC: make([][4]byte, expect),
@@ -64,7 +64,7 @@ func TestHasExt(t *testing.T) {
 	const ver = 2
 
 	// First check for when there is an extension field.
-	pkt := &Pkt{
+	pkt := &Packet{
 		V: ver,
 		X: true,
 		Extension: ExtensionHeader{
@@ -93,19 +93,19 @@ func TestPayload(t *testing.T) {
 	expect := []byte{0x01, 0x02, 0x03, 0x04, 0x05}
 
 	testPkts := [][]byte{
-		(&Pkt{
+		(&Packet{
 			V:       ver,
 			Payload: expect,
 		}).Bytes(nil),
 
-		(&Pkt{
+		(&Packet{
 			V:       ver,
 			CC:      3,
 			CSRC:    make([][4]byte, 3),
 			Payload: expect,
 		}).Bytes(nil),
 
-		(&Pkt{
+		(&Packet{
 			V: ver,
 			X: true,
 			Extension: ExtensionHeader{
@@ -115,7 +115,7 @@ func TestPayload(t *testing.T) {
 			Payload: expect,
 		}).Bytes(nil),
 
-		(&Pkt{
+		(&Packet{
 			V:    ver,
 			CC:   3,
 			CSRC: make([][4]byte, 3),

protocol/rtp/rtp.go

@@ -46,7 +46,7 @@ const (
 
 // Pkt provides fields consistent with RFC3550 definition of an rtp packet
 // The padding indicator does not need to be set manually, only the padding length
-type Pkt struct {
+type Packet struct {
 	V         uint8           // Version (currently 2).
 	p         bool            // Padding indicator (0 => padding, 1 => padding).
 	X         bool            // Extension header indicator.
@@ -69,7 +69,7 @@ type ExtensionHeader struct {
 }
 
 // Bytes provides a byte slice of the packet
-func (p *Pkt) Bytes(buf []byte) []byte {
+func (p *Packet) Bytes(buf []byte) []byte {
 	// Calculate the required length for the RTP packet.
 	headerExtensionLen := 0
 	if p.X {

protocol/rtp/rtp_test.go

@@ -35,13 +35,13 @@ import (
 // TODO (saxon): add more tests
 var rtpTests = []struct {
 	num  int
-	pkt  Pkt
+	pkt  Packet
 	want []byte
 }{
 	// No padding, no CSRC and no extension.
 	{
 		num: 1,
-		pkt: Pkt{
+		pkt: Packet{
 			V:    2,
 			p:    false,
 			X:    false,
@@ -67,7 +67,7 @@ var rtpTests = []struct {
 	// With padding.
 	{
 		num: 2,
-		pkt: Pkt{
+		pkt: Packet{
 			V:    2,
 			p:    true,
 			X:    false,
@@ -101,7 +101,7 @@ var rtpTests = []struct {
 	// With padding and CSRC.
 	{
 		num: 3,
-		pkt: Pkt{
+		pkt: Packet{
 			V:    2,
 			p:    true,
 			X:    false,
@@ -141,7 +141,7 @@ var rtpTests = []struct {
 	// With padding, CSRC and extension.
 	{
 		num: 4,
-		pkt: Pkt{
+		pkt: Packet{
 			V:    2,
 			p:    true,
 			X:    true,