mirror
/
av
mirror of https://bitbucket.org/ausocean/av.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

protocol/rtp: commented file

This commit is contained in:
Saxon 2019-04-19 18:40:15 +09:30
parent b302eafa68
commit 190d546c58
1 changed files with 30 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
protocol/rtp/client.go
View File

@ -3,13 +3,15 @@ NAME
client.go client.go
DESCRIPTION DESCRIPTION
client.go provides an RTP client that will receive RTP on a UDP connection client.go provides an RTP client, Client, that will receive RTP on a UDP
connection, perform an operation on the packets and then store in a ringBuffer
that can be read from Read, an implemntation of io.Reader.
AUTHOR AUTHOR
Saxon A. Nelson-Milton <saxon@ausocean.org> Saxon A. Nelson-Milton <saxon@ausocean.org>
LICENSE LICENSE
rtp.go is Copyright (C) 2018 the Australian Ocean Lab (AusOcean) This is Copyright (C) 2019 the Australian Ocean Lab (AusOcean).
It is free software: you can redistribute it and/or modify them 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 under the terms of the GNU General Public License as published by the
@ -19,10 +21,10 @@ LICENSE
It is distributed in the hope that it will be useful, but WITHOUT It is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
for more details. for more details.
You should have received a copy of the GNU General Public License You should have received a copy of the GNU General Public License
along with revid in gpl.txt. If not, see [GNU licenses](http://www.gnu.org/licenses). in gpl.txt. If not, see http://www.gnu.org/licenses.
*/ */
package rtp package rtp
@ -39,15 +41,24 @@ const (
chanSize = 10 chanSize = 10
) )
// Client describes an RTP client that can receive an RTP stream and implements
// io.Reader.
type Client struct { type Client struct {
conn *net.UDPConn conn *net.UDPConn // The UDP connection RTP packets will be read from.
wg sync.WaitGroup wg sync.WaitGroup // Used to wait for recv routine to finish.
done chan struct{} done chan struct{} // Used to terminate the recv routine.
ring *ring.Buffer ring *ring.Buffer // Processed data from RTP packets will be stored here.
op func([]byte) ([]byte, error) op func([]byte) ([]byte, error) // The operation to perform on received RTP packets before storing.
ErrChan chan error ErrChan chan error // Errors encountered during recv will be sent to this chan.
} }
// NewClient returns a pointer to a new Client.
//
// addr is the address of form <ip>:<port> that we expect to receive
// RTP at. op is a function, if non nil, that will be used to perform an
// operation on each received RTP packet. For example, the op func may parse
// out the RTP payload. The result of the operation is then what is stored
// in the ringBuffer for reading.
func NewClient(addr string, op func([]byte) ([]byte, error)) (*Client, error) { func NewClient(addr string, op func([]byte) ([]byte, error)) (*Client, error) {
c := &Client{ c := &Client{
done: make(chan struct{}, 10), done: make(chan struct{}, 10),
@ -68,11 +79,15 @@ func NewClient(addr string, op func([]byte) ([]byte, error)) (*Client, error) {
return c, nil return c, nil
} }
// Start will start the recv routine.
func (c *Client) Start() { func (c *Client) Start() {
c.wg.Add(1) c.wg.Add(1)
go c.recv() go c.recv()
} }
// recv will read RTP packets from the UDP connection, perform the operation
// on them given at creation of the Client and then store the result in the
// ringBuffer for Reading.
func (c *Client) recv() { func (c *Client) recv() {
defer c.wg.Done() defer c.wg.Done()
buf := make([]byte, 4096) buf := make([]byte, 4096)
@ -107,12 +122,17 @@ func (c *Client) recv() {
} }
} }
// Stop will send a done signal to the receive routine, and also close the
// connection.
func (c *Client) Stop() { func (c *Client) Stop() {
close(c.done) close(c.done)
c.conn.Close() c.conn.Close()
c.wg.Wait() c.wg.Wait()
} }
// Read implements io.Reader.
//
// Read will get the next chunk from the ringBuffer and copy the bytes to p.
func (c *Client) Read(p []byte) (int, error) { func (c *Client) Read(p []byte) (int, error) {
chunk, err := c.ring.Next(0) chunk, err := c.ring.Next(0)
if err != nil { if err != nil {