56 lines
1.1 KiB
Go
56 lines
1.1 KiB
Go
|
package jsonlines
|
||
|
|
|
||
|
|
import (
|
||
|
|
"bufio"
|
||
|
|
"bytes"
|
||
|
|
"io"
|
||
|
|
"log/slog"
|
||
|
|
"sync/atomic"
|
||
|
|
"time"
|
||
|
|
)
|
||
|
|
|
||
|
|
type Stream struct {
|
||
|
|
vendor string
|
||
|
|
onBad func(line []byte, reason string)
|
||
|
|
lastLine atomic.Int64
|
||
|
|
}
|
||
|
|
|
||
|
|
func New(vendor string, onBad func(line []byte, reason string)) *Stream {
|
||
|
|
s := &Stream{vendor: vendor, onBad: onBad}
|
||
|
|
s.lastLine.Store(time.Now().UnixNano())
|
||
|
|
return s
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Stream) LastLine() time.Time {
|
||
|
|
return time.Unix(0, s.lastLine.Load())
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Stream) BadLine(line []byte, reason string) {
|
||
|
|
slog.Warn("bad "+s.vendor+" line", "reason", reason, "line", string(line))
|
||
|
|
if s.onBad != nil {
|
||
|
|
s.onBad(line, reason)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Stream) Read(stdout io.Reader, handle func(line []byte)) {
|
||
|
|
r := bufio.NewReader(stdout)
|
||
|
|
for {
|
||
|
|
line, err := r.ReadBytes('\n')
|
||
|
|
line = bytes.TrimRight(line, "\r\n")
|
||
|
|
if len(line) > 0 {
|
||
|
|
s.lastLine.Store(time.Now().UnixNano())
|
||
|
|
if line[0] == '{' {
|
||
|
|
handle(line)
|
||
|
|
} else {
|
||
|
|
s.BadLine(line, "not a JSON object")
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
if err != io.EOF {
|
||
|
|
slog.Error("read "+s.vendor+" stdout", "err", err)
|
||
|
|
}
|
||
|
|
return
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|