apps/core
1
0
Fork 0
mirror of https://github.com/datarhei/core.git synced 2025-10-06 16:37:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

Process []byte instread of string in parser

Browse Source
This commit is contained in:
Ingo Oppermann
2024-07-26 11:31:47 +02:00
parent d391e274d7
commit 70a49f8bdb
7 changed files with 209 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
ffmpeg/parse/parser.go
View File

@@ -1,11 +1,11 @@
package parse package parse
import ( import (
"bytes"
"container/ring" "container/ring"
"fmt" "fmt"
"regexp" "regexp"
"strconv" "strconv"
"strings"
"sync" "sync"
"time" "time"
@@ -200,14 +200,14 @@ func New(config Config) Parser {
return p return p
} }
func (p *parser) Parse(line string) uint64 { func (p *parser) Parse(line []byte) uint64 {
isDefaultProgress := strings.HasPrefix(line, "frame=") isDefaultProgress := bytes.HasPrefix(line, []byte("frame="))
isFFmpegInputs := strings.HasPrefix(line, "ffmpeg.inputs:") isFFmpegInputs := bytes.HasPrefix(line, []byte("ffmpeg.inputs:"))
isFFmpegOutputs := strings.HasPrefix(line, "ffmpeg.outputs:") isFFmpegOutputs := bytes.HasPrefix(line, []byte("ffmpeg.outputs:"))
isFFmpegMapping := strings.HasPrefix(line, "ffmpeg.mapping:") isFFmpegMapping := bytes.HasPrefix(line, []byte("ffmpeg.mapping:"))
isFFmpegProgress := strings.HasPrefix(line, "ffmpeg.progress:") isFFmpegProgress := bytes.HasPrefix(line, []byte("ffmpeg.progress:"))
isHLSStreamMap := strings.HasPrefix(line, "hls.streammap:") isHLSStreamMap := bytes.HasPrefix(line, []byte("hls.streammap:"))
isAVstreamProgress := strings.HasPrefix(line, "avstream.progress:") isAVstreamProgress := bytes.HasPrefix(line, []byte("avstream.progress:"))
p.lock.log.Lock() p.lock.log.Lock()
if p.logStart.IsZero() { if p.logStart.IsZero() {
@@ -235,7 +235,7 @@ func (p *parser) Parse(line string) uint64 {
} }
if isFFmpegInputs { if isFFmpegInputs {
if err := p.parseFFmpegIO("input", strings.TrimPrefix(line, "ffmpeg.inputs:")); err != nil { if err := p.parseFFmpegIO("input", bytes.TrimPrefix(line, []byte("ffmpeg.inputs:"))); err != nil {
p.logger.WithFields(log.Fields{ p.logger.WithFields(log.Fields{
"line": line, "line": line,
"error": err, "error": err,
@@ -246,7 +246,7 @@ func (p *parser) Parse(line string) uint64 {
} }
if isHLSStreamMap { if isHLSStreamMap {
if err := p.parseHLSStreamMap(strings.TrimPrefix(line, "hls.streammap:")); err != nil { if err := p.parseHLSStreamMap(bytes.TrimPrefix(line, []byte("hls.streammap:"))); err != nil {
p.logger.WithFields(log.Fields{ p.logger.WithFields(log.Fields{
"line": line, "line": line,
"error": err, "error": err,
@@ -257,7 +257,7 @@ func (p *parser) Parse(line string) uint64 {
} }
if isFFmpegOutputs { if isFFmpegOutputs {
if err := p.parseFFmpegIO("output", strings.TrimPrefix(line, "ffmpeg.outputs:")); err != nil { if err := p.parseFFmpegIO("output", bytes.TrimPrefix(line, []byte("ffmpeg.outputs:"))); err != nil {
p.logger.WithFields(log.Fields{ p.logger.WithFields(log.Fields{
"line": line, "line": line,
"error": err, "error": err,
@@ -287,7 +287,7 @@ func (p *parser) Parse(line string) uint64 {
} }
if isFFmpegMapping { if isFFmpegMapping {
if err := p.parseFFmpegMapping(strings.TrimPrefix(line, "ffmpeg.mapping:")); err != nil { if err := p.parseFFmpegMapping(bytes.TrimPrefix(line, []byte("ffmpeg.mapping:"))); err != nil {
p.logger.WithFields(log.Fields{ p.logger.WithFields(log.Fields{
"line": line, "line": line,
"error": err, "error": err,
@@ -298,13 +298,14 @@ func (p *parser) Parse(line string) uint64 {
} }
if !isDefaultProgress && !isFFmpegProgress && !isAVstreamProgress { if !isDefaultProgress && !isFFmpegProgress && !isAVstreamProgress {
stringLine := string(line)
// Write the current non-progress line to the log // Write the current non-progress line to the log
p.addLog(line) p.addLog(stringLine)
p.lock.prelude.Lock() p.lock.prelude.Lock()
if !p.prelude.done { if !p.prelude.done {
if len(p.prelude.data) < p.prelude.headLines { if len(p.prelude.data) < p.prelude.headLines {
p.prelude.data = append(p.prelude.data, line) p.prelude.data = append(p.prelude.data, stringLine)
} else { } else {
p.prelude.tail.Value = line p.prelude.tail.Value = line
p.prelude.tail = p.prelude.tail.Next() p.prelude.tail = p.prelude.tail.Next()
@@ -315,8 +316,8 @@ func (p *parser) Parse(line string) uint64 {
p.lock.log.Lock() p.lock.log.Lock()
for _, pattern := range p.logpatterns.patterns { for _, pattern := range p.logpatterns.patterns {
if pattern.MatchString(line) { if pattern.Match(line) {
p.logpatterns.matches = append(p.logpatterns.matches, line) p.logpatterns.matches = append(p.logpatterns.matches, stringLine)
} }
} }
p.lock.log.Unlock() p.lock.log.Unlock()
@@ -363,7 +364,7 @@ func (p *parser) Parse(line string) uint64 {
// Update the progress // Update the progress
if isAVstreamProgress { if isAVstreamProgress {
if err := p.parseAVstreamProgress(strings.TrimPrefix(line, "avstream.progress:")); err != nil { if err := p.parseAVstreamProgress(bytes.TrimPrefix(line, []byte("avstream.progress:"))); err != nil {
p.logger.WithFields(log.Fields{ p.logger.WithFields(log.Fields{
"line": line, "line": line,
"error": err, "error": err,
@@ -382,7 +383,7 @@ func (p *parser) Parse(line string) uint64 {
return 0 return 0
} }
} else if isFFmpegProgress { } else if isFFmpegProgress {
if err := p.parseFFmpegProgress(strings.TrimPrefix(line, "ffmpeg.progress:")); err != nil { if err := p.parseFFmpegProgress(bytes.TrimPrefix(line, []byte("ffmpeg.progress:"))); err != nil {
p.logger.WithFields(log.Fields{ p.logger.WithFields(log.Fields{
"line": line, "line": line,
"error": err, "error": err,
@@ -466,48 +467,48 @@ func (p *parser) Parse(line string) uint64 {
return pFrames return pFrames
} }
func (p *parser) parseDefaultProgress(line string) error { func (p *parser) parseDefaultProgress(line []byte) error {
var matches []string var matches [][]byte
if matches = p.re.frame.FindStringSubmatch(line); matches != nil { if matches = p.re.frame.FindSubmatch(line); matches != nil {
if x, err := strconv.ParseUint(matches[1], 10, 64); err == nil { if x, err := strconv.ParseUint(string(matches[1]), 10, 64); err == nil {
p.progress.ffmpeg.Frame = x p.progress.ffmpeg.Frame = x
} }
} }
if matches = p.re.quantizer.FindStringSubmatch(line); matches != nil { if matches = p.re.quantizer.FindSubmatch(line); matches != nil {
if x, err := strconv.ParseFloat(matches[1], 64); err == nil { if x, err := strconv.ParseFloat(string(matches[1]), 64); err == nil {
p.progress.ffmpeg.Quantizer = x p.progress.ffmpeg.Quantizer = x
} }
} }
if matches = p.re.size.FindStringSubmatch(line); matches != nil { if matches = p.re.size.FindSubmatch(line); matches != nil {
if x, err := strconv.ParseUint(matches[1], 10, 64); err == nil { if x, err := strconv.ParseUint(string(matches[1]), 10, 64); err == nil {
p.progress.ffmpeg.Size = x * 1024 p.progress.ffmpeg.Size = x * 1024
} }
} }
if matches = p.re.time.FindStringSubmatch(line); matches != nil { if matches = p.re.time.FindSubmatch(line); matches != nil {
s := fmt.Sprintf("%sh%sm%ss%s0ms", matches[1], matches[2], matches[3], matches[4]) s := fmt.Sprintf("%sh%sm%ss%s0ms", string(matches[1]), string(matches[2]), string(matches[3]), string(matches[4]))
if x, err := time.ParseDuration(s); err == nil { if x, err := time.ParseDuration(s); err == nil {
p.progress.ffmpeg.Time.Duration = x p.progress.ffmpeg.Time.Duration = x
} }
} }
if matches = p.re.speed.FindStringSubmatch(line); matches != nil { if matches = p.re.speed.FindSubmatch(line); matches != nil {
if x, err := strconv.ParseFloat(matches[1], 64); err == nil { if x, err := strconv.ParseFloat(string(matches[1]), 64); err == nil {
p.progress.ffmpeg.Speed = x p.progress.ffmpeg.Speed = x
} }
} }
if matches = p.re.drop.FindStringSubmatch(line); matches != nil { if matches = p.re.drop.FindSubmatch(line); matches != nil {
if x, err := strconv.ParseUint(matches[1], 10, 64); err == nil { if x, err := strconv.ParseUint(string(matches[1]), 10, 64); err == nil {
p.progress.ffmpeg.Drop = x p.progress.ffmpeg.Drop = x
} }
} }
if matches = p.re.dup.FindStringSubmatch(line); matches != nil { if matches = p.re.dup.FindSubmatch(line); matches != nil {
if x, err := strconv.ParseUint(matches[1], 10, 64); err == nil { if x, err := strconv.ParseUint(string(matches[1]), 10, 64); err == nil {
p.progress.ffmpeg.Dup = x p.progress.ffmpeg.Dup = x
} }
} }
@@ -515,10 +516,10 @@ func (p *parser) parseDefaultProgress(line string) error {
return nil return nil
} }
func (p *parser) parseFFmpegIO(kind, line string) error { func (p *parser) parseFFmpegIO(kind string, line []byte) error {
processIO := []ffmpegProcessIO{} processIO := []ffmpegProcessIO{}
err := json.Unmarshal([]byte(line), &processIO) err := json.Unmarshal(line, &processIO)
if err != nil { if err != nil {
return err return err
} }
@@ -542,10 +543,10 @@ func (p *parser) parseFFmpegIO(kind, line string) error {
return nil return nil
} }
func (p *parser) parseFFmpegMapping(line string) error { func (p *parser) parseFFmpegMapping(line []byte) error {
mapping := ffmpegStreamMapping{} mapping := ffmpegStreamMapping{}
err := json.Unmarshal([]byte(line), &mapping) err := json.Unmarshal(line, &mapping)
if err != nil { if err != nil {
return err return err
} }
@@ -555,10 +556,10 @@ func (p *parser) parseFFmpegMapping(line string) error {
return nil return nil
} }
func (p *parser) parseHLSStreamMap(line string) error { func (p *parser) parseHLSStreamMap(line []byte) error {
mapping := ffmpegHLSStreamMap{} mapping := ffmpegHLSStreamMap{}
err := json.Unmarshal([]byte(line), &mapping) err := json.Unmarshal(line, &mapping)
if err != nil { if err != nil {
return err return err
} }
@@ -568,10 +569,10 @@ func (p *parser) parseHLSStreamMap(line string) error {
return nil return nil
} }
func (p *parser) parseFFmpegProgress(line string) error { func (p *parser) parseFFmpegProgress(line []byte) error {
progress := ffmpegProgress{} progress := ffmpegProgress{}
err := json.Unmarshal([]byte(line), &progress) err := json.Unmarshal(line, &progress)
if err != nil { if err != nil {
return err return err
} }
@@ -609,10 +610,10 @@ func (p *parser) parseFFmpegProgress(line string) error {
return nil return nil
} }
func (p *parser) parseAVstreamProgress(line string) error { func (p *parser) parseAVstreamProgress(line []byte) error {
progress := ffmpegAVstream{} progress := ffmpegAVstream{}
err := json.Unmarshal([]byte(line), &progress) err := json.Unmarshal(line, &progress)
if err != nil { if err != nil {
return err return err
} }
@@ -766,7 +767,7 @@ func (p *parser) addLog(line string) {
p.log.Value = process.Line{ p.log.Value = process.Line{
Timestamp: time.Now(), Timestamp: time.Now(),
Data: line, Data: p.lastLogline,
} }
p.log = p.log.Next() p.log = p.log.Next()
} }

114
ffmpeg/parse/parser_test.go
View File

@@ -17,7 +17,7 @@ func TestParserProgress(t *testing.T) {
}).(*parser) }).(*parser)
parser.prelude.done = true parser.prelude.done = true
parser.Parse("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463") parser.Parse([]byte("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463"))
d, _ := time.ParseDuration("3m58s440ms") d, _ := time.ParseDuration("3m58s440ms")
wantP := Progress{ wantP := Progress{
@@ -66,7 +66,7 @@ func TestParserPrelude(t *testing.T) {
require.Equal(t, 0, len(log)) require.Equal(t, 0, len(log))
parser.Parse("prelude") parser.Parse([]byte("prelude"))
log = parser.Prelude() log = parser.Prelude()
@@ -85,7 +85,7 @@ func TestParserLongPrelude(t *testing.T) {
require.Equal(t, 0, len(log)) require.Equal(t, 0, len(log))
for i := 0; i < 150; i++ { for i := 0; i < 150; i++ {
parser.Parse(fmt.Sprintf("prelude %3d", i)) parser.Parse([]byte(fmt.Sprintf("prelude %3d", i)))
} }
log = parser.Prelude() log = parser.Prelude()
@@ -105,7 +105,7 @@ func TestParserVeryLongPrelude(t *testing.T) {
require.Equal(t, 0, len(log)) require.Equal(t, 0, len(log))
for i := 0; i < 300; i++ { for i := 0; i < 300; i++ {
parser.Parse(fmt.Sprintf("prelude %3d", i)) parser.Parse([]byte(fmt.Sprintf("prelude %3d", i)))
} }
log = parser.Prelude() log = parser.Prelude()
@@ -122,7 +122,7 @@ func TestParserLog(t *testing.T) {
require.Equal(t, 0, len(log)) require.Equal(t, 0, len(log))
parser.Parse("bla") parser.Parse([]byte("bla"))
log = parser.Log() log = parser.Log()
@@ -134,19 +134,19 @@ func TestParserLastLogLine(t *testing.T) {
LogLines: 20, LogLines: 20,
}).(*parser) }).(*parser)
parser.Parse("foo") parser.Parse([]byte("foo"))
line := parser.LastLogline() line := parser.LastLogline()
require.Equal(t, "foo", line) require.Equal(t, "foo", line)
parser.prelude.done = true parser.prelude.done = true
parser.Parse("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463") parser.Parse([]byte("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463"))
// progress lines are not logged // progress lines are not logged
line = parser.LastLogline() line = parser.LastLogline()
require.Equal(t, "foo", line) require.Equal(t, "foo", line)
parser.Parse("bar") parser.Parse([]byte("bar"))
line = parser.LastLogline() line = parser.LastLogline()
require.Equal(t, "bar", line) require.Equal(t, "bar", line)
} }
@@ -158,10 +158,10 @@ func TestParserLogHistory(t *testing.T) {
}).(*parser) }).(*parser)
for i := 0; i < 7; i++ { for i := 0; i < 7; i++ {
parser.Parse("bla") parser.Parse([]byte("bla"))
parser.prelude.done = true parser.prelude.done = true
parser.Parse("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463") parser.Parse([]byte("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463"))
history := parser.ReportHistory() history := parser.ReportHistory()
require.Equal(t, int(math.Min(float64(i), 5)), len(history)) require.Equal(t, int(math.Min(float64(i), 5)), len(history))
@@ -207,10 +207,10 @@ func TestParserImportLogHistory(t *testing.T) {
}).(*parser) }).(*parser)
for i := 0; i < 7; i++ { for i := 0; i < 7; i++ {
parser.Parse("bla") parser.Parse([]byte("bla"))
parser.prelude.done = true parser.prelude.done = true
parser.Parse("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463") parser.Parse([]byte("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463"))
history := parser.ReportHistory() history := parser.ReportHistory()
require.Equal(t, int(math.Min(float64(i), 5)), len(history)) require.Equal(t, int(math.Min(float64(i), 5)), len(history))
@@ -271,10 +271,10 @@ func TestParserLogHistoryLength(t *testing.T) {
require.Equal(t, 0, len(history)) require.Equal(t, 0, len(history))
for i := 0; i < 5; i++ { for i := 0; i < 5; i++ {
parser.Parse("bla") parser.Parse([]byte("bla"))
parser.prelude.done = true parser.prelude.done = true
parser.Parse("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463") parser.Parse([]byte("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463"))
parser.Stop("finished", process.Usage{}) parser.Stop("finished", process.Usage{})
} }
@@ -294,10 +294,10 @@ func TestParserLogMinimalHistoryLength(t *testing.T) {
require.Equal(t, 0, len(history)) require.Equal(t, 0, len(history))
for i := 0; i < 42; i++ { for i := 0; i < 42; i++ {
parser.Parse("bla") parser.Parse([]byte("bla"))
parser.prelude.done = true parser.prelude.done = true
parser.Parse("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463") parser.Parse([]byte("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463"))
parser.Stop("finished", process.Usage{}) parser.Stop("finished", process.Usage{})
} }
@@ -353,10 +353,10 @@ func TestParserLogMinimalHistoryLengthWithoutFullHistory(t *testing.T) {
require.Equal(t, 0, len(history)) require.Equal(t, 0, len(history))
for i := 0; i < 15; i++ { for i := 0; i < 15; i++ {
parser.Parse("bla") parser.Parse([]byte("bla"))
parser.prelude.done = true parser.prelude.done = true
parser.Parse("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463") parser.Parse([]byte("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463"))
parser.Stop("finished", process.Usage{}) parser.Stop("finished", process.Usage{})
} }
@@ -375,10 +375,10 @@ func TestParserLogHistorySearch(t *testing.T) {
LogHistory: 5, LogHistory: 5,
}).(*parser) }).(*parser)
parser.Parse("foo") parser.Parse([]byte("foo"))
parser.prelude.done = true parser.prelude.done = true
parser.Parse("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463") parser.Parse([]byte("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463"))
parser.Stop("finished", process.Usage{}) parser.Stop("finished", process.Usage{})
@@ -388,10 +388,10 @@ func TestParserLogHistorySearch(t *testing.T) {
parser.ResetLog() parser.ResetLog()
parser.Parse("bar") parser.Parse([]byte("bar"))
parser.prelude.done = true parser.prelude.done = true
parser.Parse("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463") parser.Parse([]byte("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463"))
parser.Stop("finished", process.Usage{}) parser.Stop("finished", process.Usage{})
@@ -401,10 +401,10 @@ func TestParserLogHistorySearch(t *testing.T) {
parser.ResetLog() parser.ResetLog()
parser.Parse("foobar") parser.Parse([]byte("foobar"))
parser.prelude.done = true parser.prelude.done = true
parser.Parse("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463") parser.Parse([]byte("frame= 5968 fps= 25 q=19.4 size=443kB time=00:03:58.44 bitrate=5632kbits/s speed=0.999x skip=9733 drop=3522 dup=87463"))
parser.Stop("failed", process.Usage{}) parser.Stop("failed", process.Usage{})
@@ -466,7 +466,7 @@ func TestParserReset(t *testing.T) {
require.Equal(t, 0, len(log)) require.Equal(t, 0, len(log))
require.Equal(t, 0, len(prelude)) require.Equal(t, 0, len(prelude))
parser.Parse("prelude") parser.Parse([]byte("prelude"))
log = parser.Log() log = parser.Log()
prelude = parser.Prelude() prelude = parser.Prelude()
@@ -543,7 +543,7 @@ frame= 58 fps= 25 q=-1.0 Lsize=N/A time=00:00:02.32 bitrate=N/A speed=0.999x`
data := strings.Split(rawdata, "\n") data := strings.Split(rawdata, "\n")
for _, d := range data { for _, d := range data {
parser.Parse(d) parser.Parse([]byte(d))
} }
require.Equal(t, 3, len(parser.process.input), "expected 3 inputs") require.Equal(t, 3, len(parser.process.input), "expected 3 inputs")
@@ -602,7 +602,7 @@ frame= 58 fps= 25 q=-1.0 Lsize=N/A time=00:00:02.32 bitrate=N/A speed=0.999x`
data := strings.Split(rawdata, "\n") data := strings.Split(rawdata, "\n")
for _, d := range data { for _, d := range data {
parser.Parse(d) parser.Parse([]byte(d))
} }
require.Equal(t, 2, len(parser.process.input), "expected 2 inputs") require.Equal(t, 2, len(parser.process.input), "expected 2 inputs")
@@ -685,7 +685,7 @@ ffmpeg.progress:{"inputs":[{"index":0,"stream":0,"frame":21,"packet":24,"size_kb
data := strings.Split(rawdata, "\n") data := strings.Split(rawdata, "\n")
for _, d := range data { for _, d := range data {
parser.Parse(d) parser.Parse([]byte(d))
} }
require.Equal(t, 1, len(parser.process.input), "expected 1 input") require.Equal(t, 1, len(parser.process.input), "expected 1 input")
@@ -768,7 +768,7 @@ ffmpeg.progress:{"inputs":[{"index":0,"stream":0,"frame":21,"packet":24,"size_kb
data := strings.Split(rawdata, "\n") data := strings.Split(rawdata, "\n")
for _, d := range data { for _, d := range data {
parser.Parse(d) parser.Parse([]byte(d))
} }
require.Equal(t, 1, len(parser.process.input), "expected 1 input") require.Equal(t, 1, len(parser.process.input), "expected 1 input")
@@ -852,7 +852,7 @@ ffmpeg.progress:{"inputs":[{"index":0,"stream":0,"frame":21,"packet":24,"size_kb
data := strings.Split(rawdata, "\n") data := strings.Split(rawdata, "\n")
for _, d := range data { for _, d := range data {
parser.Parse(d) parser.Parse([]byte(d))
} }
require.Equal(t, 1, len(parser.process.input), "expected 1 input") require.Equal(t, 1, len(parser.process.input), "expected 1 input")
@@ -864,10 +864,10 @@ func TestParserProgressPlayout(t *testing.T) {
LogLines: 20, LogLines: 20,
}).(*parser) }).(*parser)
parser.Parse(`ffmpeg.inputs:[{"url":"playout:https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk.m3u8","format":"playout","index":0,"stream":0,"type":"video","codec":"h264","coder":"h264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":20.666666,"pix_fmt":"yuvj420p","width":1280,"height":720}]`) parser.Parse([]byte(`ffmpeg.inputs:[{"url":"playout:https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk.m3u8","format":"playout","index":0,"stream":0,"type":"video","codec":"h264","coder":"h264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":20.666666,"pix_fmt":"yuvj420p","width":1280,"height":720}]`))
parser.Parse(`ffmpeg.outputs:[{"url":"/dev/null","format":"flv","index":0,"stream":0,"type":"video","codec":"h264","coder":"libx264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":1280,"height":720},{"url":"/dev/null","format":"mp4","index":1,"stream":0,"type":"video","codec":"h264","coder":"copy","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":20.666666,"pix_fmt":"yuvj420p","width":1280,"height":720}]`) parser.Parse([]byte(`ffmpeg.outputs:[{"url":"/dev/null","format":"flv","index":0,"stream":0,"type":"video","codec":"h264","coder":"libx264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":1280,"height":720},{"url":"/dev/null","format":"mp4","index":1,"stream":0,"type":"video","codec":"h264","coder":"copy","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":20.666666,"pix_fmt":"yuvj420p","width":1280,"height":720}]`))
parser.Parse(`ffmpeg.progress:{"inputs":[{"index":0,"stream":0,"frame":7,"keyframe":1,"packet":11,"size_kb":226,"size_bytes":42}],"outputs":[{"index":0,"stream":0,"frame":7,"keyframe":1,"packet":0,"q":0.0,"size_kb":0,"size_bytes":5,"extradata_size_bytes":32},{"index":1,"stream":0,"frame":11,"packet":11,"q":-1.0,"size_kb":226}],"frame":7,"packet":0,"q":0.0,"size_kb":226,"time":"0h0m0.56s","speed":0.4,"dup":0,"drop":0}`) parser.Parse([]byte(`ffmpeg.progress:{"inputs":[{"index":0,"stream":0,"frame":7,"keyframe":1,"packet":11,"size_kb":226,"size_bytes":42}],"outputs":[{"index":0,"stream":0,"frame":7,"keyframe":1,"packet":0,"q":0.0,"size_kb":0,"size_bytes":5,"extradata_size_bytes":32},{"index":1,"stream":0,"frame":11,"packet":11,"q":-1.0,"size_kb":226}],"frame":7,"packet":0,"q":0.0,"size_kb":226,"time":"0h0m0.56s","speed":0.4,"dup":0,"drop":0}`))
parser.Parse(`avstream.progress:{"id":"playout:https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk.m3u8","url":"https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk.m3u8","stream":0,"queue":140,"aqueue":42,"dup":5,"drop":8,"enc":7,"looping":true,"duplicating":true,"gop":"key","mode":"live","input":{"state":"running","packet":148,"size_kb":1529,"time":5},"output":{"state":"running","packet":8,"size_kb":128,"time":1},"swap":{"url":"","status":"waiting","lasturl":"","lasterror":""}}`) parser.Parse([]byte(`avstream.progress:{"id":"playout:https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk.m3u8","url":"https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk.m3u8","stream":0,"queue":140,"aqueue":42,"dup":5,"drop":8,"enc":7,"looping":true,"duplicating":true,"gop":"key","mode":"live","input":{"state":"running","packet":148,"size_kb":1529,"time":5},"output":{"state":"running","packet":8,"size_kb":128,"time":1},"swap":{"url":"","status":"waiting","lasturl":"","lasterror":""}}`))
progress := parser.Progress() progress := parser.Progress()
@@ -996,11 +996,11 @@ func TestParserStreamMapping(t *testing.T) {
LogLines: 20, LogLines: 20,
}).(*parser) }).(*parser)
parser.Parse(`ffmpeg.inputs:[{"url":"https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk_720.m3u8","format":"hls","index":0,"stream":0,"type":"video","codec":"h264","coder":"h264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":1280,"height":720},{"url":"https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk_1440.m3u8","format":"hls","index":1,"stream":0,"type":"video","codec":"h264","coder":"h264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":2560,"height":1440},{"url":"anullsrc=r=44100:cl=mono","format":"lavfi","index":2,"stream":0,"type":"audio","codec":"pcm_u8","coder":"pcm_u8","bitrate_kbps":352,"duration_sec":0.000000,"language":"und","sampling_hz":44100,"layout":"mono","channels":1}]`) parser.Parse([]byte(`ffmpeg.inputs:[{"url":"https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk_720.m3u8","format":"hls","index":0,"stream":0,"type":"video","codec":"h264","coder":"h264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":1280,"height":720},{"url":"https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk_1440.m3u8","format":"hls","index":1,"stream":0,"type":"video","codec":"h264","coder":"h264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":2560,"height":1440},{"url":"anullsrc=r=44100:cl=mono","format":"lavfi","index":2,"stream":0,"type":"audio","codec":"pcm_u8","coder":"pcm_u8","bitrate_kbps":352,"duration_sec":0.000000,"language":"und","sampling_hz":44100,"layout":"mono","channels":1}]`))
parser.Parse(`hls.streammap:{"address":"http://127.0.0.1:8080/memfs/live/%v.m3u8","variants":[{"variant":0,"address":"http://127.0.0.1:8080/memfs/live/0.m3u8","streams":[0,2]},{"variant":1,"address":"http://127.0.0.1:8080/memfs/live/1.m3u8","streams":[1,3]}]}`) parser.Parse([]byte(`hls.streammap:{"address":"http://127.0.0.1:8080/memfs/live/%v.m3u8","variants":[{"variant":0,"address":"http://127.0.0.1:8080/memfs/live/0.m3u8","streams":[0,2]},{"variant":1,"address":"http://127.0.0.1:8080/memfs/live/1.m3u8","streams":[1,3]}]}`))
parser.Parse(`ffmpeg.outputs:[{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":0,"type":"video","codec":"h264","coder":"copy","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":1280,"height":720},{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":1,"type":"video","codec":"h264","coder":"copy","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":2560,"height":1440},{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":2,"type":"audio","codec":"aac","coder":"aac","bitrate_kbps":69,"duration_sec":0.000000,"language":"und","sampling_hz":44100,"layout":"mono","channels":1},{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":3,"type":"audio","codec":"aac","coder":"aac","bitrate_kbps":69,"duration_sec":0.000000,"language":"und","sampling_hz":44100,"layout":"mono","channels":1}]`) parser.Parse([]byte(`ffmpeg.outputs:[{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":0,"type":"video","codec":"h264","coder":"copy","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":1280,"height":720},{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":1,"type":"video","codec":"h264","coder":"copy","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":2560,"height":1440},{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":2,"type":"audio","codec":"aac","coder":"aac","bitrate_kbps":69,"duration_sec":0.000000,"language":"und","sampling_hz":44100,"layout":"mono","channels":1},{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":3,"type":"audio","codec":"aac","coder":"aac","bitrate_kbps":69,"duration_sec":0.000000,"language":"und","sampling_hz":44100,"layout":"mono","channels":1}]`))
parser.Parse(`ffmpeg.mapping:{"graphs":[{"index":0,"graph":[{"src_name":"Parsed_null_0","src_filter":"null","dst_name":"format","dst_filter":"format","inpad":"default","outpad":"default","timebase": "1/90000","type":"video","format":"yuvj420p","width":1280,"height":720},{"src_name":"graph 0 input from stream 0:0","src_filter":"buffer","dst_name":"Parsed_null_0","dst_filter":"null","inpad":"default","outpad":"default","timebase": "1/90000","type":"video","format":"yuvj420p","width":1280,"height":720},{"src_name":"format","src_filter":"format","dst_name":"out_0_0","dst_filter":"buffersink","inpad":"default","outpad":"default","timebase": "1/90000","type":"video","format":"yuvj420p","width":1280,"height":720}]},{"index":1,"graph":[{"src_name":"Parsed_anull_0","src_filter":"anull","dst_name":"auto_aresample_0","dst_filter":"aresample","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"u8","sampling_hz":44100,"layout":"mono"},{"src_name":"graph_1_in_2_0","src_filter":"abuffer","dst_name":"Parsed_anull_0","dst_filter":"anull","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"u8","sampling_hz":44100,"layout":"mono"},{"src_name":"format_out_0_2","src_filter":"aformat","dst_name":"out_0_2","dst_filter":"abuffersink","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"fltp","sampling_hz":44100,"layout":"mono"},{"src_name":"auto_aresample_0","src_filter":"aresample","dst_name":"format_out_0_2","dst_filter":"aformat","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"fltp","sampling_hz":44100,"layout":"mono"}]},{"index":2,"graph":[{"src_name":"Parsed_anull_0","src_filter":"anull","dst_name":"auto_aresample_0","dst_filter":"aresample","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"u8","sampling_hz":44100,"layout":"mono"},{"src_name":"graph_2_in_2_0","src_filter":"abuffer","dst_name":"Parsed_anull_0","dst_filter":"anull","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"u8","sampling_hz":44100,"layout":"mono"},{"src_name":"format_out_0_3","src_filter":"aformat","dst_name":"out_0_3","dst_filter":"abuffersink","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"fltp","sampling_hz":44100,"layout":"mono"},{"src_name":"auto_aresample_0","src_filter":"aresample","dst_name":"format_out_0_3","dst_filter":"aformat","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"fltp","sampling_hz":44100,"layout":"mono"}]}],"mapping":[{"input":{"index":0,"stream":0},"graph":{"index":0,"name":"graph 0 input from stream 0:0"},"output":null},{"input":{"index":2,"stream":0},"graph":{"index":1,"name":"graph_1_in_2_0"},"output":null},{"input":{"index":2,"stream":0},"graph":{"index":2,"name":"graph_2_in_2_0"},"output":null},{"input":null,"graph":{"index":0,"name":"out_0_0"},"output":{"index":0,"stream":0}},{"input":{"index":1,"stream":0},"output":{"index":0,"stream":1},"copy":true},{"input":null,"graph":{"index":1,"name":"out_0_2"},"output":{"index":0,"stream":2}},{"input":null,"graph":{"index":2,"name":"out_0_3"},"output":{"index":0,"stream":3}}]}`) parser.Parse([]byte(`ffmpeg.mapping:{"graphs":[{"index":0,"graph":[{"src_name":"Parsed_null_0","src_filter":"null","dst_name":"format","dst_filter":"format","inpad":"default","outpad":"default","timebase": "1/90000","type":"video","format":"yuvj420p","width":1280,"height":720},{"src_name":"graph 0 input from stream 0:0","src_filter":"buffer","dst_name":"Parsed_null_0","dst_filter":"null","inpad":"default","outpad":"default","timebase": "1/90000","type":"video","format":"yuvj420p","width":1280,"height":720},{"src_name":"format","src_filter":"format","dst_name":"out_0_0","dst_filter":"buffersink","inpad":"default","outpad":"default","timebase": "1/90000","type":"video","format":"yuvj420p","width":1280,"height":720}]},{"index":1,"graph":[{"src_name":"Parsed_anull_0","src_filter":"anull","dst_name":"auto_aresample_0","dst_filter":"aresample","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"u8","sampling_hz":44100,"layout":"mono"},{"src_name":"graph_1_in_2_0","src_filter":"abuffer","dst_name":"Parsed_anull_0","dst_filter":"anull","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"u8","sampling_hz":44100,"layout":"mono"},{"src_name":"format_out_0_2","src_filter":"aformat","dst_name":"out_0_2","dst_filter":"abuffersink","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"fltp","sampling_hz":44100,"layout":"mono"},{"src_name":"auto_aresample_0","src_filter":"aresample","dst_name":"format_out_0_2","dst_filter":"aformat","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"fltp","sampling_hz":44100,"layout":"mono"}]},{"index":2,"graph":[{"src_name":"Parsed_anull_0","src_filter":"anull","dst_name":"auto_aresample_0","dst_filter":"aresample","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"u8","sampling_hz":44100,"layout":"mono"},{"src_name":"graph_2_in_2_0","src_filter":"abuffer","dst_name":"Parsed_anull_0","dst_filter":"anull","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"u8","sampling_hz":44100,"layout":"mono"},{"src_name":"format_out_0_3","src_filter":"aformat","dst_name":"out_0_3","dst_filter":"abuffersink","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"fltp","sampling_hz":44100,"layout":"mono"},{"src_name":"auto_aresample_0","src_filter":"aresample","dst_name":"format_out_0_3","dst_filter":"aformat","inpad":"default","outpad":"default","timebase": "1/44100","type":"audio","format":"fltp","sampling_hz":44100,"layout":"mono"}]}],"mapping":[{"input":{"index":0,"stream":0},"graph":{"index":0,"name":"graph 0 input from stream 0:0"},"output":null},{"input":{"index":2,"stream":0},"graph":{"index":1,"name":"graph_1_in_2_0"},"output":null},{"input":{"index":2,"stream":0},"graph":{"index":2,"name":"graph_2_in_2_0"},"output":null},{"input":null,"graph":{"index":0,"name":"out_0_0"},"output":{"index":0,"stream":0}},{"input":{"index":1,"stream":0},"output":{"index":0,"stream":1},"copy":true},{"input":null,"graph":{"index":1,"name":"out_0_2"},"output":{"index":0,"stream":2}},{"input":null,"graph":{"index":2,"name":"out_0_3"},"output":{"index":0,"stream":3}}]}`))
parser.Parse(`ffmpeg.progress:{"inputs":[{"index":0,"stream":0,"framerate":{"min":24.975,"max":24.975,"avg":24.975},"frame":149,"keyframe":3,"packet":149,"size_kb":1467,"size_bytes":1501854},{"index":1,"stream":0,"framerate":{"min":24.975,"max":24.975,"avg":24.975},"frame":149,"keyframe":3,"packet":149,"size_kb":4428,"size_bytes":4534541},{"index":2,"stream":0,"framerate":{"min":43.066,"max":43.068,"avg":43.066},"frame":257,"keyframe":257,"packet":257,"size_kb":257,"size_bytes":263168}],"outputs":[{"index":0,"stream":0,"frame":149,"keyframe":3,"packet":149,"q":-1.0,"size_kb":1467,"size_bytes":1501923,"extradata_size_bytes":69},{"index":0,"stream":1,"frame":149,"keyframe":3,"packet":149,"q":-1.0,"size_kb":4428,"size_bytes":4534612,"extradata_size_bytes":71},{"index":0,"stream":2,"frame":257,"keyframe":256,"packet":256,"size_kb":1,"size_bytes":1046,"extradata_size_bytes":5},{"index":0,"stream":3,"frame":257,"keyframe":256,"packet":256,"size_kb":1,"size_bytes":1046,"extradata_size_bytes":5}],"frame":149,"packet":149,"q":-1.0,"size_kb":5897,"size_bytes":6038627,"time":"0h0m5.96s","speed":4.79,"dup":0,"drop":0}`) parser.Parse([]byte(`ffmpeg.progress:{"inputs":[{"index":0,"stream":0,"framerate":{"min":24.975,"max":24.975,"avg":24.975},"frame":149,"keyframe":3,"packet":149,"size_kb":1467,"size_bytes":1501854},{"index":1,"stream":0,"framerate":{"min":24.975,"max":24.975,"avg":24.975},"frame":149,"keyframe":3,"packet":149,"size_kb":4428,"size_bytes":4534541},{"index":2,"stream":0,"framerate":{"min":43.066,"max":43.068,"avg":43.066},"frame":257,"keyframe":257,"packet":257,"size_kb":257,"size_bytes":263168}],"outputs":[{"index":0,"stream":0,"frame":149,"keyframe":3,"packet":149,"q":-1.0,"size_kb":1467,"size_bytes":1501923,"extradata_size_bytes":69},{"index":0,"stream":1,"frame":149,"keyframe":3,"packet":149,"q":-1.0,"size_kb":4428,"size_bytes":4534612,"extradata_size_bytes":71},{"index":0,"stream":2,"frame":257,"keyframe":256,"packet":256,"size_kb":1,"size_bytes":1046,"extradata_size_bytes":5},{"index":0,"stream":3,"frame":257,"keyframe":256,"packet":256,"size_kb":1,"size_bytes":1046,"extradata_size_bytes":5}],"frame":149,"packet":149,"q":-1.0,"size_kb":5897,"size_bytes":6038627,"time":"0h0m5.96s","speed":4.79,"dup":0,"drop":0}`))
progress := parser.Progress() progress := parser.Progress()
@@ -1096,10 +1096,10 @@ func TestParserHLSMapping(t *testing.T) {
LogLines: 20, LogLines: 20,
}).(*parser) }).(*parser)
parser.Parse(`ffmpeg.inputs:[{"url":"https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk_720.m3u8","format":"hls","index":0,"stream":0,"type":"video","codec":"h264","coder":"h264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":1280,"height":720},{"url":"https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk_1440.m3u8","format":"hls","index":1,"stream":0,"type":"video","codec":"h264","coder":"h264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":2560,"height":1440},{"url":"anullsrc=r=44100:cl=mono","format":"lavfi","index":2,"stream":0,"type":"audio","codec":"pcm_u8","coder":"pcm_u8","bitrate_kbps":352,"duration_sec":0.000000,"language":"und","sampling_hz":44100,"layout":"mono","channels":1}]`) parser.Parse([]byte(`ffmpeg.inputs:[{"url":"https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk_720.m3u8","format":"hls","index":0,"stream":0,"type":"video","codec":"h264","coder":"h264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":1280,"height":720},{"url":"https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk_1440.m3u8","format":"hls","index":1,"stream":0,"type":"video","codec":"h264","coder":"h264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":2560,"height":1440},{"url":"anullsrc=r=44100:cl=mono","format":"lavfi","index":2,"stream":0,"type":"audio","codec":"pcm_u8","coder":"pcm_u8","bitrate_kbps":352,"duration_sec":0.000000,"language":"und","sampling_hz":44100,"layout":"mono","channels":1}]`))
parser.Parse(`hls.streammap:{"address":"http://127.0.0.1:8080/memfs/live/%v.m3u8","variants":[{"variant":0,"address":"http://127.0.0.1:8080/memfs/live/0.m3u8","streams":[0,2]},{"variant":1,"address":"http://127.0.0.1:8080/memfs/live/1.m3u8","streams":[1,3]}]}`) parser.Parse([]byte(`hls.streammap:{"address":"http://127.0.0.1:8080/memfs/live/%v.m3u8","variants":[{"variant":0,"address":"http://127.0.0.1:8080/memfs/live/0.m3u8","streams":[0,2]},{"variant":1,"address":"http://127.0.0.1:8080/memfs/live/1.m3u8","streams":[1,3]}]}`))
parser.Parse(`ffmpeg.outputs:[{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":0,"type":"video","codec":"h264","coder":"copy","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":1280,"height":720},{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":1,"type":"video","codec":"h264","coder":"copy","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":2560,"height":1440},{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":2,"type":"audio","codec":"aac","coder":"aac","bitrate_kbps":69,"duration_sec":0.000000,"language":"und","sampling_hz":44100,"layout":"mono","channels":1},{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":3,"type":"audio","codec":"aac","coder":"aac","bitrate_kbps":69,"duration_sec":0.000000,"language":"und","sampling_hz":44100,"layout":"mono","channels":1}]`) parser.Parse([]byte(`ffmpeg.outputs:[{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":0,"type":"video","codec":"h264","coder":"copy","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":1280,"height":720},{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":1,"type":"video","codec":"h264","coder":"copy","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":2560,"height":1440},{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":2,"type":"audio","codec":"aac","coder":"aac","bitrate_kbps":69,"duration_sec":0.000000,"language":"und","sampling_hz":44100,"layout":"mono","channels":1},{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":3,"type":"audio","codec":"aac","coder":"aac","bitrate_kbps":69,"duration_sec":0.000000,"language":"und","sampling_hz":44100,"layout":"mono","channels":1}]`))
parser.Parse(`ffmpeg.progress:{"inputs":[{"index":0,"stream":0,"framerate":{"min":24.975,"max":24.975,"avg":24.975},"frame":149,"keyframe":3,"packet":149,"size_kb":1467,"size_bytes":1501854},{"index":1,"stream":0,"framerate":{"min":24.975,"max":24.975,"avg":24.975},"frame":149,"keyframe":3,"packet":149,"size_kb":4428,"size_bytes":4534541},{"index":2,"stream":0,"framerate":{"min":43.066,"max":43.068,"avg":43.066},"frame":257,"keyframe":257,"packet":257,"size_kb":257,"size_bytes":263168}],"outputs":[{"index":0,"stream":0,"frame":149,"keyframe":3,"packet":149,"q":-1.0,"size_kb":1467,"size_bytes":1501923,"extradata_size_bytes":69},{"index":0,"stream":1,"frame":149,"keyframe":3,"packet":149,"q":-1.0,"size_kb":4428,"size_bytes":4534612,"extradata_size_bytes":71},{"index":0,"stream":2,"frame":257,"keyframe":256,"packet":256,"size_kb":1,"size_bytes":1046,"extradata_size_bytes":5},{"index":0,"stream":3,"frame":257,"keyframe":256,"packet":256,"size_kb":1,"size_bytes":1046,"extradata_size_bytes":5}],"frame":149,"packet":149,"q":-1.0,"size_kb":5897,"size_bytes":6038627,"time":"0h0m5.96s","speed":4.79,"dup":0,"drop":0}`) parser.Parse([]byte(`ffmpeg.progress:{"inputs":[{"index":0,"stream":0,"framerate":{"min":24.975,"max":24.975,"avg":24.975},"frame":149,"keyframe":3,"packet":149,"size_kb":1467,"size_bytes":1501854},{"index":1,"stream":0,"framerate":{"min":24.975,"max":24.975,"avg":24.975},"frame":149,"keyframe":3,"packet":149,"size_kb":4428,"size_bytes":4534541},{"index":2,"stream":0,"framerate":{"min":43.066,"max":43.068,"avg":43.066},"frame":257,"keyframe":257,"packet":257,"size_kb":257,"size_bytes":263168}],"outputs":[{"index":0,"stream":0,"frame":149,"keyframe":3,"packet":149,"q":-1.0,"size_kb":1467,"size_bytes":1501923,"extradata_size_bytes":69},{"index":0,"stream":1,"frame":149,"keyframe":3,"packet":149,"q":-1.0,"size_kb":4428,"size_bytes":4534612,"extradata_size_bytes":71},{"index":0,"stream":2,"frame":257,"keyframe":256,"packet":256,"size_kb":1,"size_bytes":1046,"extradata_size_bytes":5},{"index":0,"stream":3,"frame":257,"keyframe":256,"packet":256,"size_kb":1,"size_bytes":1046,"extradata_size_bytes":5}],"frame":149,"packet":149,"q":-1.0,"size_kb":5897,"size_bytes":6038627,"time":"0h0m5.96s","speed":4.79,"dup":0,"drop":0}`))
progress := parser.Progress() progress := parser.Progress()
@@ -1132,14 +1132,14 @@ func TestParserPatterns(t *testing.T) {
}, },
}) })
p.Parse("some foobar more") p.Parse([]byte("some foobar more"))
require.Empty(t, p.Report().Matches) require.Empty(t, p.Report().Matches)
p.Parse("foobar some more") p.Parse([]byte("foobar some more"))
require.Equal(t, 1, len(p.Report().Matches)) require.Equal(t, 1, len(p.Report().Matches))
require.Equal(t, "foobar some more", p.Report().Matches[0]) require.Equal(t, "foobar some more", p.Report().Matches[0])
p.Parse("some more foobar") p.Parse([]byte("some more foobar"))
require.Equal(t, 2, len(p.Report().Matches)) require.Equal(t, 2, len(p.Report().Matches))
require.Equal(t, "some more foobar", p.Report().Matches[1]) require.Equal(t, "some more foobar", p.Report().Matches[1])
@@ -1165,3 +1165,25 @@ func TestParserPatternsError(t *testing.T) {
require.Equal(t, 1, len(parser.Report().Matches)) require.Equal(t, 1, len(parser.Report().Matches))
} }
func BenchmarkParserString(b *testing.B) {
parser := New(Config{
LogLines: 100,
})
parser.Parse([]byte(`ffmpeg.inputs:[{"url":"https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk_720.m3u8","format":"hls","index":0,"stream":0,"type":"video","codec":"h264","coder":"h264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":1280,"height":720},{"url":"https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk_1440.m3u8","format":"hls","index":1,"stream":0,"type":"video","codec":"h264","coder":"h264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":2560,"height":1440},{"url":"anullsrc=r=44100:cl=mono","format":"lavfi","index":2,"stream":0,"type":"audio","codec":"pcm_u8","coder":"pcm_u8","bitrate_kbps":352,"duration_sec":0.000000,"language":"und","sampling_hz":44100,"layout":"mono","channels":1}]`))
parser.Parse([]byte(`hls.streammap:{"address":"http://127.0.0.1:8080/memfs/live/%v.m3u8","variants":[{"variant":0,"address":"http://127.0.0.1:8080/memfs/live/0.m3u8","streams":[0,2]},{"variant":1,"address":"http://127.0.0.1:8080/memfs/live/1.m3u8","streams":[1,3]}]}`))
parser.Parse([]byte(`ffmpeg.outputs:[{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":0,"type":"video","codec":"h264","coder":"copy","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":1280,"height":720},{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":1,"type":"video","codec":"h264","coder":"copy","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":25.000000,"pix_fmt":"yuvj420p","width":2560,"height":1440},{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":2,"type":"audio","codec":"aac","coder":"aac","bitrate_kbps":69,"duration_sec":0.000000,"language":"und","sampling_hz":44100,"layout":"mono","channels":1},{"url":"http://127.0.0.1:8080/memfs/live/%v.m3u8","format":"hls","index":0,"stream":3,"type":"audio","codec":"aac","coder":"aac","bitrate_kbps":69,"duration_sec":0.000000,"language":"und","sampling_hz":44100,"layout":"mono","channels":1}]`))
data := [][]byte{
[]byte(`ffmpeg.progress:{"inputs":[{"index":0,"stream":0,"framerate":{"min":24.975,"max":24.975,"avg":24.975},"frame":149,"keyframe":3,"packet":149,"size_kb":1467,"size_bytes":1501854},{"index":1,"stream":0,"framerate":{"min":24.975,"max":24.975,"avg":24.975},"frame":149,"keyframe":3,"packet":149,"size_kb":4428,"size_bytes":4534541},{"index":2,"stream":0,"framerate":{"min":43.066,"max":43.068,"avg":43.066},"frame":257,"keyframe":257,"packet":257,"size_kb":257,"size_bytes":263168}],"outputs":[{"index":0,"stream":0,"frame":149,"keyframe":3,"packet":149,"q":-1.0,"size_kb":1467,"size_bytes":1501923,"extradata_size_bytes":69},{"index":0,"stream":1,"frame":149,"keyframe":3,"packet":149,"q":-1.0,"size_kb":4428,"size_bytes":4534612,"extradata_size_bytes":71},{"index":0,"stream":2,"frame":257,"keyframe":256,"packet":256,"size_kb":1,"size_bytes":1046,"extradata_size_bytes":5},{"index":0,"stream":3,"frame":257,"keyframe":256,"packet":256,"size_kb":1,"size_bytes":1046,"extradata_size_bytes":5}],"frame":149,"packet":149,"q":-1.0,"size_kb":5897,"size_bytes":6038627,"time":"0h0m5.96s","speed":4.79,"dup":0,"drop":0}`),
[]byte(`[https @ 0x557c840d1080] Opening 'https://ch-fra-n16.livespotting.com/vpu/e9slfpe3/z60wzayk_720_100794.ts' for reading`),
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
parser.Parse(data[0])
parser.Parse(data[1])
}
}

7
ffmpeg/probe/prober.go
View File

@@ -1,6 +1,7 @@
package probe package probe
import ( import (
"bytes"
"strings" "strings"
"time" "time"
@@ -54,14 +55,14 @@ func (p *prober) Probe() Probe {
return probe return probe
} }
func (p *prober) Parse(line string) uint64 { func (p *prober) Parse(line []byte) uint64 {
if strings.HasPrefix(line, "avstream.progress:") { if bytes.HasPrefix(line, []byte("avstream.progress:")) {
return 0 return 0
} }
p.data = append(p.data, process.Line{ p.data = append(p.data, process.Line{
Timestamp: time.Now(), Timestamp: time.Now(),
Data: line, Data: string(line),
}) })
return 0 return 0

6
ffmpeg/probe/prober_test.go
View File

@@ -56,7 +56,7 @@ Press [q] to stop, [?] for help`
data := strings.Split(rawdata, "\n") data := strings.Split(rawdata, "\n")
for _, d := range data { for _, d := range data {
prober.Parse(d) prober.Parse([]byte(d))
} }
prober.ResetStats() prober.ResetStats()
@@ -169,8 +169,8 @@ Press [q] to stop, [?] for help`
func TestJSON(t *testing.T) { func TestJSON(t *testing.T) {
prober := New(Config{}).(*prober) prober := New(Config{}).(*prober)
prober.Parse("foobar") prober.Parse([]byte("foobar"))
prober.Parse(`ffmpeg.inputs:[{"url":"https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk.m3u8","format":"playout","index":0,"stream":0,"type":"video","codec":"h264","coder":"h264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":20.666666,"pix_fmt":"yuvj420p","width":1280,"height":720}]`) prober.Parse([]byte(`ffmpeg.inputs:[{"url":"https://cdn.livespotting.com/vpu/e9slfpe3/z60wzayk.m3u8","format":"playout","index":0,"stream":0,"type":"video","codec":"h264","coder":"h264","bitrate_kbps":0,"duration_sec":0.000000,"language":"und","fps":20.666666,"pix_fmt":"yuvj420p","width":1280,"height":720}]`))
prober.ResetStats() prober.ResetStats()

18
process/parser.go
View File

@@ -10,7 +10,7 @@ type Parser interface {
// Parse parses the given line and returns an indicator // Parse parses the given line and returns an indicator
// for progress (e.g. based on the contents of the line, // for progress (e.g. based on the contents of the line,
// or previous line, ...) // or previous line, ...)
Parse(line string) uint64 Parse(line []byte) uint64
// Stop tells the parser that the process stopped and provides // Stop tells the parser that the process stopped and provides
// its exit state. // its exit state.
@@ -50,12 +50,12 @@ func NewNullParser() Parser {
var _ Parser = &nullParser{} var _ Parser = &nullParser{}
func (p *nullParser) Parse(string) uint64 { return 1 } func (p *nullParser) Parse(line []byte) uint64 { return 1 }
func (p *nullParser) Stop(string, Usage) {} func (p *nullParser) Stop(string, Usage) {}
func (p *nullParser) ResetStats() {} func (p *nullParser) ResetStats() {}
func (p *nullParser) ResetLog() {} func (p *nullParser) ResetLog() {}
func (p *nullParser) Log() []Line { return []Line{} } func (p *nullParser) Log() []Line { return []Line{} }
func (p *nullParser) IsRunning() bool { return true } func (p *nullParser) IsRunning() bool { return true }
type bufferParser struct { type bufferParser struct {
log []Line log []Line
@@ -69,10 +69,10 @@ func NewBufferParser() Parser {
var _ Parser = &bufferParser{} var _ Parser = &bufferParser{}
func (p *bufferParser) Parse(line string) uint64 { func (p *bufferParser) Parse(line []byte) uint64 {
p.log = append(p.log, Line{ p.log = append(p.log, Line{
Timestamp: time.Now(), Timestamp: time.Now(),
Data: line, Data: string(line),
}) })
return 1 return 1
} }

33
process/process.go
View File

@@ -167,13 +167,12 @@ type States struct {
// Process represents a ffmpeg process // Process represents a ffmpeg process
type process struct { type process struct {
binary string binary string
args []string args []string
cmd *exec.Cmd cmd *exec.Cmd
pid int32 pid int32
stdout io.ReadCloser stdout io.ReadCloser
lastLine string state struct {
state struct {
state stateType state stateType
time time.Time time time.Time
states States states States
@@ -575,7 +574,7 @@ func (p *process) start() error {
if err != nil { if err != nil {
p.setState(stateFailed) p.setState(stateFailed)
p.parser.Parse(err.Error()) p.parser.Parse([]byte(err.Error()))
p.logger.WithError(err).Error().Log("Command failed") p.logger.WithError(err).Error().Log("Command failed")
p.reconnect(p.delay(stateFailed)) p.reconnect(p.delay(stateFailed))
@@ -587,7 +586,7 @@ func (p *process) start() error {
if err := p.callbacks.onBeforeStart(); err != nil { if err := p.callbacks.onBeforeStart(); err != nil {
p.setState(stateFailed) p.setState(stateFailed)
p.parser.Parse(err.Error()) p.parser.Parse([]byte(err.Error()))
p.logger.WithError(err).Error().Log("Starting failed") p.logger.WithError(err).Error().Log("Starting failed")
p.reconnect(p.delay(stateFailed)) p.reconnect(p.delay(stateFailed))
@@ -599,7 +598,7 @@ func (p *process) start() error {
if err := p.cmd.Start(); err != nil { if err := p.cmd.Start(); err != nil {
p.setState(stateFailed) p.setState(stateFailed)
p.parser.Parse(err.Error()) p.parser.Parse([]byte(err.Error()))
p.logger.WithError(err).Error().Log("Command failed") p.logger.WithError(err).Error().Log("Command failed")
p.reconnect(p.delay(stateFailed)) p.reconnect(p.delay(stateFailed))
@@ -770,7 +769,7 @@ func (p *process) stop(wait bool, reason string) error {
} }
if err != nil { if err != nil {
p.parser.Parse(err.Error()) p.parser.Parse([]byte(err.Error()))
p.debuglogger.WithFields(log.Fields{ p.debuglogger.WithFields(log.Fields{
"state": p.getStateString(), "state": p.getStateString(),
"order": p.getOrder(), "order": p.getOrder(),
@@ -857,7 +856,7 @@ func (p *process) staler(ctx context.Context) {
// may kick in. // may kick in.
func (p *process) reader() { func (p *process) reader() {
scanner := bufio.NewScanner(p.stdout) scanner := bufio.NewScanner(p.stdout)
scanner.Split(scanLine) scanner.Split(scanLines)
// Reset the parser statistics // Reset the parser statistics
p.parser.ResetStats() p.parser.ResetStats()
@@ -868,9 +867,7 @@ func (p *process) reader() {
var n uint64 = 0 var n uint64 = 0
for scanner.Scan() { for scanner.Scan() {
line := scanner.Text() line := scanner.Bytes()
p.lastLine = line
// Parse the output line from ffmpeg // Parse the output line from ffmpeg
n = p.parser.Parse(line) n = p.parser.Parse(line)
@@ -886,12 +883,12 @@ func (p *process) reader() {
if err := scanner.Err(); err != nil { if err := scanner.Err(); err != nil {
p.logger.Debug().WithError(err).Log("") p.logger.Debug().WithError(err).Log("")
p.parser.Parse(err.Error()) p.parser.Parse([]byte(err.Error()))
} }
p.stopReasonLock.Lock() p.stopReasonLock.Lock()
if len(p.stopReason) != 0 { if len(p.stopReason) != 0 {
p.parser.Parse(p.stopReason) p.parser.Parse([]byte(p.stopReason))
p.stopReason = "" p.stopReason = ""
} }
p.stopReasonLock.Unlock() p.stopReasonLock.Unlock()
@@ -1062,7 +1059,7 @@ func (p *process) delay(state stateType) time.Duration {
} }
// scanLine splits the data on \r, \n, or \r\n line endings // scanLine splits the data on \r, \n, or \r\n line endings
func scanLine(data []byte, atEOF bool) (advance int, token []byte, err error) { func scanLines(data []byte, atEOF bool) (advance int, token []byte, err error) {
// Skip leading spaces. // Skip leading spaces.
start := 0 start := 0
for width := 0; start < len(data); start += width { for width := 0; start < len(data); start += width {

63
process/process_test.go
View File

@@ -1,12 +1,15 @@
package process package process
import ( import (
"bufio"
"bytes"
"fmt" "fmt"
"sync" "sync"
"testing" "testing"
"time" "time"
"github.com/datarhei/core/v16/internal/testhelper" "github.com/datarhei/core/v16/internal/testhelper"
"github.com/datarhei/core/v16/math/rand"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
@@ -701,3 +704,63 @@ func TestProcessCallbacksOnBeforeStart(t *testing.T) {
require.Equal(t, 1, len(lines)) require.Equal(t, 1, len(lines))
require.Equal(t, "no, not now", lines[0].Data) require.Equal(t, "no, not now", lines[0].Data)
} }
func BenchmarkScannerText(b *testing.B) {
data := []byte{}
for i := 0; i < 1000; i++ {
line := rand.String(100) + "\n"
data = append(data, []byte(line)...)
}
b.ResetTimer()
lastline := ""
for i := 0; i < b.N; i++ {
r := bytes.NewReader(data)
scanner := bufio.NewScanner(r)
scanner.Split(bufio.ScanLines)
for scanner.Scan() {
line := scanner.Text()
lastline = line
}
err := scanner.Err()
require.NoError(b, err)
}
fmt.Printf("%s\n", lastline)
}
func BenchmarkScannerBytes(b *testing.B) {
data := []byte{}
for i := 0; i < 1000; i++ {
line := rand.String(100) + "\n"
data = append(data, []byte(line)...)
}
b.ResetTimer()
lastline := []byte{}
for i := 0; i < b.N; i++ {
r := bytes.NewReader(data)
scanner := bufio.NewScanner(r)
scanner.Split(bufio.ScanLines)
for scanner.Scan() {
line := scanner.Bytes()
lastline = line
}
err := scanner.Err()
require.NoError(b, err)
}
fmt.Printf("%s\n", lastline)
}