diff --git a/codec/codecutil/lex.go b/codec/codecutil/lex.go index 2b0f1b05..e727e07d 100644 --- a/codec/codecutil/lex.go +++ b/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 diff --git a/codec/codecutil/lex_test.go b/codec/codecutil/lex_test.go index 9264f3d1..63162abd 100644 --- a/codec/codecutil/lex_test.go +++ b/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()) } diff --git a/codec/h264/lex.go b/codec/h264/lex.go index b210ee25..176c8b3b 100644 --- a/codec/h264/lex.go +++ b/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 diff --git a/codec/h265/lex.go b/codec/h265/lex.go index 7593fe5e..ebe34013 100644 --- a/codec/h265/lex.go +++ b/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) diff --git a/codec/h265/lex_test.go b/codec/h265/lex_test.go index 02ed5f1f..1a409e4c 100644 --- a/codec/h265/lex_test.go +++ b/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) } diff --git a/codec/mjpeg/lex.go b/codec/mjpeg/lex.go index 21717fe6..da2ecae1 100644 --- a/codec/mjpeg/lex.go +++ b/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 diff --git a/input/audio/audio_test.go b/input/audio/audio_test.go index f26bb4c3..5618c63a 100644 --- a/input/audio/audio_test.go +++ b/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() } diff --git a/revid/config.go b/revid/config.go index e34a1f1a..cf108db6 100644 --- a/revid/config.go +++ b/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. diff --git a/revid/revid.go b/revid/revid.go index 43075c08..f64f13fb 100644 --- a/revid/revid.go +++ b/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() }