audio: removed bufSize arg in Lex funcs and use ByteLexer

codec/codecutil/lex.go

@@ -30,21 +30,25 @@ import (
 	"time"
 )
 
-// ByteLexer is used to lex a certain number of bytes per a given delay, the number is configured upon construction.
+// ByteLexer is used to lex bytes using a buffer size which is configured upon construction.
 type ByteLexer struct {
-	bufSize int
+	bufSize *int
 }
 
 // NewByteLexer returns a pointer to a ByteLexer with the given buffer size.
-func NewByteLexer(bufSize int) (*ByteLexer, error) {
-	if bufSize <= 0 {
-		return nil, fmt.Errorf("invalid buffer size: %v", bufSize)
-	}
-	return &ByteLexer{bufSize: bufSize}, nil
+func NewByteLexer(bufSize *int) *ByteLexer {
+	return &ByteLexer{bufSize: bufSize}
 }
 
-// Lex reads l.bufSize bytes from src and writes them to dst every t seconds.
+// Lex reads *l.bufSize bytes from src and writes them to dst every t seconds.
 func (l *ByteLexer) Lex(dst io.Writer, src io.Reader, t time.Duration) error {
+	if l.bufSize == nil {
+		return fmt.Errorf("buffer size has not been set")
+	}
+	bufSize := *l.bufSize
+	if bufSize <= 0 {
+		return fmt.Errorf("invalid buffer size: %v", bufSize)
+	}
 	if t < 0 {
 		return fmt.Errorf("invalid delay: %v", t)
 	}
@@ -55,7 +59,7 @@ func (l *ByteLexer) Lex(dst io.Writer, src io.Reader, t time.Duration) error {
 		tick = ticker.C
 	}
 
-	buf := make([]byte, l.bufSize)
+	buf := make([]byte, bufSize)
 	for {
 		if t != 0 {
 			<-tick

codec/codecutil/lex_test.go

@@ -36,7 +36,7 @@ var lexTests = []struct {
 	data []byte
 	t    time.Duration
 	n    int
-	fail bool
+	fail bool // Whether or not this test should fail.
 }{
 	{[]byte{0x10, 0x00, 0xf3, 0x45, 0xfe, 0xd2, 0xaa, 0x4e}, time.Millisecond, 4, false},
 	{[]byte{0x10, 0x00, 0xf3, 0x45, 0xfe, 0xd2, 0xaa, 0x4e}, time.Millisecond, 3, false},
@@ -51,16 +51,12 @@ func TestByteLexer(t *testing.T) {
 	for i, tt := range lexTests {
 		t.Run(strconv.Itoa(i), func(t *testing.T) {
 			dst := bytes.NewBuffer([]byte{})
-			l, err := NewByteLexer(tt.n)
-			if err != nil {
+			l := NewByteLexer(&tt.n)
+			err := l.Lex(dst, bytes.NewReader(tt.data), tt.t)
+			if err != nil && err != io.EOF {
 				if !tt.fail {
 					t.Errorf("unexpected error: %v", err.Error())
 				}
-				return
-			}
-			err = l.Lex(dst, bytes.NewReader(tt.data), tt.t)
-			if err != nil && err != io.EOF {
-				t.Errorf("unexpected error: %v", err.Error())
 			} else if !bytes.Equal(dst.Bytes(), tt.data) {
 				t.Errorf("data before and after lex are not equal: want %v, got %v", tt.data, dst.Bytes())
 			}

codec/h264/lex.go

@@ -49,7 +49,7 @@ var h264Prefix = [...]byte{0x00, 0x00, 0x01, 0x09, 0xf0}
 // to dst with successive writes being performed not earlier than the specified
 // delay. NAL units are split after type 1 (Coded slice of a non-IDR picture), 5
 // (Coded slice of a IDR picture) and 8 (Picture parameter set).
-func Lex(dst io.Writer, src io.Reader, delay time.Duration, n int) error {
+func Lex(dst io.Writer, src io.Reader, delay time.Duration) error {
 	var tick <-chan time.Time
 	if delay == 0 {
 		tick = noDelay

codec/h265/lex.go

@@ -70,7 +70,7 @@ func NewLexer(donl bool) *Lexer {
 // Lex continually reads RTP packets from the io.Reader src and lexes into
 // access units which are written to the io.Writer dst. Lex expects that for
 // each read from src, a single RTP packet is received.
-func (l *Lexer) Lex(dst io.Writer, src io.Reader, delay time.Duration, n int) error {
+func (l *Lexer) Lex(dst io.Writer, src io.Reader, delay time.Duration) error {
 	buf := make([]byte, maxRTPSize)
 	for {
 		n, err := src.Read(buf)

codec/h265/lex_test.go

@@ -246,7 +246,7 @@ func TestLex(t *testing.T) {
 	for testNum, test := range tests {
 		r := &rtpReader{packets: test.packets}
 		d := &destination{}
-		err := NewLexer(test.donl).Lex(d, r, 0, 0)
+		err := NewLexer(test.donl).Lex(d, r, 0)
 		if err != nil {
 			t.Fatalf("error lexing: %v\n", err)
 		}

codec/mjpeg/lex.go

@@ -45,7 +45,7 @@ func init() {
 
 // Lex parses MJPEG frames read from src into separate writes to dst with
 // successive writes being performed not earlier than the specified delay.
-func Lex(dst io.Writer, src io.Reader, delay time.Duration, n int) error {
+func Lex(dst io.Writer, src io.Reader, delay time.Duration) error {
 	var tick <-chan time.Time
 	if delay == 0 {
 		tick = noDelay

input/audio/audio_test.go

@@ -61,7 +61,9 @@ func TestDevice(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
-	go codecutil.LexBytes(ioutil.Discard, ai, time.Duration(ac.RecPeriod*float64(time.Second)), ai.ChunkSize())
+	chunkSize := ai.ChunkSize()
+	lexer := codecutil.NewByteLexer(&chunkSize)
+	go lexer.Lex(ioutil.Discard, ai, time.Duration(ac.RecPeriod*float64(time.Second)))
 	time.Sleep(time.Duration(ac.RecPeriod*float64(time.Second)) * time.Duration(n))
 	ai.Stop()
 }

revid/config.go

@@ -230,6 +230,7 @@ type Config struct {
 	RecPeriod  float64 // How many seconds to record at a time.
 	Channels   int     // Number of audio channels, 1 for mono, 2 for stereo.
 	BitDepth   int     // Sample bit depth.
+	ChunkSize  int     // ChunkSize is the size of the chunks in the audio.Device's ringbuffer.
 
 	RTPAddress     string // RTPAddress defines the RTP output destination.
 	BurstPeriod    uint   // BurstPeriod defines the revid burst period in seconds.

revid/revid.go

@@ -106,7 +106,7 @@ type Revid struct {
 	cmd *exec.Cmd
 
 	// lexTo, encoder and packer handle transcoding the input stream.
-	lexTo func(dest io.Writer, src io.Reader, delay time.Duration, bufSize int) error
+	lexTo func(dest io.Writer, src io.Reader, delay time.Duration) error
 
 	// encoders will hold the multiWriteCloser that writes to encoders from the lexer.
 	encoders io.WriteCloser
@@ -294,7 +294,7 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io.
 		r.lexTo = h265.NewLexer(false).Lex
 	case Audio:
 		r.setupInput = r.startAudioDevice
-		r.lexTo = codecutil.LexBytes
+		r.lexTo = codecutil.NewByteLexer(&r.config.ChunkSize).Lex
 	}
 
 	return nil
@@ -564,7 +564,7 @@ func (r *Revid) startRaspivid() (func() error, error) {
 	}
 
 	r.wg.Add(1)
-	go r.processFrom(stdout, 0, 0)
+	go r.processFrom(stdout, 0)
 	return nil, nil
 }
 
@@ -606,7 +606,7 @@ func (r *Revid) startV4L() (func() error, error) {
 	}
 
 	r.wg.Add(1)
-	go r.processFrom(stdout, time.Duration(0), 0)
+	go r.processFrom(stdout, time.Duration(0))
 	return nil, nil
 }
 
@@ -621,7 +621,7 @@ func (r *Revid) setupInputForFile() (func() error, error) {
 
 	// TODO(kortschak): Maybe we want a context.Context-aware parser that we can stop.
 	r.wg.Add(1)
-	go r.processFrom(f, 0, 0)
+	go r.processFrom(f, 0)
 	return func() error { return f.Close() }, nil
 }
 
@@ -660,8 +660,9 @@ func (r *Revid) startAudioDevice() (func() error, error) {
 	}
 
 	// Process output from audio device.
+	r.config.ChunkSize = ai.ChunkSize()
 	r.wg.Add(1)
-	go r.processFrom(ai, time.Duration(float64(time.Second)/r.config.WriteRate), ai.ChunkSize())
+	go r.processFrom(ai, time.Duration(float64(time.Second)/r.config.WriteRate))
 	return func() error {
 		ai.Stop()
 		return nil
@@ -732,7 +733,7 @@ func (r *Revid) startRTSPCamera() (func() error, error) {
 
 	// Start reading data from the RTP client.
 	r.wg.Add(1)
-	go r.processFrom(rtpClt, time.Second/time.Duration(r.config.FrameRate), 0)
+	go r.processFrom(rtpClt, time.Second/time.Duration(r.config.FrameRate))
 
 	return func() error {
 		rtspClt.Close()
@@ -770,9 +771,9 @@ func parseSvrRTCPPort(resp rtsp.Response) (int, error) {
 	return 0, errors.New("SETUP response did not provide RTCP port")
 }
 
-func (r *Revid) processFrom(read io.Reader, delay time.Duration, bufSize int) {
+func (r *Revid) processFrom(read io.Reader, delay time.Duration) {
 	r.config.Logger.Log(logger.Info, pkg+"reading input data")
-	r.err <- r.lexTo(r.encoders, read, delay, bufSize)
+	r.err <- r.lexTo(r.encoders, read, delay)
 	r.config.Logger.Log(logger.Info, pkg+"finished reading input data")
 	r.wg.Done()
 }