Merge branch 'master' into treat-make

cmd/treatment/main.go

@@ -92,6 +92,10 @@ var varMap = map[string]string{
 	"AudioFilePath": "string",
 }
 
+const audioCmd = "aplay"
+
+func initCommand(l logging.Logger) { checkPath(audioCmd, l) }
+
 func main() {
 	mts.Meta = meta.NewWith([][2]string{{metaPreambleKey, metaPreambleData}})
 
@@ -274,7 +278,6 @@ func setChannels(mode string, l logging.Logger) error {
 		return fmt.Errorf("channel set command failed: %s", &errBuff)
 	}
 
-
 	l.Info("mode set to", "mode", mode)
 	return nil
 }

cmd/treatment/pi.go

@@ -1,36 +0,0 @@
-/*
-DESCRIPTION
-  pi.go defines an initialisation function for use when running on
-  a Raspberry Pi.
-
-AUTHORS
-  Ella Pietraroia <ella@ausocean.org>
-  Scott Barnard <scott@ausocean.org>
-  Saxon Nelson-Milton <saxon@ausocean.org>
-
-LICENSE
-  Copyright (C) 2020 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 main
-
-import (
-	"bitbucket.org/ausocean/utils/logging"
-)
-
-const audioCmd = "aplay"
-
-func initCommand(l logging.Logger) { checkPath(audioCmd, l) }

cmd/vidforward/main.go

@@ -38,7 +38,6 @@ import (
 	"os"
 	"reflect"
 	"strconv"
-	"strings"
 	"sync"
 	"time"
 	"runtime/debug"
@@ -449,55 +448,35 @@ func (m *broadcastManager) createOrUpdate(broadcast Broadcast) error {
 
 	// Try to get any old broadcasts for the provided MAC.
 	maybeOld, ok := m.broadcasts[broadcast.mac]
-
+	if ok {
-	// If there's no old broadcast, we need to create a new revid pipeline.
+		if maybeOld.rv != nil {
-	if !ok {
+			m.log.Debug("stopping old revid pipeline", "mac", broadcast.mac)
-		maybeOld = &broadcast
+			closeDone := make(chan struct{})
+			go func() { maybeOld.rv.Stop(); close(closeDone) }()
+			select {
+			case <-closeDone:
+				m.log.Debug("stopped old revid pipeline", "mac", broadcast.mac)
+			case <-time.After(5 * time.Second):
+				m.log.Warning("could not stop old revid pipeline, looks like we'll end up with some leaked memory then :(", "mac", broadcast.mac)
+			}
+		}
 	}
 
-	m.log.Debug("attempting to start init or start pipeline")
 	var err error
-	maybeOld.rv, err = m.initOrStartPipeline(maybeOld.rv, broadcast.urls)
+	maybeOld.rv, err = newRevid(m.log, broadcast.urls)
 	if err != nil {
-		return fmt.Errorf("could not get revid pipeline: %w", err)
+		return fmt.Errorf("could not create new revid: %w", err)
 	}
-	m.log.Debug("finished attempting to init or start pipeline")
+	maybeOld.rv.Start()
 
-	// If the URLS have changed, we need to update the revid pipeline.
-	// We won't enter this if we've just created a new revid pipeline.
-	if !reflect.DeepEqual(maybeOld.urls, broadcast.urls) {
-		m.log.Debug("RTMP URLs have changed, updating pipeline config", "mac", broadcast.mac, "old", maybeOld.urls, "new", broadcast.urls)
-		urls := strings.Join(broadcast.urls, ",")
-		err := maybeOld.rv.Update(map[string]string{"RTMPURL": urls})
-		if err != nil {
-			return fmt.Errorf("could not update revid: %w", err)
-		}
-		m.log.Debug("finished updating pipeline config")
-	}
-
-	// If the status has changed, we need to update accordingly.
-	// i.e. if the status is slate, we need to start writing the slate image.
-	// We won't enter this if we've just created a new revid pipeline.
-	if maybeOld.status != broadcast.status {
-		m.log.Debug("status has changed, starting or stopping slate", "mac", broadcast.mac, "old", maybeOld.status, "new", broadcast.status)
-		switch broadcast.status {
-		case statusActive, statusPlay, statusCreate:
 	m.log.Info("updating configuration for mac", "mac", broadcast.mac)
 	signal, ok := m.slateExitSignals[broadcast.mac]
 	if ok {
 		close(signal)
 		delete(m.slateExitSignals, broadcast.mac)
 	}
-		case statusSlate:
-			m.log.Info("slate request")
-			// If there's a signal channel it means that we're already writing the slate
-			// image and theres nothing to do, so return.
-			_, ok := m.slateExitSignals[broadcast.mac]
-			if ok {
-				m.log.Warning("already writing slate")
-				return nil
-			}
 
+	if broadcast.status == statusSlate {
 		// First create a signal that can be used to stop the slate writing routine.
 		// This will be provided to the writeSlate routine below.
 		signalCh := make(chan struct{})
@@ -507,9 +486,6 @@ func (m *broadcastManager) createOrUpdate(broadcast Broadcast) error {
 		if err != nil {
 			return fmt.Errorf("could not write slate and check for errors: %w", err)
 		}
-		default:
-			return fmt.Errorf("unknown status string: %s", broadcast.status)
-		}
 	}
 
 	// We need to give the revid pipeline to the new broadcast record.