mirror of https://bitbucket.org/ausocean/av.git
revid: moved code relating to inputs to separate file called inputs.go
This commit is contained in:
Saxon
parent
fc5edb9adc
commit
a37b250e3c
|
|
@ -0,0 +1,281 @@
|
||||||
|
/*
|
||||||
|
DESCRIPTION
|
||||||
|
inputs.go contains code for interfacing with various revid inputs to obtain
|
||||||
|
input data streams.
|
||||||
|
|
||||||
|
AUTHORS
|
||||||
|
Saxon A. Nelson-Milton <saxon@ausocean.org>
|
||||||
|
Alan Noble <alan@ausocean.org>
|
||||||
|
Dan Kortschak <dan@ausocean.org>
|
||||||
|
Trek Hopton <trek@ausocean.org>
|
||||||
|
|
||||||
|
LICENSE
|
||||||
|
revid is Copyright (C) 2017-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 revid
|
||||||
|
|
||||||
|
import (
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"net"
|
||||||
|
"os"
|
||||||
|
"os/exec"
|
||||||
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"bitbucket.org/ausocean/av/codec/codecutil"
|
||||||
|
"bitbucket.org/ausocean/av/protocol/rtcp"
|
||||||
|
"bitbucket.org/ausocean/av/protocol/rtp"
|
||||||
|
"bitbucket.org/ausocean/av/protocol/rtsp"
|
||||||
|
"bitbucket.org/ausocean/utils/logger"
|
||||||
|
)
|
||||||
|
|
||||||
|
// startRaspivid sets up things for input from raspivid i.e. starts
|
||||||
|
// a raspivid process and pipes it's data output.
|
||||||
|
func (r *Revid) startRaspivid() (func() error, error) {
|
||||||
|
r.config.Logger.Log(logger.Info, pkg+"starting raspivid")
|
||||||
|
|
||||||
|
const disabled = "0"
|
||||||
|
args := []string{
|
||||||
|
"--output", "-",
|
||||||
|
"--nopreview",
|
||||||
|
"--timeout", disabled,
|
||||||
|
"--width", fmt.Sprint(r.config.Width),
|
||||||
|
"--height", fmt.Sprint(r.config.Height),
|
||||||
|
"--bitrate", fmt.Sprint(r.config.Bitrate),
|
||||||
|
"--framerate", fmt.Sprint(r.config.FrameRate),
|
||||||
|
"--rotation", fmt.Sprint(r.config.Rotation),
|
||||||
|
"--brightness", fmt.Sprint(r.config.Brightness),
|
||||||
|
"--saturation", fmt.Sprint(r.config.Saturation),
|
||||||
|
"--exposure", fmt.Sprint(r.config.Exposure),
|
||||||
|
"--awb", fmt.Sprint(r.config.AutoWhiteBalance),
|
||||||
|
}
|
||||||
|
|
||||||
|
if r.config.FlipHorizontal {
|
||||||
|
args = append(args, "--hflip")
|
||||||
|
}
|
||||||
|
|
||||||
|
if r.config.FlipVertical {
|
||||||
|
args = append(args, "--vflip")
|
||||||
|
}
|
||||||
|
if r.config.FlipHorizontal {
|
||||||
|
args = append(args, "--hflip")
|
||||||
|
}
|
||||||
|
|
||||||
|
switch r.config.InputCodec {
|
||||||
|
default:
|
||||||
|
return nil, fmt.Errorf("revid: invalid input codec: %v", r.config.InputCodec)
|
||||||
|
case codecutil.H264:
|
||||||
|
args = append(args,
|
||||||
|
"--codec", "H264",
|
||||||
|
"--inline",
|
||||||
|
"--intra", fmt.Sprint(r.config.MinFrames),
|
||||||
|
)
|
||||||
|
if r.config.Quantization != 0 {
|
||||||
|
args = append(args, "-qp", fmt.Sprint(r.config.Quantization))
|
||||||
|
}
|
||||||
|
case codecutil.MJPEG:
|
||||||
|
args = append(args, "--codec", "MJPEG")
|
||||||
|
}
|
||||||
|
r.config.Logger.Log(logger.Info, pkg+"raspivid args", "raspividArgs", strings.Join(args, " "))
|
||||||
|
r.cmd = exec.Command("raspivid", args...)
|
||||||
|
|
||||||
|
stdout, err := r.cmd.StdoutPipe()
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
err = r.cmd.Start()
|
||||||
|
if err != nil {
|
||||||
|
r.config.Logger.Log(logger.Fatal, pkg+"cannot start raspivid", "error", err.Error())
|
||||||
|
}
|
||||||
|
|
||||||
|
r.wg.Add(1)
|
||||||
|
go r.processFrom(stdout, 0)
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// startV4l sets up webcam input and starts the revid.processFrom routine.
|
||||||
|
func (r *Revid) startV4L() (func() error, error) {
|
||||||
|
const defaultVideo = "/dev/video0"
|
||||||
|
|
||||||
|
r.config.Logger.Log(logger.Info, pkg+"starting webcam")
|
||||||
|
if r.config.InputPath == "" {
|
||||||
|
r.config.Logger.Log(logger.Info, pkg+"using default video device", "device", defaultVideo)
|
||||||
|
r.config.InputPath = defaultVideo
|
||||||
|
}
|
||||||
|
|
||||||
|
args := []string{
|
||||||
|
"-i", r.config.InputPath,
|
||||||
|
"-f", "h264",
|
||||||
|
"-r", fmt.Sprint(r.config.FrameRate),
|
||||||
|
}
|
||||||
|
|
||||||
|
args = append(args,
|
||||||
|
"-b:v", fmt.Sprint(r.config.Bitrate),
|
||||||
|
"-maxrate", fmt.Sprint(r.config.Bitrate),
|
||||||
|
"-bufsize", fmt.Sprint(r.config.Bitrate/2),
|
||||||
|
"-s", fmt.Sprintf("%dx%d", r.config.Width, r.config.Height),
|
||||||
|
"-",
|
||||||
|
)
|
||||||
|
|
||||||
|
r.config.Logger.Log(logger.Info, pkg+"ffmpeg args", "args", strings.Join(args, " "))
|
||||||
|
r.cmd = exec.Command("ffmpeg", args...)
|
||||||
|
|
||||||
|
stdout, err := r.cmd.StdoutPipe()
|
||||||
|
if err != nil {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
err = r.cmd.Start()
|
||||||
|
if err != nil {
|
||||||
|
r.config.Logger.Log(logger.Fatal, pkg+"cannot start webcam", "error", err.Error())
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
r.wg.Add(1)
|
||||||
|
go r.processFrom(stdout, time.Duration(0))
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// setupInputForFile sets up input from file and starts the revid.processFrom
|
||||||
|
// routine.
|
||||||
|
func (r *Revid) setupInputForFile() (func() error, error) {
|
||||||
|
f, err := os.Open(r.config.InputPath)
|
||||||
|
if err != nil {
|
||||||
|
r.config.Logger.Log(logger.Error, err.Error())
|
||||||
|
r.Stop()
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// TODO(kortschak): Maybe we want a context.Context-aware parser that we can stop.
|
||||||
|
r.wg.Add(1)
|
||||||
|
go r.processFrom(f, 0)
|
||||||
|
return func() error { return f.Close() }, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// startRTSPCamera uses RTSP to request an RTP stream from an IP camera. An RTP
|
||||||
|
// client is created from which RTP packets containing either h264/h265 can be
|
||||||
|
// read by the selected lexer.
|
||||||
|
func (r *Revid) startRTSPCamera() (func() error, error) {
|
||||||
|
rtspClt, local, remote, err := rtsp.NewClient(r.config.RTSPURL)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
resp, err := rtspClt.Options()
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
r.config.Logger.Log(logger.Info, pkg+"RTSP OPTIONS response", "response", resp.String())
|
||||||
|
|
||||||
|
resp, err = rtspClt.Describe()
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
r.config.Logger.Log(logger.Info, pkg+"RTSP DESCRIBE response", "response", resp.String())
|
||||||
|
|
||||||
|
resp, err = rtspClt.Setup("track1", fmt.Sprintf("RTP/AVP;unicast;client_port=%d-%d", rtpPort, rtcpPort))
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
r.config.Logger.Log(logger.Info, pkg+"RTSP SETUP response", "response", resp.String())
|
||||||
|
rtpCltAddr, rtcpCltAddr, rtcpSvrAddr, err := formAddrs(local, remote, *resp)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
resp, err = rtspClt.Play()
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
r.config.Logger.Log(logger.Info, pkg+"RTSP server PLAY response", "response", resp.String())
|
||||||
|
|
||||||
|
rtpClt, err := rtp.NewClient(rtpCltAddr)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
rtcpClt, err := rtcp.NewClient(rtcpCltAddr, rtcpSvrAddr, rtpClt, r.config.Logger.Log)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check errors from RTCP client until it has stopped running.
|
||||||
|
go func() {
|
||||||
|
for {
|
||||||
|
err, ok := <-rtcpClt.Err()
|
||||||
|
if ok {
|
||||||
|
r.config.Logger.Log(logger.Warning, pkg+"RTCP error", "error", err.Error())
|
||||||
|
} else {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
// Start the RTCP client.
|
||||||
|
rtcpClt.Start()
|
||||||
|
|
||||||
|
// Start reading data from the RTP client.
|
||||||
|
r.wg.Add(1)
|
||||||
|
go r.processFrom(rtpClt, time.Second/time.Duration(r.config.FrameRate))
|
||||||
|
|
||||||
|
return func() error {
|
||||||
|
rtspClt.Close()
|
||||||
|
rtcpClt.Stop()
|
||||||
|
return nil
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// formAddrs is a helper function to form the addresses for the RTP client,
|
||||||
|
// RTCP client, and the RTSP server's RTCP addr using the local, remote addresses
|
||||||
|
// of the RTSP conn, and the SETUP method response.
|
||||||
|
func formAddrs(local, remote *net.TCPAddr, setupResp rtsp.Response) (rtpCltAddr, rtcpCltAddr, rtcpSvrAddr string, err error) {
|
||||||
|
svrRTCPPort, err := parseSvrRTCPPort(setupResp)
|
||||||
|
if err != nil {
|
||||||
|
return "", "", "", err
|
||||||
|
}
|
||||||
|
rtpCltAddr = strings.Split(local.String(), ":")[0] + ":" + strconv.Itoa(rtpPort)
|
||||||
|
rtcpCltAddr = strings.Split(local.String(), ":")[0] + ":" + strconv.Itoa(rtcpPort)
|
||||||
|
rtcpSvrAddr = strings.Split(remote.String(), ":")[0] + ":" + strconv.Itoa(svrRTCPPort)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// parseServerRTCPPort is a helper function to get the RTSP server's RTCP port.
|
||||||
|
func parseSvrRTCPPort(resp rtsp.Response) (int, error) {
|
||||||
|
transport := resp.Header.Get("Transport")
|
||||||
|
for _, p := range strings.Split(transport, ";") {
|
||||||
|
if strings.Contains(p, "server_port") {
|
||||||
|
port, err := strconv.Atoi(strings.Split(p, "-")[1])
|
||||||
|
if err != nil {
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
return port, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return 0, errors.New("SETUP response did not provide RTCP port")
|
||||||
|
}
|
||||||
|
|
||||||
|
// processFrom is run as a routine to read from a input data source, lex and
|
||||||
|
// then send individual access units to revid's encoders.
|
||||||
|
func (r *Revid) processFrom(read io.Reader, delay time.Duration) {
|
||||||
|
r.config.Logger.Log(logger.Info, pkg+"reading input data")
|
||||||
|
r.err <- r.lexTo(r.encoders, read, delay)
|
||||||
|
r.config.Logger.Log(logger.Info, pkg+"finished reading input data")
|
||||||
|
r.wg.Done()
|
||||||
|
}
|
||||||
235
revid/revid.go
235
revid/revid.go
|
|
@ -32,8 +32,6 @@ import (
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"net"
|
|
||||||
"os"
|
|
||||||
"os/exec"
|
"os/exec"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
@ -46,9 +44,6 @@ import (
|
||||||
"bitbucket.org/ausocean/av/codec/mjpeg"
|
"bitbucket.org/ausocean/av/codec/mjpeg"
|
||||||
"bitbucket.org/ausocean/av/container/flv"
|
"bitbucket.org/ausocean/av/container/flv"
|
||||||
"bitbucket.org/ausocean/av/container/mts"
|
"bitbucket.org/ausocean/av/container/mts"
|
||||||
"bitbucket.org/ausocean/av/protocol/rtcp"
|
|
||||||
"bitbucket.org/ausocean/av/protocol/rtp"
|
|
||||||
"bitbucket.org/ausocean/av/protocol/rtsp"
|
|
||||||
"bitbucket.org/ausocean/iot/pi/netsender"
|
"bitbucket.org/ausocean/iot/pi/netsender"
|
||||||
"bitbucket.org/ausocean/utils/ioext"
|
"bitbucket.org/ausocean/utils/ioext"
|
||||||
"bitbucket.org/ausocean/utils/logger"
|
"bitbucket.org/ausocean/utils/logger"
|
||||||
|
|
@ -576,233 +571,3 @@ func (r *Revid) Update(vars map[string]string) error {
|
||||||
r.config.Logger.Log(logger.Info, pkg+"revid config changed", "config", fmt.Sprintf("%+v", r.config))
|
r.config.Logger.Log(logger.Info, pkg+"revid config changed", "config", fmt.Sprintf("%+v", r.config))
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// startRaspivid sets up things for input from raspivid i.e. starts
|
|
||||||
// a raspivid process and pipes it's data output.
|
|
||||||
func (r *Revid) startRaspivid() (func() error, error) {
|
|
||||||
r.config.Logger.Log(logger.Info, pkg+"starting raspivid")
|
|
||||||
|
|
||||||
const disabled = "0"
|
|
||||||
args := []string{
|
|
||||||
"--output", "-",
|
|
||||||
"--nopreview",
|
|
||||||
"--timeout", disabled,
|
|
||||||
"--width", fmt.Sprint(r.config.Width),
|
|
||||||
"--height", fmt.Sprint(r.config.Height),
|
|
||||||
"--bitrate", fmt.Sprint(r.config.Bitrate),
|
|
||||||
"--framerate", fmt.Sprint(r.config.FrameRate),
|
|
||||||
"--rotation", fmt.Sprint(r.config.Rotation),
|
|
||||||
"--brightness", fmt.Sprint(r.config.Brightness),
|
|
||||||
"--saturation", fmt.Sprint(r.config.Saturation),
|
|
||||||
"--exposure", fmt.Sprint(r.config.Exposure),
|
|
||||||
"--awb", fmt.Sprint(r.config.AutoWhiteBalance),
|
|
||||||
}
|
|
||||||
|
|
||||||
if r.config.FlipHorizontal {
|
|
||||||
args = append(args, "--hflip")
|
|
||||||
}
|
|
||||||
|
|
||||||
if r.config.FlipVertical {
|
|
||||||
args = append(args, "--vflip")
|
|
||||||
}
|
|
||||||
if r.config.FlipHorizontal {
|
|
||||||
args = append(args, "--hflip")
|
|
||||||
}
|
|
||||||
|
|
||||||
switch r.config.InputCodec {
|
|
||||||
default:
|
|
||||||
return nil, fmt.Errorf("revid: invalid input codec: %v", r.config.InputCodec)
|
|
||||||
case codecutil.H264:
|
|
||||||
args = append(args,
|
|
||||||
"--codec", "H264",
|
|
||||||
"--inline",
|
|
||||||
"--intra", fmt.Sprint(r.config.MinFrames),
|
|
||||||
)
|
|
||||||
if r.config.Quantization != 0 {
|
|
||||||
args = append(args, "-qp", fmt.Sprint(r.config.Quantization))
|
|
||||||
}
|
|
||||||
case codecutil.MJPEG:
|
|
||||||
args = append(args, "--codec", "MJPEG")
|
|
||||||
}
|
|
||||||
r.config.Logger.Log(logger.Info, pkg+"raspivid args", "raspividArgs", strings.Join(args, " "))
|
|
||||||
r.cmd = exec.Command("raspivid", args...)
|
|
||||||
|
|
||||||
stdout, err := r.cmd.StdoutPipe()
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
err = r.cmd.Start()
|
|
||||||
if err != nil {
|
|
||||||
r.config.Logger.Log(logger.Fatal, pkg+"cannot start raspivid", "error", err.Error())
|
|
||||||
}
|
|
||||||
|
|
||||||
r.wg.Add(1)
|
|
||||||
go r.processFrom(stdout, 0)
|
|
||||||
return nil, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (r *Revid) startV4L() (func() error, error) {
|
|
||||||
const defaultVideo = "/dev/video0"
|
|
||||||
|
|
||||||
r.config.Logger.Log(logger.Info, pkg+"starting webcam")
|
|
||||||
if r.config.InputPath == "" {
|
|
||||||
r.config.Logger.Log(logger.Info, pkg+"using default video device", "device", defaultVideo)
|
|
||||||
r.config.InputPath = defaultVideo
|
|
||||||
}
|
|
||||||
|
|
||||||
args := []string{
|
|
||||||
"-i", r.config.InputPath,
|
|
||||||
"-f", "h264",
|
|
||||||
"-r", fmt.Sprint(r.config.FrameRate),
|
|
||||||
}
|
|
||||||
|
|
||||||
args = append(args,
|
|
||||||
"-b:v", fmt.Sprint(r.config.Bitrate),
|
|
||||||
"-maxrate", fmt.Sprint(r.config.Bitrate),
|
|
||||||
"-bufsize", fmt.Sprint(r.config.Bitrate/2),
|
|
||||||
"-s", fmt.Sprintf("%dx%d", r.config.Width, r.config.Height),
|
|
||||||
"-",
|
|
||||||
)
|
|
||||||
|
|
||||||
r.config.Logger.Log(logger.Info, pkg+"ffmpeg args", "args", strings.Join(args, " "))
|
|
||||||
r.cmd = exec.Command("ffmpeg", args...)
|
|
||||||
|
|
||||||
stdout, err := r.cmd.StdoutPipe()
|
|
||||||
if err != nil {
|
|
||||||
return nil, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
err = r.cmd.Start()
|
|
||||||
if err != nil {
|
|
||||||
r.config.Logger.Log(logger.Fatal, pkg+"cannot start webcam", "error", err.Error())
|
|
||||||
return nil, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
r.wg.Add(1)
|
|
||||||
go r.processFrom(stdout, time.Duration(0))
|
|
||||||
return nil, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// setupInputForFile sets things up for getting input from a file
|
|
||||||
func (r *Revid) setupInputForFile() (func() error, error) {
|
|
||||||
f, err := os.Open(r.config.InputPath)
|
|
||||||
if err != nil {
|
|
||||||
r.config.Logger.Log(logger.Error, err.Error())
|
|
||||||
r.Stop()
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
// TODO(kortschak): Maybe we want a context.Context-aware parser that we can stop.
|
|
||||||
r.wg.Add(1)
|
|
||||||
go r.processFrom(f, 0)
|
|
||||||
return func() error { return f.Close() }, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// startRTSPCamera uses RTSP to request an RTP stream from an IP camera. An RTP
|
|
||||||
// client is created from which RTP packets containing either h264/h265 can read
|
|
||||||
// by the selected lexer.
|
|
||||||
func (r *Revid) startRTSPCamera() (func() error, error) {
|
|
||||||
rtspClt, local, remote, err := rtsp.NewClient(r.config.RTSPURL)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
resp, err := rtspClt.Options()
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
r.config.Logger.Log(logger.Info, pkg+"RTSP OPTIONS response", "response", resp.String())
|
|
||||||
|
|
||||||
resp, err = rtspClt.Describe()
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
r.config.Logger.Log(logger.Info, pkg+"RTSP DESCRIBE response", "response", resp.String())
|
|
||||||
|
|
||||||
resp, err = rtspClt.Setup("track1", fmt.Sprintf("RTP/AVP;unicast;client_port=%d-%d", rtpPort, rtcpPort))
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
r.config.Logger.Log(logger.Info, pkg+"RTSP SETUP response", "response", resp.String())
|
|
||||||
rtpCltAddr, rtcpCltAddr, rtcpSvrAddr, err := formAddrs(local, remote, *resp)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
resp, err = rtspClt.Play()
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
r.config.Logger.Log(logger.Info, pkg+"RTSP server PLAY response", "response", resp.String())
|
|
||||||
|
|
||||||
rtpClt, err := rtp.NewClient(rtpCltAddr)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
rtcpClt, err := rtcp.NewClient(rtcpCltAddr, rtcpSvrAddr, rtpClt, r.config.Logger.Log)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Check errors from RTCP client until it has stopped running.
|
|
||||||
go func() {
|
|
||||||
for {
|
|
||||||
err, ok := <-rtcpClt.Err()
|
|
||||||
if ok {
|
|
||||||
r.config.Logger.Log(logger.Warning, pkg+"RTCP error", "error", err.Error())
|
|
||||||
} else {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
// Start the RTCP client.
|
|
||||||
rtcpClt.Start()
|
|
||||||
|
|
||||||
// Start reading data from the RTP client.
|
|
||||||
r.wg.Add(1)
|
|
||||||
go r.processFrom(rtpClt, time.Second/time.Duration(r.config.FrameRate))
|
|
||||||
|
|
||||||
return func() error {
|
|
||||||
rtspClt.Close()
|
|
||||||
rtcpClt.Stop()
|
|
||||||
return nil
|
|
||||||
}, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// formAddrs is a helper function to form the addresses for the RTP client,
|
|
||||||
// RTCP client, and the RTSP server's RTCP addr using the local, remote addresses
|
|
||||||
// of the RTSP conn, and the SETUP method response.
|
|
||||||
func formAddrs(local, remote *net.TCPAddr, setupResp rtsp.Response) (rtpCltAddr, rtcpCltAddr, rtcpSvrAddr string, err error) {
|
|
||||||
svrRTCPPort, err := parseSvrRTCPPort(setupResp)
|
|
||||||
if err != nil {
|
|
||||||
return "", "", "", err
|
|
||||||
}
|
|
||||||
rtpCltAddr = strings.Split(local.String(), ":")[0] + ":" + strconv.Itoa(rtpPort)
|
|
||||||
rtcpCltAddr = strings.Split(local.String(), ":")[0] + ":" + strconv.Itoa(rtcpPort)
|
|
||||||
rtcpSvrAddr = strings.Split(remote.String(), ":")[0] + ":" + strconv.Itoa(svrRTCPPort)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
// parseServerRTCPPort is a helper function to get the RTSP server's RTCP port.
|
|
||||||
func parseSvrRTCPPort(resp rtsp.Response) (int, error) {
|
|
||||||
transport := resp.Header.Get("Transport")
|
|
||||||
for _, p := range strings.Split(transport, ";") {
|
|
||||||
if strings.Contains(p, "server_port") {
|
|
||||||
port, err := strconv.Atoi(strings.Split(p, "-")[1])
|
|
||||||
if err != nil {
|
|
||||||
return 0, err
|
|
||||||
}
|
|
||||||
return port, nil
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return 0, errors.New("SETUP response did not provide RTCP port")
|
|
||||||
}
|
|
||||||
|
|
||||||
func (r *Revid) processFrom(read io.Reader, delay time.Duration) {
|
|
||||||
r.config.Logger.Log(logger.Info, pkg+"reading input data")
|
|
||||||
r.err <- r.lexTo(r.encoders, read, delay)
|
|
||||||
r.config.Logger.Log(logger.Info, pkg+"finished reading input data")
|
|
||||||
r.wg.Done()
|
|
||||||
}
|
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue