Merged in blitzwechseln (pull request #34)

all: make naming more idiomatic
This commit is contained in:
kortschak 2018-06-17 12:30:10 +00:00
commit 62cbef4ed6
10 changed files with 87 additions and 82 deletions

View File

@ -241,7 +241,7 @@ func main() {
var vs int
if *useNetsender {
// initialize NetSender and use NetSender's logger
config.Logger = netsender.GetLogger()
config.Logger = netsender.Logger()
var err error
err = ns.Init(nil, nil, nil)
@ -267,7 +267,7 @@ func main() {
continue
}
if vs != ns.GetVarSum() {
if vs != ns.VarSum() {
// vars changed
vars, err := ns.Vars()
if err != nil {
@ -275,7 +275,7 @@ func main() {
time.Sleep(netSendRetryTime)
continue
}
vs = ns.GetVarSum()
vs = ns.VarSum()
if vars["mode"] == "Paused" {
if !paused {
config.Logger.Log(progName, "Info", "Pausing revid")
@ -306,7 +306,7 @@ func sendTo(ns *netsender.Sender) error {
inputs := netsender.MakePins(ns.GetConfigParam("ip"), "X")
for i, pin := range inputs {
if pin.Name == "X23" {
inputs[i].Value = int(revidInst.GetBitrate())
inputs[i].Value = revidInst.Bitrate()
}
}

View File

@ -62,15 +62,15 @@ type flvGenerator struct {
isGenerating bool
}
// GetInputChan returns the input channel to the generator. This is where the
// InputChan returns the input channel to the generator. This is where the
// raw data frames are entered into the generator
func (g *flvGenerator) GetInputChan() chan []byte {
func (g *flvGenerator) InputChan() chan []byte {
return g.inputChan
}
// GetOutputChan retuns the output chan of the generator - this is where the
// OutputChan retuns the output chan of the generator - this is where the
// flv packets (more specifically tags) are outputted.
func (g *flvGenerator) GetOutputChan() chan []byte {
func (g *flvGenerator) OutputChan() <-chan []byte {
return g.outputChan
}

View File

@ -28,8 +28,8 @@ LICENSE
package generator
type Generator interface {
GetInputChan() chan []byte
GetOutputChan() chan []byte
InputChan() chan []byte
OutputChan() <-chan []byte
Start()
Stop()
}

View File

@ -86,15 +86,15 @@ type tsGenerator struct {
isGenerating bool
}
// getInputChan returns a handle to the nalInputChan (inputChan) so that nal units
// InputChan returns a handle to the nalInputChan (inputChan) so that nal units
// can be passed to the generator and processed
func (g *tsGenerator) GetInputChan() chan []byte {
func (g *tsGenerator) InputChan() chan []byte {
return g.nalInputChan
}
// GetOutputChan returns a handle to the generator output chan where the mpegts
// OutputChan returns a handle to the generator output chan where the mpegts
// packets will show up once ready to go
func (g *tsGenerator) GetOutputChan() chan []byte {
func (g *tsGenerator) OutputChan() <-chan []byte {
return g.outputChan
}

View File

@ -37,9 +37,9 @@ const (
outputBufferSize = 10000
)
// h264Parser provides properties and methods to allow for the parsing of a
// H264 provides properties and methods to allow for the parsing of a
// h264 stream - i.e. to allow extraction of the individual access units
type h264Parser struct {
type H264 struct {
inputBuffer []byte
isParsing bool
parserOutputChanRef chan []byte
@ -48,9 +48,9 @@ type h264Parser struct {
delay uint
}
// NewH264Parser returns an instance of the h264Parser struct
func NewH264Parser() (p *h264Parser) {
p = new(h264Parser)
// NewH264Parser returns an instance of the H264 struct
func NewH264Parser() (p *H264) {
p = new(H264)
p.isParsing = true
p.inputChan = make(chan byte, inputChanSize)
p.delay = 0
@ -60,12 +60,12 @@ func NewH264Parser() (p *h264Parser) {
// Stop simply sets the isParsing flag to false to indicate to the parser that
// we don't want to interpret incoming data anymore - this will also make the
// parser jump out of the parse func
func (p *h264Parser) Stop() {
func (p *H264) Stop() {
p.isParsing = false
}
// Start starts the parse func as a goroutine so that incoming data is interpreted
func (p *h264Parser) Start() {
func (p *H264) Start() {
p.isParsing = true
go p.parse()
}
@ -73,31 +73,31 @@ func (p *h264Parser) Start() {
// SetDelay sets a delay inbetween each buffer output. Useful if we're parsing
// a file but want to replicate the speed of incoming video frames from a
// camera
func (p *h264Parser) SetDelay(delay uint) {
func (p *H264) SetDelay(delay uint) {
p.delay = delay
}
// GetInputChan returns a handle to the input channel of the parser
func (p *h264Parser) GetInputChan() chan byte {
// InputChan returns a handle to the input channel of the parser
func (p *H264) InputChan() chan byte {
return p.inputChan
}
// GetOutputChan returns a handle to the output chan of the parser
func (p *h264Parser) GetOutputChan() chan []byte {
// OutputChan returns a handle to the output chan of the parser
func (p *H264) OutputChan() <-chan []byte {
return p.userOutputChanRef
}
// SetOutputChan sets the parser output chan to the passed output chan. This is
// useful if we want the parser output to go directly to a generator of some sort
// for packetization.
func (p *h264Parser) SetOutputChan(aChan chan []byte) {
p.parserOutputChanRef = aChan
p.userOutputChanRef = aChan
func (p *H264) SetOutputChan(o chan []byte) {
p.parserOutputChanRef = o
p.userOutputChanRef = o
}
// parse interprets an incoming h264 stream and extracts individual frames
// aka access units
func (p *h264Parser) parse() {
func (p *H264) parse() {
outputBuffer := make([]byte, 0, outputBufferSize)
searchingForEnd := false
for p.isParsing {

View File

@ -29,7 +29,7 @@ package parser
const frameStartCode = 0xD8
type mjpegParser struct {
type MJPEG struct {
inputBuffer []byte
isParsing bool
parserOutputChanRef chan []byte
@ -38,39 +38,39 @@ type mjpegParser struct {
delay uint
}
func NewMJPEGParser(inputChanLen int) (p *mjpegParser) {
p = new(mjpegParser)
func NewMJPEGParser(inputChanLen int) (p *MJPEG) {
p = new(MJPEG)
p.isParsing = true
p.inputChan = make(chan byte, inputChanLen)
return
}
func (p *mjpegParser) Stop() {
func (p *MJPEG) Stop() {
p.isParsing = false
}
func (p *mjpegParser) Start() {
func (p *MJPEG) Start() {
go p.parse()
}
func (p *mjpegParser) SetDelay(delay uint) {
func (p *MJPEG) SetDelay(delay uint) {
p.delay = delay
}
func (p *mjpegParser) GetInputChan() chan byte {
func (p *MJPEG) InputChan() chan byte {
return p.inputChan
}
func (p *mjpegParser) GetOutputChan() chan []byte {
func (p *MJPEG) OutputChan() <-chan []byte {
return p.userOutputChanRef
}
func (p *mjpegParser) SetOutputChan(aChan chan []byte) {
p.parserOutputChanRef = aChan
p.userOutputChanRef = aChan
func (p *MJPEG) SetOutputChan(o chan []byte) {
p.parserOutputChanRef = o
p.userOutputChanRef = o
}
func (p *mjpegParser) parse() {
func (p *MJPEG) parse() {
var outputBuffer []byte
for p.isParsing {
aByte := <-p.inputChan

View File

@ -43,8 +43,8 @@ var (
type Parser interface {
Stop()
Start()
GetInputChan() chan byte
GetOutputChan() chan []byte
InputChan() chan byte
OutputChan() <-chan []byte
SetOutputChan(achan chan []byte)
SetDelay(delay uint)
}

View File

@ -83,23 +83,23 @@ const (
// Revid provides methods to control a revid session; providing methods
// to start, stop and change the state of an instance using the Config struct.
type Revid struct {
ffmpegPath string
tempDir string
ringBuffer *ring.Buffer
config Config
isRunning bool
inputFile *os.File
generator generator.Generator
parser parser.Parser
cmd *exec.Cmd
inputReader *bufio.Reader
ffmpegStdin io.WriteCloser
outputChan chan []byte
setupInput func() error
getFrame func() []byte
destination loadSender
rtmpInst rtmp.Session
currentBitrate int64
ffmpegPath string
tempDir string
ringBuffer *ring.Buffer
config Config
isRunning bool
inputFile *os.File
generator generator.Generator
parser parser.Parser
cmd *exec.Cmd
inputReader *bufio.Reader
ffmpegStdin io.WriteCloser
outputChan chan []byte
setupInput func() error
getFrame func() []byte
destination loadSender
rtmpInst rtmp.Session
bitrate int
}
// NewRevid returns a pointer to a new Revid with the desired
@ -116,14 +116,18 @@ func New(c Config) (*Revid, error) {
return &r, nil
}
// 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
// Bitrate returns the result of the most recent bitrate check.
func (r *Revid) Bitrate() int {
return r.bitrate
}
// GetConfigRef returns a pointer to the revidInst's Config struct object
func (r *Revid) GetConfigRef() *Config {
// Config returns the Revid's config.
func (r *Revid) Config() *Config {
// FIXME(kortschak): This is a massive footgun and should not exist.
// Since the config's fields are accessed in running goroutines, any
// mutation is a data race. With bad luck a data race is possible by
// reading the returned value since it is possible for the running
// Ravid to mutate the config it holds.
return &r.config
}
@ -199,13 +203,13 @@ func (r *Revid) reset(config Config) error {
// 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.parser.SetOutputChan(r.generator.InputChan())
return nil
}
// ChangeConfig changes the current configuration of the Revid instance.
func (r *Revid) ChangeConfig(c Config) error {
// SetConfig changes the current configuration of the receiver.
func (r *Revid) SetConfig(c Config) error {
// FIXME(kortschak): This is reimplemented in cmd/revid-cli/main.go.
// The implementation in the command is used and this is not.
// Decide on one or the other.
@ -233,7 +237,7 @@ func (r *Revid) Log(logType, m string) {
fmt.Println(logType + ": " + m)
}
// IsRunning returns true if the revid is currently running and false otherwise
// IsRunning returns whether the receiver is running.
func (r *Revid) IsRunning() bool {
return r.isRunning
}
@ -296,7 +300,7 @@ func (r *Revid) getFrameNoPacketization() []byte {
// getFramePacketization gets a frame from the generators output chan - the
// the generator being an mpegts or flv generator depending on the config
func (r *Revid) getFramePacketization() []byte {
return <-r.generator.GetOutputChan()
return <-r.generator.OutputChan()
}
// packClips takes data segments; whether that be tsPackets or mjpeg frames and
@ -308,7 +312,7 @@ func (r *Revid) packClips() {
select {
// TODO: This is temporary, need to work out how to make this work
// for cases when there is not packetisation.
case frame := <-r.generator.GetOutputChan():
case frame := <-r.generator.OutputChan():
lenOfFrame := len(frame)
if lenOfFrame > ringBufferElementSize {
r.Log(Warning, fmt.Sprintf("Frame was too big: %v bytes, getting another one!", lenOfFrame))
@ -417,8 +421,9 @@ func (r *Revid) outputClips() {
now = time.Now()
deltaTime := now.Sub(prevTime)
if deltaTime > bitrateTime {
r.currentBitrate = int64(float64(bytes*8) / float64(deltaTime/1e9))
r.Log(Debug, fmt.Sprintf("Bitrate: %v bits/s\n", r.currentBitrate))
// FIXME(kortschak): For subsecond deltaTime, this will give infinite bitrate.
r.bitrate = int(float64(bytes*8) / float64(deltaTime/time.Second))
r.Log(Debug, fmt.Sprintf("Bitrate: %v bits/s\n", r.bitrate))
r.Log(Debug, fmt.Sprintf("Ring buffer size: %v\n", r.ringBuffer.Len()))
prevTime = now
bytes = 0
@ -513,7 +518,7 @@ func (r *Revid) readCamera() {
r.Log(Error, "No data from camera!")
time.Sleep(cameraRetryPeriod)
default:
r.parser.GetInputChan() <- data[0]
r.parser.InputChan() <- data[0]
}
}
r.Log(Info, "Not trying to read from camera anymore!")
@ -542,7 +547,7 @@ func (r *Revid) readFile() error {
return err
}
for i := range data {
r.parser.GetInputChan() <- data[i]
r.parser.InputChan() <- data[i]
}
r.inputFile.Close()
return nil

View File

@ -223,7 +223,7 @@ func newRtmpSender(url string, timeout uint, retries int, log func(lvl, msg stri
var err error
for n := 0; n < retries; n++ {
sess = rtmp.NewSession(url, timeout)
err = sess.StartSession()
err = sess.Open()
if err == nil {
break
}
@ -269,7 +269,7 @@ func (s *rtmpSender) restart() error {
}
for n := 0; n < s.retries; n++ {
s.sess = rtmp.NewSession(s.url, s.timeout)
err = s.sess.StartSession()
err = s.sess.Open()
if err == nil {
break
}

View File

@ -48,7 +48,7 @@ import (
// Session provides an interface for sending flv tags over rtmp.
type Session interface {
StartSession() error
Open() error
Write([]byte) (int, error)
Close() error
}
@ -71,9 +71,9 @@ func NewSession(url string, connectTimeout uint) Session {
}
}
// StartSession establishes an rtmp connection with the url passed into the
// Open establishes an rtmp connection with the url passed into the
// constructor
func (s *session) StartSession() error {
func (s *session) Open() error {
if s.rtmp != nil {
return errors.New("rtmp: attempt to start already running session")
}