av/revid/Revid.go

689 lines
18 KiB
Go
Raw Normal View History

/*
NAME
2018-04-16 08:12:16 +03:00
Revid.go
DESCRIPTION
See Readme.md
AUTHORS
Saxon A. Nelson-Milton <saxon@ausocean.org>
Alan Noble <alan@ausocean.org>
LICENSE
revid is Copyright (C) 2017 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
along with revid in gpl.txt. If not, see [GNU licenses](http://www.gnu.org/licenses).
*/
// revid is a testbed for re-muxing and re-directing video streams as MPEG-TS over various protocols.
package revid
import (
"bufio"
"bytes"
2018-01-31 04:00:03 +03:00
"errors"
"fmt"
"io"
"io/ioutil"
"net/http"
"os"
"os/exec"
"strconv"
2018-03-17 16:29:42 +03:00
"sync"
2018-04-16 07:54:21 +03:00
"time"
2018-02-14 10:02:57 +03:00
2018-03-13 08:32:52 +03:00
"bitbucket.org/ausocean/av/generator"
2018-03-14 04:18:03 +03:00
"bitbucket.org/ausocean/av/parser"
2018-03-13 08:15:42 +03:00
"bitbucket.org/ausocean/av/ringbuffer"
2018-03-13 11:29:18 +03:00
"bitbucket.org/ausocean/av/rtmp"
2018-03-14 04:18:03 +03:00
/*
"../generator"
"../parser"
"../ringbuffer"
"../rtmp"
*/)
// Misc constants
const (
clipDuration = 1 // s
mp2tPacketSize = 188 // MPEG-TS packet size
mp2tMaxPackets = 2016 * clipDuration // # first multiple of 7 and 8 greater than 2000
2018-03-14 01:11:33 +03:00
ringBufferSize = 1000
ringBufferElementSize = 150000
httpTimeOut = 5 // s
packetsPerFrame = 7
2018-03-13 12:14:30 +03:00
h264BufferSize = 1000000
2018-04-19 12:48:29 +03:00
bitrateTime = 10 // s
mjpegParserInChanLen = 100000
2018-04-16 08:13:29 +03:00
ffmpegPath = "/usr/local/bin/ffmpeg"
2018-03-14 04:18:03 +03:00
rtmpConnectionTimout = 10
outputChanSize = 10000
2018-04-16 08:32:00 +03:00
cameraRetryPeriod = 5 * time.Second
2018-04-24 08:38:30 +03:00
sendFailedDelay = 5
maxSendFailedErrorCount = 500
2018-04-24 08:49:53 +03:00
clipSizeThreshold = 11
2018-04-30 23:10:41 +03:00
rtmpConnectionMaxTries = 5
2018-01-10 06:57:56 +03:00
)
2018-01-24 07:12:22 +03:00
// Log Types
2018-01-24 04:40:03 +03:00
const (
2018-01-24 07:12:22 +03:00
Error = "Error"
2018-01-24 04:40:03 +03:00
Warning = "Warning"
2018-01-24 07:12:22 +03:00
Info = "Info"
Debug = "Debug"
2018-01-24 04:40:03 +03:00
)
2018-04-17 08:06:45 +03:00
2018-04-16 08:12:16 +03:00
// Revid provides methods to control a revid session; providing methods
// to start, stop and change the state of an instance using the Config struct.
2018-04-16 08:12:16 +03:00
type Revid interface {
2018-01-10 06:57:56 +03:00
Start()
Stop()
2018-03-17 16:29:42 +03:00
changeState(newconfig Config) error
2018-01-23 06:15:06 +03:00
GetConfigRef() *Config
2018-01-24 05:31:33 +03:00
Log(logType, m string)
2018-01-24 05:26:33 +03:00
IsRunning() bool
}
2018-04-16 08:12:16 +03:00
// The revid struct provides fields to describe the state of a Revid.
type revid struct {
ffmpegPath string
tempDir string
ringBuffer ringbuffer.RingBuffer
config Config
isRunning bool
outputFile *os.File
inputFile *os.File
generator generator.Generator
parser parser.Parser
cmd *exec.Cmd
ffmpegCmd *exec.Cmd
inputReader *bufio.Reader
ffmpegStdin io.WriteCloser
outputChan chan []byte
setupInput func() error
setupOutput func() error
getFrame func() []byte
2018-03-14 04:18:03 +03:00
sendClip func(clip []byte) error
rtmpInst rtmp.RTMPSession
2018-04-16 07:54:21 +03:00
mutex sync.Mutex
2018-04-24 08:38:30 +03:00
currentBitrate int64
2018-01-23 06:15:06 +03:00
}
2018-04-16 08:17:50 +03:00
// NewRevid returns a pointer to a new revid with the desired
// configuration, and/or an error if construction of the new instant was not
// successful.
2018-04-16 08:17:50 +03:00
func NewRevid(config Config) (r *revid, err error) {
2018-04-16 08:12:16 +03:00
r = new(revid)
2018-03-17 16:29:42 +03:00
r.mutex = sync.Mutex{}
2018-01-31 09:13:04 +03:00
r.ringBuffer = ringbuffer.NewRingBuffer(ringBufferSize, ringBufferElementSize)
2018-03-17 16:29:42 +03:00
err = r.changeState(config)
if err != nil {
2018-03-17 16:29:42 +03:00
r = nil
return
}
2018-03-13 07:33:31 +03:00
r.outputChan = make(chan []byte, outputChanSize)
r.parser.Start()
go r.packClips()
return
}
2018-04-24 08:38:30 +03:00
// Returns the currently saved bitrate from the most recent bitrate check
// check bitrate output delay in consts for this period
func (r *revid) GetBitrate() int64 {
return r.currentBitrate
}
// GetConfigRef returns a pointer to the revidInst's Config struct object
2018-04-16 08:12:16 +03:00
func (r *revid) GetConfigRef() *Config {
return &r.config
}
2018-04-16 08:12:16 +03:00
// changeState swaps the current config of a revid with the passed
// configuration; checking validity and returning errors if not valid.
2018-04-16 08:12:16 +03:00
func (r *revid) changeState(config Config) error {
r.config.Logger = config.Logger
2018-02-10 10:08:14 +03:00
err := config.Validate(r)
if err != nil {
return errors.New("Config struct is bad!: " + err.Error())
}
r.config = config
2018-01-23 06:15:06 +03:00
switch r.config.Output {
case File:
r.sendClip = r.sendClipToFile
r.setupOutput = r.setupOutputForFile
case FfmpegRtmp:
r.setupOutput = r.setupOutputForFfmpegRtmp
r.sendClip = r.sendClipToFfmpegRtmp
case NativeRtmp:
r.setupOutput = r.setupOutputForLibRtmp
r.sendClip = r.sendClipToLibRtmp
case Http:
r.sendClip = r.sendClipToHTTP
}
switch r.config.Input {
case Raspivid:
r.setupInput = r.setupInputForRaspivid
case File:
r.setupInput = r.setupInputForFile
2018-01-10 04:32:16 +03:00
}
switch r.config.InputCodec {
case H264:
r.Log(Info, "Using H264 parser!")
r.parser = parser.NewH264Parser()
case Mjpeg:
r.Log(Info, "Using MJPEG parser!")
r.parser = parser.NewMJPEGParser(mjpegParserInChanLen)
}
switch r.config.Packetization {
case None:
// no packetisation - Revid output chan grabs raw data straight from parser
r.parser.SetOutputChan(r.outputChan)
r.getFrame = r.getFrameNoPacketization
goto noPacketizationSetup
case Mpegts:
r.Log(Info, "Using MPEGTS packetisation!")
frameRateAsInt, _ := strconv.Atoi(r.config.FrameRate)
r.generator = generator.NewTsGenerator(uint(frameRateAsInt))
case Flv:
r.Log(Info, "Using FLV packetisation!")
frameRateAsInt, _ := strconv.Atoi(r.config.FrameRate)
r.generator = generator.NewFlvGenerator(true, true, uint(frameRateAsInt))
}
// We have packetization of some sort, so we want to send data to Generator
// to perform packetization
r.getFrame = r.getFramePacketization
r.parser.SetOutputChan(r.generator.GetInputChan())
r.generator.Start()
2018-03-14 04:18:03 +03:00
noPacketizationSetup:
return nil
2018-01-10 04:32:16 +03:00
}
2018-03-17 16:29:42 +03:00
// ChangeConfig changes the current configuration of the revid instance.
2018-04-16 08:12:16 +03:00
func (r *revid) ChangeConfig(config Config) (err error) {
2018-03-17 16:29:42 +03:00
r.Stop()
2018-04-16 08:17:50 +03:00
r, err = NewRevid(config)
2018-03-17 16:29:42 +03:00
if err != nil {
return
}
r.Start()
return
}
// Log takes a logtype and message and tries to send this information to the
2018-04-16 08:12:16 +03:00
// logger provided in the revid config - if there is one, otherwise the message
// is sent to stdout
2018-04-16 08:12:16 +03:00
func (r *revid) Log(logType, m string) {
2018-04-22 07:34:48 +03:00
if r.config.Verbosity == Yes {
if r.config.Logger != nil {
r.config.Logger.Log(logType, m)
return
2018-03-17 16:29:42 +03:00
}
2018-04-23 05:29:51 +03:00
fmt.Println(logType + ": " + m)
}
2018-01-24 07:12:22 +03:00
}
2018-04-16 08:12:16 +03:00
// IsRunning returns true if the revid is currently running and false otherwise
func (r *revid) IsRunning() bool {
return r.isRunning
}
2018-04-16 08:12:16 +03:00
// Start invokes a revid to start processing video from a defined input
// and packetising (if theres packetization) to a defined output.
2018-04-16 08:12:16 +03:00
func (r *revid) Start() {
2018-03-17 16:29:42 +03:00
r.mutex.Lock()
defer r.mutex.Unlock()
if r.isRunning {
2018-04-16 08:12:16 +03:00
r.Log(Warning, "revid.Start() called but revid already running!")
return
}
r.Log(Info, "Starting Revid!")
2018-03-14 12:47:00 +03:00
if r.setupOutput != nil {
err := r.setupOutput()
if err != nil {
2018-05-03 07:04:20 +03:00
r.Log(Error, err.Error())
2018-03-14 12:47:00 +03:00
return
}
}
2018-02-14 10:02:57 +03:00
go r.setupInput()
go r.outputClips()
r.isRunning = true
}
// Stop halts any processing of video data from a camera or file
2018-04-16 08:12:16 +03:00
func (r *revid) Stop() {
2018-03-17 16:29:42 +03:00
r.mutex.Lock()
defer r.mutex.Unlock()
if !r.isRunning {
2018-04-16 08:12:16 +03:00
r.Log(Warning, "revid.Stop() called but revid not running!")
2018-03-17 16:29:42 +03:00
return
}
2018-04-16 07:54:21 +03:00
r.Log(Info, "Stopping revid!")
r.isRunning = false
// If a cmd process is running, we kill!
if r.cmd != nil && r.cmd.Process != nil {
r.cmd.Process.Kill()
}
2018-05-02 03:28:00 +03:00
r.rtmpInst.EndSession()
}
// getFrameNoPacketization gets a frame directly from the revid output chan
// as we don't need to go through the generator with no packetization settings
2018-04-16 08:12:16 +03:00
func (r *revid) getFrameNoPacketization() []byte {
return <-r.outputChan
}
// getFramePacketization gets a frame from the generators output chan - the
// the generator being an mpegts or flv generator depending on the config
2018-04-16 08:12:16 +03:00
func (r *revid) getFramePacketization() []byte {
return <-(r.generator.GetOutputChan())
}
// flushDataPacketization removes data from the revid inst's coutput chan
2018-04-16 08:12:16 +03:00
func (r *revid) flushData() {
2018-04-25 10:25:42 +03:00
for len(r.outputChan) > 0 {
<-(r.outputChan)
}
}
// packClips takes data segments; whether that be tsPackets or mjpeg frames and
// packs them into clips consisting of the amount frames specified in the config
2018-04-16 08:12:16 +03:00
func (r *revid) packClips() {
clipSize := 0
packetCount := 0
for {
// Get some memory from the ring buffer for out clip
2018-04-19 07:12:36 +03:00
var clip []byte
2018-04-19 11:07:57 +03:00
var err error
if clip, err = r.ringBuffer.Get(); err != nil {
r.Log(Error, err.Error())
r.Log(Warning, "Clearing output chan!")
2018-04-24 07:17:05 +03:00
// Keep clearing output chan until out buffer has some space
2018-04-26 10:00:18 +03:00
//r.rtmpInst.EndSession()
2018-04-24 08:22:18 +03:00
for clip, err = r.ringBuffer.Get(); err != nil; {
2018-04-24 07:17:05 +03:00
r.flushData()
2018-04-25 07:50:17 +03:00
clip, err = r.ringBuffer.Get()
2018-04-24 07:17:05 +03:00
}
2018-04-25 08:12:16 +03:00
r.Log(Debug, "Finally got mem from ringbuffer!")
}
for {
frame := r.getFrame()
lenOfFrame := len(frame)
2018-04-25 05:58:30 +03:00
if lenOfFrame > ringBufferElementSize {
r.Log(Warning,fmt.Sprintf("Frame was too big: %v bytes, getting another one!", lenOfFrame))
2018-04-25 05:58:30 +03:00
frame = r.getFrame()
lenOfFrame = len(frame)
}
upperBound := clipSize + lenOfFrame
copy(clip[clipSize:upperBound], frame)
packetCount++
clipSize += lenOfFrame
2018-04-25 05:44:17 +03:00
fpcAsInt,err := strconv.Atoi(r.config.FramesPerClip)
if err != nil {
2018-04-25 05:39:53 +03:00
r.Log(Error, "Frames per clip not quite right! Defaulting to 1!")
2018-04-25 05:41:41 +03:00
r.config.FramesPerClip = "1"
2018-04-25 05:39:53 +03:00
}
if packetCount >= fpcAsInt {
if err := r.ringBuffer.DoneWriting(clipSize); err != nil {
r.Log(Error, err.Error())
r.Log(Warning, "Dropping clip!")
}
clipSize = 0
packetCount = 0
break
}
}
}
}
2018-04-22 07:34:48 +03:00
func (r* revid)reboot() {
cmd := exec.Command("sudo", "reboot")
err := cmd.Run()
2018-04-22 07:34:48 +03:00
r.Log(Error, err.Error())
r.Log(Info, "Rebooting!")
}
// outputClips takes the clips produced in the packClips method and outputs them
2018-04-16 08:12:16 +03:00
// to the desired output defined in the revid config
func (r *revid) outputClips() {
now := time.Now()
prevTime := now
bytes := 0
2018-03-14 01:13:36 +03:00
delay := 0
for r.isRunning {
// Here we slow things down as much as we can to decrease cpu usage
switch {
2018-03-14 00:51:33 +03:00
case r.ringBuffer.GetNoOfElements() < 2:
2018-03-13 12:29:15 +03:00
delay++
time.Sleep(time.Duration(delay) * time.Millisecond)
2018-03-13 13:49:24 +03:00
case delay > 0:
delay--
}
2018-03-13 13:57:34 +03:00
// If the ringbuffer has something we can read and send off
2018-04-19 13:03:37 +03:00
if clip, err := r.ringBuffer.Read(); err == nil {
bytes += len(clip)
2018-04-16 07:54:21 +03:00
errorCount := 0
2018-04-25 08:12:16 +03:00
r.Log(Debug,"About to send!")
2018-04-25 07:01:14 +03:00
err2 := r.sendClip(clip)
2018-04-25 08:12:16 +03:00
r.Log(Debug,"Finished send!")
2018-05-03 07:58:14 +03:00
2018-04-25 07:01:14 +03:00
for ; err2 != nil; errorCount++ {
2018-04-24 08:38:30 +03:00
r.Log(Warning, "Send failed trying again!")
// If the clip size is not bigger than the threshold then we classify
// it as junk and we don't try to send it off again.
if len(clip) >= clipSizeThreshold {
2018-04-25 07:01:14 +03:00
err2 = r.sendClip(clip)
if err2 == nil {
break
}
} else {
break
}
2018-04-24 08:38:30 +03:00
// So that we don't fill up the log, once we reach the maxSendFailedErrorCount
// we will add some delay to slow things down until we have a connection again
if errorCount > maxSendFailedErrorCount {
time.Sleep(time.Duration(sendFailedDelay) * time.Second)
}
2018-04-25 07:01:14 +03:00
r.Log(Error, err2.Error())
2018-04-26 09:47:09 +03:00
if r.config.Output == NativeRtmp && errorCount > 10 {
r.rtmpInst.EndSession()
2018-05-03 07:58:14 +03:00
r.rtmpInst = rtmp.NewRTMPSession(r.config.RtmpUrl, rtmpConnectionTimout)
2018-04-26 09:47:09 +03:00
err = r.rtmpInst.StartSession()
2018-04-14 12:38:52 +03:00
}
}
// let the ringbuffer know that we're done with the memory we grabbed when
// we call ringBuffer.Get()
if err := r.ringBuffer.DoneReading(); err != nil {
r.Log(Error, err.Error())
}
// Log some information regarding bitrate and ring buffer size if it's time
now = time.Now()
deltaTime := now.Sub(prevTime)
if deltaTime > time.Duration(bitrateTime)*time.Second {
2018-04-24 08:47:45 +03:00
r.currentBitrate = int64(float64(bytes*8)/float64(deltaTime/1e9))
r.Log(Info, fmt.Sprintf("Bitrate: %v bits/s\n", r.currentBitrate ))
r.Log(Info, fmt.Sprintf("Ring buffer size: %v\n", r.ringBuffer.GetNoOfElements()))
prevTime = now
bytes = 0
}
2018-04-25 08:12:16 +03:00
} else {
2018-04-25 11:58:39 +03:00
//r.Log(Error, err.Error())
}
}
2018-03-14 12:24:11 +03:00
r.Log(Info, "Not outputting clips anymore!")
}
// senClipToFile writes the passed clip to a file
2018-04-16 08:12:16 +03:00
func (r *revid) sendClipToFile(clip []byte) error {
2018-03-14 04:18:03 +03:00
_, err := r.outputFile.Write(clip)
if err != nil {
return err
}
return nil
}
// sendClipToHTTP takes a clip and an output url and posts through http.
2018-04-16 08:12:16 +03:00
func (r *revid) sendClipToHTTP(clip []byte) error {
timeout := time.Duration(httpTimeOut * time.Second)
client := http.Client{Timeout: timeout}
2018-03-14 04:18:03 +03:00
url := r.config.HttpAddress + strconv.Itoa(len(clip))
r.Log(Debug, fmt.Sprintf("Posting %s (%d bytes)\n", url, len(clip)))
resp, err := client.Post(url, "video/mp2t", bytes.NewReader(clip))
if err != nil {
return fmt.Errorf("Error posting to %s: %s", url, err)
}
defer resp.Body.Close()
body, err := ioutil.ReadAll(resp.Body)
if err == nil {
r.Log(Debug, fmt.Sprintf("%s\n", body))
} else {
r.Log(Error, err.Error())
}
return nil
}
// sendClipToFfmpegRtmp sends the clip over the current rtmp connection using
// an ffmpeg process.
2018-04-16 08:12:16 +03:00
func (r *revid) sendClipToFfmpegRtmp(clip []byte) (err error) {
_, err = r.ffmpegStdin.Write(clip)
return
}
// sendClipToLibRtmp send the clip over the current rtmp connection using the
// c based librtmp library
2018-04-16 08:12:16 +03:00
func (r *revid) sendClipToLibRtmp(clip []byte) (err error) {
2018-03-14 04:18:03 +03:00
err = r.rtmpInst.WriteFrame(clip, uint(len(clip)))
2018-02-14 10:02:57 +03:00
return
2018-02-13 12:46:21 +03:00
}
// setupOutputForFfmpegRtmp sets up output to rtmp using an ffmpeg process
2018-04-16 08:12:16 +03:00
func (r *revid) setupOutputForFfmpegRtmp() error {
r.ffmpegCmd = exec.Command(ffmpegPath,
"-f", "h264",
"-r", r.config.FrameRate,
"-i", "-",
"-f", "lavfi",
"-i", "aevalsrc=0",
"-fflags", "nobuffer",
"-vcodec", "copy",
"-acodec", "aac",
"-map", "0:0",
"-map", "1:0",
"-strict", "experimental",
"-f", "flv",
r.config.RtmpUrl,
)
var err error
r.ffmpegStdin, err = r.ffmpegCmd.StdinPipe()
if err != nil {
r.Log(Error, err.Error())
r.Stop()
return err
}
err = r.ffmpegCmd.Start()
if err != nil {
r.Log(Error, err.Error())
r.Stop()
return err
}
return nil
}
// setupOutputForLibRtmp sets up rtmp output using the wrapper for the c based
// librtmp library - makes connection and starts comms etc.
2018-04-30 23:10:41 +03:00
func (r *revid) setupOutputForLibRtmp() error {
2018-05-02 03:38:51 +03:00
r.rtmpInst = rtmp.NewRTMPSession(r.config.RtmpUrl, rtmpConnectionTimout)
err := r.rtmpInst.StartSession()
for noOfTries := 0; err != nil && noOfTries < rtmpConnectionMaxTries; noOfTries++ {
2018-05-02 03:28:00 +03:00
r.rtmpInst.EndSession()
2018-04-30 23:10:41 +03:00
r.Log(Error, err.Error())
r.Log(Info, "Trying to establish rtmp connection again!")
r.rtmpInst = rtmp.NewRTMPSession(r.config.RtmpUrl, rtmpConnectionTimout)
err = r.rtmpInst.StartSession()
}
2018-05-02 06:54:01 +03:00
if err != nil {
2018-05-03 07:04:20 +03:00
return err
2018-05-02 06:54:01 +03:00
}
2018-05-03 07:04:20 +03:00
return err
2018-02-13 12:46:21 +03:00
}
// setupOutputForFile sets up an output file to output data to
2018-04-16 08:12:16 +03:00
func (r *revid) setupOutputForFile() (err error) {
r.outputFile, err = os.Create(r.config.OutputFileName)
return
}
// setupInputForRaspivid sets up things for input from raspivid i.e. starts
// a raspivid process and pipes it's data output.
2018-04-16 08:12:16 +03:00
func (r *revid) setupInputForRaspivid() error {
r.Log(Info, "Starting raspivid!")
switch r.config.InputCodec {
case H264:
switch r.config.QuantizationMode {
case QuantizationOn:
2018-05-03 11:27:36 +03:00
if config.Flip == config.No {
r.cmd = exec.Command("raspivid",
"-cd", "H264",
"-o", "-",
"-n",
"-t", r.config.Timeout,
"-b", "0",
"-qp", r.config.Quantization,
"-w", r.config.Width,
"-h", r.config.Height,
"-fps", r.config.FrameRate,
"-ih",
"-g", r.config.IntraRefreshPeriod,
// "-ro", r.config.Rotate,
)
} else {
r.cmd = exec.Command("raspivid",
"-cd", "H264",
"-o", "-",
"-n",
"-t", r.config.Timeout,
"-b", "0",
"-qp", r.config.Quantization,
"-w", r.config.Width,
"-h", r.config.Height,
"-fps", r.config.FrameRate,
"-ih",
"-g", r.config.IntraRefreshPeriod,
"-hf",
)
}
case QuantizationOff:
2018-05-03 11:27:36 +03:00
if config.Flip == config.No {
r.cmd = exec.Command("raspivid",
"-cd", "H264",
"-o", "-",
"-n",
"-t", r.config.Timeout,
"-b", r.config.Bitrate,
"-w", r.config.Width,
"-h", r.config.Height,
"-fps", r.config.FrameRate,
"-ih",
"-g", r.config.IntraRefreshPeriod,
)
} else {
r.cmd = exec.Command("raspivid",
"-cd", "H264",
"-o", "-",
"-n",
"-t", r.config.Timeout,
"-b", r.config.Bitrate,
"-w", r.config.Width,
"-h", r.config.Height,
"-fps", r.config.FrameRate,
"-ih",
"-g", r.config.IntraRefreshPeriod,
"-hf",
)
}
}
case Mjpeg:
r.cmd = exec.Command("raspivid",
"-cd", "MJPEG",
"-o", "-",
"-n",
"-t", r.config.Timeout,
"-fps", r.config.FrameRate,
)
}
2018-03-14 13:07:07 +03:00
stdout, _ := r.cmd.StdoutPipe()
err := r.cmd.Start()
r.inputReader = bufio.NewReader(stdout)
if err != nil {
r.Log(Error, err.Error())
return err
}
go r.readCamera()
return nil
}
// setupInputForFile sets things up for getting input from a file
2018-04-16 08:12:16 +03:00
func (r *revid) setupInputForFile() error {
2018-03-14 04:18:03 +03:00
fps, _ := strconv.Atoi(r.config.FrameRate)
r.parser.SetDelay(uint(float64(1000) / float64(fps)))
2018-02-14 10:02:57 +03:00
r.readFile()
return nil
}
// testRtmp is useful to check robustness of connections. Intended to be run as
// goroutine. After every 'delayTime' the rtmp connection is ended and then
// restarted
2018-04-16 08:12:16 +03:00
func (r *revid) testRtmp(delayTime uint) {
for {
2018-03-14 04:18:03 +03:00
time.Sleep(time.Duration(delayTime) * time.Millisecond)
r.rtmpInst.EndSession()
r.rtmpInst.StartSession()
}
}
2018-04-16 08:12:16 +03:00
// readCamera reads data from the defined camera while the revid is running.
// TODO: use ringbuffer here instead of allocating mem every time!
2018-04-16 08:12:16 +03:00
func (r *revid) readCamera() {
r.Log(Info, "Reading camera data!")
for r.isRunning {
data := make([]byte, 1)
_, err := io.ReadFull(r.inputReader, data)
switch {
// We know this means we're getting nothing from the cam
2018-04-24 07:17:05 +03:00
case ( err != nil && err.Error() == "EOF" && r.isRunning ) || ( err != nil && r.isRunning):
r.Log(Error, "No data from camera!")
2018-04-16 08:29:49 +03:00
time.Sleep(cameraRetryPeriod)
default:
2018-01-31 04:00:03 +03:00
r.parser.GetInputChan() <- data[0]
2018-01-11 09:49:33 +03:00
}
}
r.Log(Info, "Not trying to read from camera anymore!")
}
2018-04-16 08:12:16 +03:00
// readFile reads data from the defined file while the revid is running.
func (r *revid) readFile() error {
var err error
r.inputFile, err = os.Open(r.config.InputFileName)
if err != nil {
r.Log(Error, err.Error())
r.Stop()
return err
}
stats, err := r.inputFile.Stat()
if err != nil {
r.Log(Error, "Could not get input file stats!")
r.Stop()
return err
}
data := make([]byte, stats.Size())
_, err = r.inputFile.Read(data)
if err != nil {
r.Log(Error, err.Error())
r.Stop()
return err
}
for i := range data {
r.parser.GetInputChan() <- data[i]
}
r.inputFile.Close()
return nil
}