merge master into turbidity-probe

cmd/rv/probe.go

@@ -34,11 +34,12 @@ import (
 	"os"
 	"time"
 
+	"gocv.io/x/gocv"
+	"gonum.org/v1/gonum/stat"
+
 	"bitbucket.org/ausocean/av/codec/h264"
 	"bitbucket.org/ausocean/av/turbidity"
 	"bitbucket.org/ausocean/utils/logger"
-	"gocv.io/x/gocv"
-	"gonum.org/v1/gonum/stat"
 )
 
 // Misc constants.

cmd/rv/probe_circleci.go

@@ -39,7 +39,7 @@ type turbidityProbe struct {
 	sharpness, contrast float64
 }
 
-// NewTurbidityProbe returns a empty turbidity probe for CircleCI testing only.
+// NewTurbidityProbe returns an empty turbidity probe for CircleCI testing only.
 func NewTurbidityProbe(log logger.Logger, delay time.Duration) (*turbidityProbe, error) {
 	tp := new(turbidityProbe)
 	return tp, nil
@@ -47,7 +47,7 @@ func NewTurbidityProbe(log logger.Logger, delay time.Duration) (*turbidityProbe,
 
 // Write performs no operation for CircleCI testing only.
 func (tp *turbidityProbe) Write(p []byte) (int, error) {
-	return len(p), nil
+	return 0, nil
 }
 
 func (tp *turbidityProbe) Close() error {

codec/codecutil/lex.go

@@ -69,8 +69,11 @@ func (l *ByteLexer) Lex(dst io.Writer, src io.Reader, d time.Duration) error {
 	for {
 		<-ticker.C
 		off, err := src.Read(buf)
-		if err != nil {
+		// The only error that will stop the lexer is an EOF.
+		if err == io.EOF {
 			return err
+		} else if err != nil {
+			continue
 		}
 		_, err = dst.Write(buf[:off])
 		if err != nil {

device/alsa/alsa.go

@@ -30,6 +30,7 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"io"
 	"sync"
 	"time"
 
@@ -376,6 +377,10 @@ func (d *ALSA) input() {
 				d.dev.Close()
 				d.dev = nil
 			}
+			err := d.buf.Close()
+			if err != nil {
+				d.l.Log(logger.Error, "unable to close pool buffer", "error", err)
+			}
 			return
 		}
 
@@ -405,7 +410,6 @@ func (d *ALSA) input() {
 			d.l.Log(logger.Warning, "old audio data overwritten")
 		default:
 			d.l.Log(logger.Error, "unexpected ringbuffer error", "error", err.Error())
-			return
 		}
 	}
 }
@@ -413,13 +417,37 @@ func (d *ALSA) input() {
 // Read reads from the ringbuffer, returning the number of bytes read upon success.
 func (d *ALSA) Read(p []byte) (int, error) {
 	// Ready ringbuffer for read.
+	d.l.Log(logger.Debug, pkg+"getting next chunk ready")
 	_, err := d.buf.Next(rbNextTimeout)
 	if err != nil {
-		return 0, err
+		switch err {
+		case io.EOF:
+			d.l.Log(logger.Debug, pkg+"EOF from Next")
+			return 0, err
+		case pool.ErrTimeout:
+			d.l.Log(logger.Debug, pkg+"pool buffer timeout")
+			return 0, err
+		default:
+			d.l.Log(logger.Error, pkg+"unexpected error from Next", "error", err.Error())
+			return 0, err
+		}
 	}
 
 	// Read from pool buffer.
-	return d.buf.Read(p)
+	d.l.Log(logger.Debug, pkg+"reading from buffer")
+	n, err := d.buf.Read(p)
+	if err != nil {
+		switch err {
+		case io.EOF:
+			d.l.Log(logger.Debug, pkg+"EOF from Read")
+			return n, err
+		default:
+			d.l.Log(logger.Error, pkg+"unexpected error from Read", "error", err.Error())
+			return n, err
+		}
+	}
+	d.l.Log(logger.Debug, fmt.Sprintf("%v read %v bytes", pkg, n))
+	return n, nil
 }
 
 // formatBuffer returns audio that has been converted to the desired format.