revid: added start and stop functions to AudioInput for proper closing

revid/audio-input.go

@@ -49,18 +49,17 @@ const (
 var log *logger.Logger
 
 // AudioInput holds everything we need to know about the audio input stream.
-// NB: At 44100 Hz frame rate, 2 channels and 16-bit samples, a period of 5 seconds
+// Note: At 44100 Hz sample rate, 2 channels and 16-bit samples, a period of 5 seconds
-// results in PCM data chunks of 882000 bytes! A longer period exceeds datastore's 1MB blob limit.
+// results in PCM data chunks of 882000 bytes. A longer period exceeds datastore's 1MB blob limit.
 type AudioInput struct {
-	mu     sync.Mutex // mu protects the AudioInput.
+	mu     sync.Mutex
-	mode   string     // operating mode, either "Normal" or "Paused"
+	source string // Name of audio source, or empty for the default source.
-	source string     // name of audio source, or empty for the default source
+	mode   string // Operating mode, either "Running", "Paused", or "Stopped".
 
-	dev       *alsa.Device // audio input device
+	dev       *alsa.Device // Audio input device.
-	ab        alsa.Buffer  // ALSA's buffer
+	ab        alsa.Buffer  // ALSA's buffer.
-	rb        *ring.Buffer // our buffer
+	rb        *ring.Buffer // Our buffer.
-	chunkSize int
+	chunkSize int          // This is the number of bytes that will be stored at a time.
-	vs        int // our "var sum" to track var changes
 
 	*AudioConfig
 }
@@ -74,17 +73,15 @@ type AudioConfig struct {
 	Codec      uint8
 }
 
-// NewAudioInput starts recording audio and returns an AudioInput struct which the audio can be read from.
+// NewAudioInput initializes and returns an AudioInput struct which can be started, read from, and stopped.
 func NewAudioInput(cfg *AudioConfig) *AudioInput {
-
+	// Initialize logger.
 	logLevel := int(logger.Debug)
-
 	validLogLevel := true
 	if logLevel < int(logger.Debug) || logLevel > int(logger.Fatal) {
 		logLevel = int(logger.Info)
 		validLogLevel = false
 	}
-
 	logSender := smartlogger.New(logPath)
 	log = logger.New(int8(logLevel), &logSender.LogRoller)
 	log.Log(logger.Info, "log-netsender: Logger Initialized")
@@ -101,17 +98,42 @@ func NewAudioInput(cfg *AudioConfig) *AudioInput {
 		log.Log(logger.Fatal, "alsa.open failed", "error", err.Error())
 	}
 
-	// Capture audio in periods of a.RecPeriod seconds, and buffer rbDuration seconds in total.
+	// Setup ring buffer to capture audio in periods of a.RecPeriod seconds, and buffer rbDuration seconds in total.
 	a.ab = a.dev.NewBufferDuration(time.Second * time.Duration(a.RecPeriod))
 	a.chunkSize = (((len(a.ab.Data) / a.dev.BufferFormat().Channels) * a.Channels) / a.dev.BufferFormat().Rate) * a.SampleRate
 	rbLen := rbDuration / a.RecPeriod
 	a.rb = ring.NewBuffer(rbLen, a.chunkSize, rbTimeout)
 
-	go a.input()
+	a.mode = "Paused"
 
 	return a
 }
 
+// Start will start recording audio and writing to the output.
+func (a *AudioInput) Start() {
+	a.mu.Lock()
+	mode := a.mode
+	a.mu.Unlock()
+	switch mode {
+	case "Paused":
+		go a.input()
+	case "Stopped":
+
+	}
+}
+
+// Stop will stop recording audio and close
+func (a *AudioInput) Stop() {
+	a.mode = "Stopped"
+	if a.dev != nil {
+		log.Log(logger.Debug, "Closing", "source", a.source)
+		a.dev.Close()
+		a.dev = nil
+	}
+
+}
+
+// ChunkSize returns the AudioInput's chunkSize, ie. the number of bytes of audio written to output at a time.
 func (a *AudioInput) ChunkSize() int {
 	return a.chunkSize
 }
@@ -204,7 +226,7 @@ func (a *AudioInput) open() error {
 	case 32:
 		aFmt = alsa.S32_LE
 	default:
-		return fmt.Errorf("Unsupported sample bits %v\n", a.BitDepth)
+		return fmt.Errorf("unsupported sample bits %v", a.BitDepth)
 	}
 	_, err = a.dev.NegotiateFormat(aFmt)
 	if err != nil {
@@ -226,17 +248,17 @@ func (a *AudioInput) open() error {
 
 // input continously records audio and writes it to the ringbuffer.
 // Re-opens the device and tries again if ASLA returns an error.
-// Spends a lot of time sleeping in Paused mode.
-// ToDo: Currently, reading audio and writing to the ringbuffer are synchronous.
-// Need a way to asynchronously read from the ALSA buffer, i.e.,  _while_ it is recording to avoid any gaps.
 func (a *AudioInput) input() {
 	for {
 		a.mu.Lock()
 		mode := a.mode
 		a.mu.Unlock()
-		if mode == "Paused" {
+		switch mode {
+		case "Paused":
 			time.Sleep(time.Duration(a.RecPeriod) * time.Second)
 			continue
+		case "Stopped":
+			break
 		}
 		log.Log(logger.Debug, "Recording audio for period", "seconds", a.RecPeriod)
 		a.mu.Lock()

revid/audio-input_test.go

@@ -3,10 +3,13 @@ package revid
 import (
 	"testing"
 
+	"bitbucket.org/ausocean/av/container/mts"
+	"bitbucket.org/ausocean/av/container/mts/meta"
 	"bitbucket.org/ausocean/iot/pi/netsender"
 )
 
-func TestAudioInputStart(t *testing.T) {
+func TestAudioInputNew(t *testing.T) {
+	mts.Meta = meta.New()
 	var logger testLogger
 	ns, err := netsender.New(&logger, nil, nil, nil)
 	if err != nil {
@@ -20,10 +23,9 @@ func TestAudioInputStart(t *testing.T) {
 	rv, err := New(c, ns)
 	if err != nil {
 		t.Errorf("revid.New failed with error %v", err)
+	} else if rv == nil {
+		t.Errorf("revid.New did not return a new revid")
 	}
 
-	err = rv.Start()
+	rv.Stop()
-	if err != nil {
-		t.Errorf("revid.Start failed with error %v", err)
-	}
 }

revid/revid.go

@@ -627,8 +627,12 @@ func (r *Revid) startAudioInput() (func() error, error) {
 		Codec:      r.config.InputCodec,
 	}
 	ai := NewAudioInput(ac)
+	r.wg.Add(1)
 	go r.processFrom(ai, time.Second/time.Duration(r.config.WriteRate), ai.ChunkSize())
-	return nil, nil
+	return func() error {
+		ai.Stop()
+		return nil
+	}, nil
 }
 
 func (r *Revid) processFrom(read io.Reader, delay time.Duration, bufSize int) {