Merged in deshim (pull request #47)

revid,stream: improve package organisation

Approved-by: saxon.milton@gmail.com <saxon.milton@gmail.com>
This commit is contained in:
kortschak 2018-08-24 21:17:17 +00:00
commit ea33007173
3 changed files with 10 additions and 614 deletions

View File

@ -1,246 +0,0 @@
/*
NAME
parse.go
DESCRIPTION
See Readme.md
AUTHOR
Dan Kortschak <dan@ausocean.org>
LICENSE
parse.go is Copyright (C) 2017 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
along with revid in gpl.txt. If not, see http://www.gnu.org/licenses.
*/
// Package parse provides parsers for video encodings.
package parse
import (
"bufio"
"bytes"
"fmt"
"io"
"time"
)
var noDelay = make(chan time.Time)
func init() {
close(noDelay)
}
var h264Prefix = [...]byte{0x00, 0x00, 0x01, 0x09, 0xf0}
// H264 lexes H.264 NAL units read from src into separate writes 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 H264(dst io.Writer, src io.Reader, delay time.Duration) error {
var tick <-chan time.Time
if delay == 0 {
tick = noDelay
} else {
ticker := time.NewTicker(delay)
defer ticker.Stop()
tick = ticker.C
}
const bufSize = 8 << 10
c := newScanner(src, make([]byte, 4<<10)) // Standard file buffer size.
buf := make([]byte, len(h264Prefix), bufSize)
copy(buf, h264Prefix[:])
writeOut := false
outer:
for {
var b byte
var err error
buf, b, err = c.scanUntilZeroInto(buf)
if err != nil {
if err != io.EOF {
return err
}
break
}
for n := 1; b == 0x0 && n < 4; n++ {
b, err = c.readByte()
if err != nil {
if err != io.EOF {
return err
}
break outer
}
buf = append(buf, b)
if b != 0x1 || (n != 2 && n != 3) {
continue
}
if writeOut {
<-tick
_, err := dst.Write(buf[:len(buf)-(n+1)])
if err != nil {
return err
}
buf = make([]byte, len(h264Prefix)+n, bufSize)
copy(buf, h264Prefix[:])
buf = append(buf, 1)
writeOut = false
}
b, err = c.readByte()
if err != nil {
if err != io.EOF {
return err
}
break outer
}
buf = append(buf, b)
// http://www.itu.int/rec/dologin_pub.asp?lang=e&id=T-REC-H.264-200305-S!!PDF-E&type=items
// Table 7-1 NAL unit type codes
const (
nonIdrPic = 1
idrPic = 5
paramSet = 8
)
switch nalTyp := b & 0x1f; nalTyp {
case nonIdrPic, idrPic, paramSet:
writeOut = true
}
}
}
if len(buf) == len(h264Prefix) {
return nil
}
<-tick
_, err := dst.Write(buf)
return err
}
// scanner is a byte scanner.
type scanner struct {
buf []byte
off int
// r is the source of data for the scanner.
r io.Reader
}
// newScanner returns a scanner initialised with an io.Reader and a read buffer.
func newScanner(r io.Reader, buf []byte) *scanner {
return &scanner{r: r, buf: buf[:0]}
}
// scanUntilZeroInto scans the scanner's underlying io.Reader until a zero byte
// has been read, appending all read bytes to dst. The resulting appended data,
// the last read byte and whether the last read byte was zero are returned.
func (c *scanner) scanUntilZeroInto(dst []byte) (res []byte, b byte, err error) {
outer:
for {
var i int
for i, b = range c.buf[c.off:] {
if b != 0x0 {
continue
}
dst = append(dst, c.buf[c.off:c.off+i+1]...)
c.off += i + 1
break outer
}
dst = append(dst, c.buf[c.off:]...)
err = c.reload()
if err != nil {
break
}
}
return dst, b, err
}
// readByte is an unexported ReadByte.
func (c *scanner) readByte() (byte, error) {
if c.off >= len(c.buf) {
err := c.reload()
if err != nil {
return 0, err
}
}
b := c.buf[c.off]
c.off++
return b, nil
}
// reload re-fills the scanner's buffer.
func (c *scanner) reload() error {
n, err := c.r.Read(c.buf[:cap(c.buf)])
c.buf = c.buf[:n]
if err != nil {
if err != io.EOF {
return err
}
if n == 0 {
return io.EOF
}
}
c.off = 0
return nil
}
// MJPEG parses MJPEG frames read from src into separate writes to dst with
// successive writes being performed not earlier than the specified delay.
func MJPEG(dst io.Writer, src io.Reader, delay time.Duration) error {
var tick <-chan time.Time
if delay == 0 {
tick = noDelay
} else {
ticker := time.NewTicker(delay)
defer ticker.Stop()
tick = ticker.C
}
r := bufio.NewReader(src)
for {
buf := make([]byte, 2, 4<<10)
n, err := r.Read(buf)
if n < 2 {
return nil
}
if err != nil {
return err
}
if !bytes.Equal(buf, []byte{0xff, 0xd8}) {
return fmt.Errorf("parser: not MJPEG frame start: %#v", buf)
}
var last byte
for {
b, err := r.ReadByte()
if err != nil {
return err
}
buf = append(buf, b)
if last == 0xff && b == 0xd9 {
break
}
last = b
}
<-tick
_, err = dst.Write(buf)
if err != nil {
return err
}
}
}

View File

@ -1,344 +0,0 @@
/*
NAME
parse_test.go
DESCRIPTION
See Readme.md
AUTHOR
Dan Kortschak <dan@ausocean.org>
LICENSE
parse_test.go is Copyright (C) 2017 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
along with revid in gpl.txt. If not, see http://www.gnu.org/licenses.
*/
package parse
import (
"bytes"
"fmt"
"reflect"
"testing"
"time"
)
var h264Tests = []struct {
name string
input []byte
delay time.Duration
want [][]byte
err error
}{
{
name: "empty",
},
{
name: "null short",
input: []byte{0x00, 0x00, 0x01, 0x0},
delay: 0,
want: [][]byte{{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x01, 0x0}},
},
{
name: "null short delayed",
input: []byte{0x00, 0x00, 0x01, 0x0},
delay: time.Millisecond,
want: [][]byte{{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x01, 0x0}},
},
{
name: "full short type 1",
input: []byte{
0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x01, 0x41, 'w', 'i', 't', 'h',
0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd',
},
delay: 0,
want: [][]byte{
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x01, 0x41, 'w', 'i', 't', 'h'},
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd'},
},
},
{
name: "full short type 5",
input: []byte{
0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x01, 0x45, 'w', 'i', 't', 'h',
0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd',
},
delay: 0,
want: [][]byte{
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x01, 0x45, 'w', 'i', 't', 'h'},
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd'},
},
},
{
name: "full short type 8",
input: []byte{
0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x01, 0x48, 'w', 'i', 't', 'h',
0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd',
},
delay: 0,
want: [][]byte{
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x01, 0x48, 'w', 'i', 't', 'h'},
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd'},
},
},
{
name: "full short delayed",
input: []byte{
0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x01, 0x41, 'w', 'i', 't', 'h',
0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd',
},
delay: time.Millisecond,
want: [][]byte{
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x01, 0x41, 'w', 'i', 't', 'h'},
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd'},
},
},
{
name: "full long type 1",
input: []byte{
0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x00, 0x01, 0x41, 'w', 'i', 't', 'h',
0x00, 0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd',
},
delay: 0,
want: [][]byte{
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x00, 0x01, 0x41, 'w', 'i', 't', 'h'},
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd'},
},
},
{
name: "full long type 5",
input: []byte{
0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x00, 0x01, 0x45, 'w', 'i', 't', 'h',
0x00, 0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd',
},
delay: 0,
want: [][]byte{
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x00, 0x01, 0x45, 'w', 'i', 't', 'h'},
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd'},
},
},
{
name: "full long type 8",
input: []byte{
0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x00, 0x01, 0x48, 'w', 'i', 't', 'h',
0x00, 0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd',
},
delay: 0,
want: [][]byte{
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x00, 0x01, 0x48, 'w', 'i', 't', 'h'},
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd'},
},
},
{
name: "full long delayed",
input: []byte{
0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x00, 0x01, 0x41, 'w', 'i', 't', 'h',
0x00, 0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd',
},
delay: time.Millisecond,
want: [][]byte{
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'u', 'l', 'l',
0x00, 0x00, 0x00, 0x01, 0x00, 'f', 'r', 'a', 'm', 'e',
0x00, 0x00, 0x00, 0x01, 0x41, 'w', 'i', 't', 'h'},
{0x0, 0x0, 0x1, 0x9, 0xf0, 0x00, 0x00, 0x00, 0x01, 0x00, 'l', 'e', 'n', 'g', 't', 'h',
0x00, 0x00, 0x00, 0x01, 0x00, 's', 'p', 'r', 'e', 'a', 'd'},
},
},
}
func TestH264(t *testing.T) {
for _, test := range h264Tests {
var buf chunkWriter
err := H264(&buf, bytes.NewReader(test.input), test.delay)
if fmt.Sprint(err) != fmt.Sprint(test.err) {
t.Errorf("unexpected error for %q: got:%v want:%v", test.name, err, test.err)
}
if err != nil {
continue
}
got := [][]byte(buf)
if !reflect.DeepEqual(got, test.want) {
t.Errorf("unexpected result for %q:\ngot :%#v\nwant:%#v", test.name, got, test.want)
}
}
}
var mjpegTests = []struct {
name string
input []byte
delay time.Duration
want [][]byte
err error
}{
{
name: "empty",
},
{
name: "null",
input: []byte{0xff, 0xd8, 0xff, 0xd9},
delay: 0,
want: [][]byte{{0xff, 0xd8, 0xff, 0xd9}},
},
{
name: "null delayed",
input: []byte{0xff, 0xd8, 0xff, 0xd9},
delay: time.Millisecond,
want: [][]byte{{0xff, 0xd8, 0xff, 0xd9}},
},
{
name: "full",
input: []byte{
0xff, 0xd8, 'f', 'u', 'l', 'l', 0xff, 0xd9,
0xff, 0xd8, 'f', 'r', 'a', 'm', 'e', 0xff, 0xd9,
0xff, 0xd8, 'w', 'i', 't', 'h', 0xff, 0xd9,
0xff, 0xd8, 'l', 'e', 'n', 'g', 't', 'h', 0xff, 0xd9,
0xff, 0xd8, 's', 'p', 'r', 'e', 'a', 'd', 0xff, 0xd9,
},
delay: 0,
want: [][]byte{
{0xff, 0xd8, 'f', 'u', 'l', 'l', 0xff, 0xd9},
{0xff, 0xd8, 'f', 'r', 'a', 'm', 'e', 0xff, 0xd9},
{0xff, 0xd8, 'w', 'i', 't', 'h', 0xff, 0xd9},
{0xff, 0xd8, 'l', 'e', 'n', 'g', 't', 'h', 0xff, 0xd9},
{0xff, 0xd8, 's', 'p', 'r', 'e', 'a', 'd', 0xff, 0xd9},
},
},
{
name: "full delayed",
input: []byte{
0xff, 0xd8, 'f', 'u', 'l', 'l', 0xff, 0xd9,
0xff, 0xd8, 'f', 'r', 'a', 'm', 'e', 0xff, 0xd9,
0xff, 0xd8, 'w', 'i', 't', 'h', 0xff, 0xd9,
0xff, 0xd8, 'l', 'e', 'n', 'g', 't', 'h', 0xff, 0xd9,
0xff, 0xd8, 's', 'p', 'r', 'e', 'a', 'd', 0xff, 0xd9,
},
delay: time.Millisecond,
want: [][]byte{
{0xff, 0xd8, 'f', 'u', 'l', 'l', 0xff, 0xd9},
{0xff, 0xd8, 'f', 'r', 'a', 'm', 'e', 0xff, 0xd9},
{0xff, 0xd8, 'w', 'i', 't', 'h', 0xff, 0xd9},
{0xff, 0xd8, 'l', 'e', 'n', 'g', 't', 'h', 0xff, 0xd9},
{0xff, 0xd8, 's', 'p', 'r', 'e', 'a', 'd', 0xff, 0xd9},
},
},
}
func TestMJEG(t *testing.T) {
for _, test := range mjpegTests {
var buf chunkWriter
err := MJPEG(&buf, bytes.NewReader(test.input), test.delay)
if fmt.Sprint(err) != fmt.Sprint(test.err) {
t.Errorf("unexpected error for %q: got:%v want:%v", test.name, err, test.err)
}
if err != nil {
continue
}
got := [][]byte(buf)
if !reflect.DeepEqual(got, test.want) {
t.Errorf("unexpected result for %q:\ngot :%#v\nwant:%#v", test.name, got, test.want)
}
}
}
type chunkWriter [][]byte
func (w *chunkWriter) Write(b []byte) (int, error) {
*w = append(*w, b)
return len(b), nil
}
func TestScannerReadByte(t *testing.T) {
data := []byte("Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.")
for _, size := range []int{1, 2, 8, 1 << 10} {
r := newScanner(bytes.NewReader(data), make([]byte, size))
var got []byte
for {
b, err := r.readByte()
if err != nil {
break
}
got = append(got, b)
}
if !bytes.Equal(got, data) {
t.Errorf("unexpected result for buffer size %d:\ngot :%q\nwant:%q", size, got, data)
}
}
}
func TestScannerScanUntilZero(t *testing.T) {
data := []byte("Lorem ipsum dolor sit amet, consectetur adipiscing elit,\x00 sed do eiusmod tempor incididunt ut \x00labore et dolore magna aliqua.")
for _, size := range []int{1, 2, 8, 1 << 10} {
r := newScanner(bytes.NewReader(data), make([]byte, size))
var got [][]byte
for {
buf, _, err := r.scanUntilZeroInto(nil)
got = append(got, buf)
if err != nil {
break
}
}
want := bytes.SplitAfter(data, []byte{0})
if !reflect.DeepEqual(got, want) {
t.Errorf("unexpected result for buffer zie %d:\ngot :%q\nwant:%q", size, got, want)
}
}
}

View File

@ -38,10 +38,10 @@ import (
"strconv"
"time"
"bitbucket.org/ausocean/av/parse"
"bitbucket.org/ausocean/av/rtmp"
"bitbucket.org/ausocean/av/stream"
"bitbucket.org/ausocean/av/stream/flv"
"bitbucket.org/ausocean/av/stream/lex"
"bitbucket.org/ausocean/av/stream/mts"
"bitbucket.org/ausocean/iot/pi/netsender"
"bitbucket.org/ausocean/utils/ring"
@ -91,7 +91,7 @@ type Revid struct {
config Config
isRunning bool
encoder stream.Encoder
parse func(dst io.Writer, src io.Reader, delay time.Duration) error
lexTo func(dst stream.Encoder, src io.Reader, delay time.Duration) error
cmd *exec.Cmd
inputReader io.ReadCloser
ffmpegStdin io.WriteCloser
@ -180,24 +180,24 @@ func (r *Revid) reset(config Config) error {
}
switch r.config.InputCodec {
case H264:
r.Log(Info, "Using H264 parser!")
r.parse = parse.H264
r.Log(Info, "using H264 lexer")
r.lexTo = lex.H264
case Mjpeg:
r.Log(Info, "Using MJPEG parser!")
r.parse = parse.MJPEG
r.Log(Info, "using MJPEG lexer")
r.lexTo = lex.MJPEG
}
switch r.config.Packetization {
case None:
// no packetisation - Revid output chan grabs raw data straight from parser
r.parse = func(dst io.Writer, src io.Reader, _ time.Duration) error {
r.lexTo = func(dst stream.Encoder, src io.Reader, _ time.Duration) error {
// FIXME(kortschak): Reduce this allocation mess. It exists
// because we do not know that the dst will not modify the
// buffer. It shouldn't, but ...
for {
var b [4 << 10]byte
n, rerr := src.Read(b[:])
_, werr := dst.Write(b[:n])
werr := dst.Encode(b[:n])
if rerr != nil {
return rerr
}
@ -473,7 +473,7 @@ func (r *Revid) startRaspivid() error {
r.inputReader = stdout
go func() {
r.Log(Info, "Reading camera data!")
r.parse(chunkWriter{r.encoder}, r.inputReader, 0)
r.lexTo(r.encoder, r.inputReader, 0)
r.Log(Info, "Not trying to read from camera anymore!")
}()
return nil
@ -496,19 +496,5 @@ func (r *Revid) setupInputForFile() error {
defer f.Close()
// TODO(kortschak): Maybe we want a context.Context-aware parser that we can stop.
return r.parse(chunkWriter{r.encoder}, f, delay)
}
// TODO(kortschak): Remove this type and revise the signature of
// the parsers to accept an stream.Encoder.
//
// chunkWriter is a shim between the new function-based approach
// and the old flow-based approach.
type chunkWriter struct {
stream.Encoder
}
func (w chunkWriter) Write(b []byte) (int, error) {
err := w.Encoder.Encode(b)
return len(b), err
return r.lexTo(r.encoder, f, delay)
}