Merge branch 'master' of https://bitbucket.org/ausocean/av

cmd/revid-cli/main.go

@@ -45,14 +45,6 @@ import (
 	"bitbucket.org/ausocean/utils/logger"
 )
 
-const (
-	// progName is the program name for logging purposes.
-	progName = "revid-cli"
-
-	// Logging is set to INFO level.
-	defaultLogVerbosity = logger.Info
-)
-
 // Revid modes
 const (
 	normal = "Normal"
@@ -67,12 +59,14 @@ const (
 	revidStopTime       = 5 * time.Second
 	defaultLogPath      = "/var/log/netsender"
 	pkg                 = "revid-cli:"
+	defaultLogVerbosity = logger.Info
+	defaultSleepTime    = 60 // Seconds
 )
 
 // canProfile is set to false with revid-cli is built with "-tags profile".
 var canProfile = true
 
-// The logger that will be used throughout
+// The logger that will be used throughout.
 var log *logger.Logger
 
 const (
@@ -100,9 +94,7 @@ func main() {
 		return
 	}
 
-	if err := run(cfg); err != nil {
+	run(cfg)
-		log.Log(logger.Fatal, pkg+"failed to run revid", "error", err.Error())
-	}
 }
 
 // handleFlags parses command line flags and returns a revid configuration
@@ -267,16 +259,17 @@ func handleFlags() revid.Config {
 }
 
 // initialize then run the main NetSender client
-func run(cfg revid.Config) error {
+func run(cfg revid.Config) {
 	log.Log(logger.Info, pkg+"running in NetSender mode")
 
-	var vars map[string]string
-
 	var rv *revid.Revid
 
 	readPin := func(pin *netsender.Pin) error {
 		switch {
 		case pin.Name == "X23":
+			if rv == nil {
+				pin.Value = -1
+			}
 			pin.Value = rv.Bitrate()
 		case pin.Name[0] == 'X':
 			return sds.ReadSystem(pin)
@@ -288,35 +281,10 @@ func run(cfg revid.Config) error {
 
 	ns, err := netsender.New(log, nil, readPin, nil)
 	if err != nil {
-		return err
+		log.Log(logger.Fatal, pkg+"could not initialise netsender client")
-	}
-
-	rv, err = revid.New(cfg, ns)
-	if err != nil {
-		log.Log(logger.Fatal, pkg+"could not initialise revid", "error", err.Error())
-	}
-
-	vars, _ = ns.Vars()
-	vs := ns.VarSum()
-
-	// Update revid to get latest config settings from netreceiver.
-	err = rv.Update(vars)
-	if err != nil {
-		return err
-	}
-
-	// If mode on netreceiver isn't paused then we can start revid.
-	if ns.Mode() != paused && ns.Mode() != burst {
-		err = rv.Start()
-		if err != nil {
-			return err
-		}
-	}
-
-	if ns.Mode() == burst {
-		ns.SetMode(paused, &vs)
 	}
 
+	var vs int
 	for {
 		err = ns.Run()
 		if err != nil {
@@ -325,10 +293,13 @@ func run(cfg revid.Config) error {
 			continue
 		}
 
-		// If var sum hasn't change we continue
+		// If var sum hasn't changed we continue.
-		if vs == ns.VarSum() {
+		var vars map[string]string
+		newVs := ns.VarSum()
+		if vs == newVs {
 			goto sleep
 		}
+		vs = newVs
 
 		vars, err = ns.Vars()
 		if err != nil {
@@ -336,35 +307,52 @@ func run(cfg revid.Config) error {
 			time.Sleep(netSendRetryTime)
 			continue
 		}
-		vs = ns.VarSum()
+
+		if rv == nil {
+			rv, err = revid.New(cfg, ns)
+			if err != nil {
+				log.Log(logger.Warning, pkg+"could not initialise revid", "error", err.Error())
+				goto sleep
+			}
+		}
 
 		err = rv.Update(vars)
 		if err != nil {
-			return err
+			log.Log(logger.Warning, pkg+"Couldn't update revid", "error", err.Error())
+			goto sleep
 		}
 
 		switch ns.Mode() {
 		case paused:
+			rv.Stop()
 		case normal:
 			err = rv.Start()
 			if err != nil {
-				return err
+				log.Log(logger.Warning, pkg+"could not start revid", "error", err.Error())
+				ns.SetMode(paused, &vs)
+				goto sleep
 			}
 		case burst:
 			log.Log(logger.Info, pkg+"Starting burst...")
+
 			err = rv.Start()
 			if err != nil {
-				return err
+				log.Log(logger.Warning, pkg+"could not start burst", "error", err.Error())
+				ns.SetMode(paused, &vs)
+				goto sleep
 			}
+
 			time.Sleep(time.Duration(rv.Config().BurstPeriod) * time.Second)
 			log.Log(logger.Info, pkg+"Stopping burst...")
 			rv.Stop()
 			ns.SetMode(paused, &vs)
 		}
+
 	sleep:
 		sleepTime, err := strconv.Atoi(ns.Param("mp"))
 		if err != nil {
-			return err
+			log.Log(logger.Error, pkg+"could not get sleep time, using default")
+			sleepTime = defaultSleepTime
 		}
 		time.Sleep(time.Duration(sleepTime) * time.Second)
 	}

init/Makefile

@@ -1,16 +1,18 @@
 # install files and directories required by NetSender clients (such as gpio-netsender, revid-cli, etc.)
 # NB: the default (soft) install does not override conf files
 USER := $(shell whoami)
+PATH := /usr/local/go/bin:$(PATH)
 
 .SILENT:make_dirs
 .SILENT:soft_copy_files
 .SILENT:hard_copy_files
+.SILENT:syncreboot
 .SILENT:clean
 
-install: as_root make_dirs soft_copy_files
+install: as_root make_dirs soft_copy_files syncreboot
 	@echo "Install complete" 
 
-install_hard: as_root make_dirs hard_copy_files
+install_hard: as_root make_dirs hard_copy_files syncreboot
 	@echo "Hard install complete" 
 
 as_root:
@@ -38,6 +40,7 @@ soft_copy_files:
 		echo "/etc/netsender.conf left unmodified" ; \
 	else \
 		cp netsender.conf /etc; \
+		chown pi /etc/netsender.conf; \
 	fi
 
 hard_copy_files:
@@ -51,6 +54,10 @@ hard_copy_files:
 		cp /etc/netsender.conf /etc/netsender.conf.bak ; \
 	fi
 	cp -f netsender.conf /etc
+	chown pi /etc/netsender.conf
+
+syncreboot:
+	cd ../../utils/cmd/syncreboot; make; make install
 
 clean: as_root
 	rm -rf /var/netsender

revid/config.go

@@ -126,7 +126,7 @@ func (c *Config) Validate(r *Revid) error {
 	case No:
 	case NothingDefined:
 		c.LogLevel = defaultVerbosity
-		c.Logger.Log(logger.Warning, pkg+"no LogLevel mode defined, defaulting",
+		c.Logger.Log(logger.Info, pkg+"no LogLevel mode defined, defaulting",
 			"LogLevel", defaultVerbosity)
 	default:
 		return errors.New("bad LogLevel defined in config")
@@ -135,7 +135,7 @@ func (c *Config) Validate(r *Revid) error {
 	switch c.Input {
 	case Raspivid, V4L, File:
 	case NothingDefined:
-		c.Logger.Log(logger.Warning, pkg+"no input type defined, defaulting", "input",
+		c.Logger.Log(logger.Info, pkg+"no input type defined, defaulting", "input",
 			defaultInput)
 		c.Input = defaultInput
 	default:
@@ -160,10 +160,10 @@ func (c *Config) Validate(r *Revid) error {
 		}
 
 	case NothingDefined:
-		c.Logger.Log(logger.Warning, pkg+"no input codec defined, defaulting",
+		c.Logger.Log(logger.Info, pkg+"no input codec defined, defaulting",
 			"inputCodec", defaultInputCodec)
 		c.InputCodec = defaultInputCodec
-		c.Logger.Log(logger.Warning, pkg+"defaulting quantization", "quantization",
+		c.Logger.Log(logger.Info, pkg+"defaulting quantization", "quantization",
 			defaultQuantization)
 		c.Quantization = defaultQuantization
 
@@ -187,8 +187,9 @@ func (c *Config) Validate(r *Revid) error {
 				"framesPerClip", defaultFramesPerClip)
 			c.FramesPerClip = defaultFramesPerClip
 			c.Packetization = Flv
+			c.SendRetry = true
 		case NothingDefined:
-			c.Logger.Log(logger.Warning, pkg+"no output defined, defaulting", "output",
+			c.Logger.Log(logger.Info, pkg+"no output defined, defaulting", "output",
 				defaultOutput)
 			c.Outputs[i] = defaultOutput
 			c.Packetization = defaultPacketization
@@ -204,43 +205,43 @@ func (c *Config) Validate(r *Revid) error {
 	}
 
 	if c.BurstPeriod == 0 {
-		c.Logger.Log(logger.Warning, pkg+"no burst period defined, defaulting", "burstPeriod", defaultBurstPeriod)
+		c.Logger.Log(logger.Info, pkg+"no burst period defined, defaulting", "burstPeriod", defaultBurstPeriod)
 		c.BurstPeriod = defaultBurstPeriod
 	}
 
 	if c.FramesPerClip < 1 {
-		c.Logger.Log(logger.Warning, pkg+"no FramesPerClip defined, defaulting",
+		c.Logger.Log(logger.Info, pkg+"no FramesPerClip defined, defaulting",
 			"framesPerClip", defaultFramesPerClip)
 		c.FramesPerClip = defaultFramesPerClip
 	}
 
 	if c.Width == 0 {
-		c.Logger.Log(logger.Warning, pkg+"no width defined, defaulting", "width", defaultWidth)
+		c.Logger.Log(logger.Info, pkg+"no width defined, defaulting", "width", defaultWidth)
 		c.Width = defaultWidth
 	}
 
 	if c.Height == 0 {
-		c.Logger.Log(logger.Warning, pkg+"no height defined, defaulting", "height", defaultHeight)
+		c.Logger.Log(logger.Info, pkg+"no height defined, defaulting", "height", defaultHeight)
 		c.Height = defaultHeight
 	}
 
 	if c.FrameRate == 0 {
-		c.Logger.Log(logger.Warning, pkg+"no frame rate defined, defaulting", "fps", defaultFrameRate)
+		c.Logger.Log(logger.Info, pkg+"no frame rate defined, defaulting", "fps", defaultFrameRate)
 		c.FrameRate = defaultFrameRate
 	}
 
 	if c.Bitrate == 0 {
-		c.Logger.Log(logger.Warning, pkg+"no bitrate defined, defaulting", "bitrate", defaultBitrate)
+		c.Logger.Log(logger.Info, pkg+"no bitrate defined, defaulting", "bitrate", defaultBitrate)
 		c.Bitrate = defaultBitrate
 	}
 
 	if c.IntraRefreshPeriod == 0 {
-		c.Logger.Log(logger.Warning, pkg+"no intra refresh defined, defaulting", "intraRefresh", defaultIntraRefreshPeriod)
+		c.Logger.Log(logger.Info, pkg+"no intra refresh defined, defaulting", "intraRefresh", defaultIntraRefreshPeriod)
 		c.IntraRefreshPeriod = defaultIntraRefreshPeriod
 	}
 
 	if c.Quantization == 0 {
-		c.Logger.Log(logger.Warning, pkg+"no quantization defined, defaulting", "quantization", defaultQuantization)
+		c.Logger.Log(logger.Info, pkg+"no quantization defined, defaulting", "quantization", defaultQuantization)
 		c.Quantization = defaultQuantization
 	} else if c.Quantization > 51 {
 		return errors.New("quantisation is over threshold")

revid/revid.go

@@ -54,7 +54,7 @@ const (
 	mtsRbSize        = 100
 	mtsRbElementSize = 150000
 	flvRbSize        = 1000
-	flvRbElementSize = 10000
+	flvRbElementSize = 100000
 	writeTimeout     = 10 * time.Millisecond
 	readTimeout      = 10 * time.Millisecond
 )
@@ -355,6 +355,9 @@ func (r *Revid) Start() error {
 	go r.outputClips()
 	r.config.Logger.Log(logger.Info, pkg+"setting up input and receiving content")
 	err := r.setupInput()
+	if err != nil {
+		r.Stop()
+	}
 	return err
 }
 
@@ -537,7 +540,7 @@ loop:
 			err = dest.send()
 			if err == nil {
 				r.config.Logger.Log(logger.Debug, pkg+"sent clip to output "+strconv.Itoa(i))
-			} else if r.config.SendRetry == false {
+			} else if !r.config.SendRetry {
 				r.config.Logger.Log(logger.Warning, pkg+"send to output "+strconv.Itoa(i)+" failed", "error", err.Error())
 			} else {
 				r.config.Logger.Log(logger.Error, pkg+"send to output "+strconv.Itoa(i)+
@@ -546,19 +549,16 @@ loop:
 				if err != nil && chunk.Len() > 11 {
 					r.config.Logger.Log(logger.Error, pkg+"second send attempted failed, restarting connection", "error", err.Error())
 					for err != nil {
-						time.Sleep(sendFailedDelay)
 						if rs, ok := dest.(restarter); ok {
 							r.config.Logger.Log(logger.Debug, pkg+"restarting session", "session", rs)
 							err = rs.restart()
 							if err != nil {
 								r.config.Logger.Log(logger.Error, pkg+"failed to restart rtmp session", "error", err.Error())
-								r.setIsRunning(false)
+								time.Sleep(sendFailedDelay)
-								return
+								continue
 							}
-
+							r.config.Logger.Log(logger.Info, pkg+"restarted rtmp session, sending again")
-							r.config.Logger.Log(logger.Info, pkg+"restarted rtmp session")
 						}
-
 						err = dest.send()
 						if err != nil {
 							r.config.Logger.Log(logger.Error, pkg+"send failed again, with error", "error", err.Error())

revid/senders.go

@@ -298,6 +298,9 @@ func (s *rtmpSender) load(c *ring.Chunk) error {
 
 func (s *rtmpSender) send() error {
 	_, err := s.chunk.WriteTo(s.conn)
+	if err == rtmp.ErrInvalidFlvTag {
+		return nil
+	}
 	return err
 }
 
@@ -307,17 +310,14 @@ func (s *rtmpSender) release() {
 }
 
 func (s *rtmpSender) restart() error {
-	err := s.conn.Close()
+	s.close()
-	if err != nil {
+	var err error
-		return err
-	}
 	for n := 0; n < s.retries; n++ {
 		s.conn, err = rtmp.Dial(s.url, s.timeout, s.log)
 		if err == nil {
 			break
 		}
 		s.log(logger.Error, err.Error())
-		s.conn.Close()
 		if n < s.retries-1 {
 			s.log(logger.Info, pkg+"retry rtmp connection")
 		}
@@ -326,7 +326,10 @@ func (s *rtmpSender) restart() error {
 }
 
 func (s *rtmpSender) close() error {
+	if s.conn != nil {
 		return s.conn.Close()
+	}
+	return nil
 }
 
 // udpSender implements loadSender for a native udp destination.

rtmp/conn.go

@@ -155,7 +155,7 @@ func (c *Conn) Write(data []byte) (int, error) {
 		return 0, errNotConnected
 	}
 	if len(data) < flvTagheaderSize {
-		return 0, errInvalidFlvTag
+		return 0, ErrInvalidFlvTag
 	}
 	if data[0] == packetTypeInfo || (data[0] == 'F' && data[1] == 'L' && data[2] == 'V') {
 		return 0, errUnimplemented

rtmp/rtmp.go

@@ -158,7 +158,7 @@ var (
 	errNotWritable   = errors.New("rtmp: connection not writable")
 	errInvalidHeader = errors.New("rtmp: invalid header")
 	errInvalidBody   = errors.New("rtmp: invalid body")
-	errInvalidFlvTag = errors.New("rtmp: invalid FLV tag")
+	ErrInvalidFlvTag = errors.New("rtmp: invalid FLV tag")
 	errUnimplemented = errors.New("rtmp: unimplemented feature")
 )
 

rtmp/rtmp_test.go

@@ -216,7 +216,7 @@ type rtmpSender struct {
 
 func (rs *rtmpSender) Write(p []byte) (int, error) {
 	n, err := rs.conn.Write(p)
-	if err != errInvalidFlvTag && err != nil {
+	if err != ErrInvalidFlvTag && err != nil {
 		return 0, err
 	}
 	return n, nil