Правильный способ фильтрации потока байтов в Голанге?

Question

Я хочу отфильтровать STDOUT из такой команды, что я сохраняю только первую и последнюю строку любого смежного блока строк с завершенными \ r (в основном игнорируя индикаторы прогресса).

Вот моя попытка (ориг код делает больше, это упрощенная версия, но в основном фильтрация должна произойти вход приходит, а не в конце):

package main 

import (
    "bytes" 
    "fmt" 
    "os/exec" 
) 

var cr = []byte("\r") 
var lf = []byte("\n") 

func main() { 
    input1 := []byte("a\nb\n\nprogress 98%\r") 
    input2 := []byte("progress 99%\r") 
    input3 := []byte("progress 100%\r") 
    input4 := []byte("\n\nc\n") 

    var stream []byte 
    stream = append(stream, input1...) 
    stream = append(stream, input2...) 
    stream = append(stream, input3...) 
    stream = append(stream, input4...) 

    fmt.Printf("stream:\n%s\n", stream) 

    streamer := &myFilter{} 
    streamer.Write(input1) 
    streamer.Write(input2) 
    streamer.Write(input3) 
    streamer.Write(input4) 
    final := streamer.Bytes() 

    fmt.Printf("streamer:\n%s\n\n", final) 

    cmd := exec.Command("bash", "-c", "perl -e '$|++; print qq[a\nb\n\nprogress: 98%\r]; for (99..100) { print qq[progess: $_%\r]; sleep(1); } print qq[\n\nc\n]'") 
    cmd.Stdout = &myFilter{} 
    cmd.Start() 
    cmd.Wait() 
    fromCmd := cmd.Stdout.(*myFilter).Bytes() 

    fmt.Printf("fromCmd:\n%s\n", fromCmd) 
} 

type myFilter struct { 
    partialLine []byte 
    storage  []byte 
} 

func (w *myFilter) Write(p []byte) (n int, err error) { 
    // in order to filter out all but the first and last line of a set of \r 
    // terminated lines (a progress bar), we need to collect whole \n terminated 
    // lines 
    lines := bytes.SplitAfter(p, lf) 

    if len(w.partialLine) > 0 || (len(lines) == 1 && !bytes.HasSuffix(p, lf)) { 
     w.partialLine = append(w.partialLine, lines[0]...) 

     partialComplete := false 
     if len(lines) > 1 { 
      lines = lines[1:] 
      partialComplete = true 

     } else { 
      lines = nil 
      if bytes.HasSuffix(p, lf) { 
       partialComplete = true 
      } 
     } 

     if partialComplete { 
      w.filterCR(w.partialLine) 
      w.partialLine = nil 
     } 
    } 

    lastLineIndex := len(lines) - 1 
    if lastLineIndex > -1 && !bytes.HasSuffix(p, lf) { 
     w.partialLine, lines = lines[lastLineIndex], lines[:lastLineIndex] 
    } 

    for _, line := range lines { 
     w.filterCR(line) 
    } 

    return len(p), nil 
} 

func (w *myFilter) filterCR(p []byte) { 
    if bytes.Contains(p, cr) { 
     lines := bytes.Split(p, cr) 
     w.store(lines[0]) 
     w.store(lf) 

     if len(lines) > 2 { 
      w.store(lines[len(lines)-2]) 
      w.store(lf) 
     } 
    } else { 
     w.store(p) 
    } 
} 

func (w *myFilter) store(p []byte) { 
    w.storage = append(w.storage, p...) 
} 

func (w *myFilter) Bytes() []byte { 
    if len(w.partialLine) > 0 { 
     w.filterCR(w.partialLine) 
    } 
    return w.storage 
} 

Мой вывод:

stream: 
a 
b 

progress 100% 

c 

streamer: 
a 
b 

progress 98% 
progress 100% 

c 


fromCmd: 
a 
b 

ss: 100% 
progess: 100% 

c 

Что я хочу, это результат, который вы видите из «fromCmd», чтобы соответствовать выходному сигналу, полученному от «стримера».

Что я делаю неправильно, почему мой фактический результат кажется «поврежденным», почему реальный запуск команды ведет себя по-разному с моим тестом «стримера» и что лучший способ фильтровать STDOUT?

Answer 1

Ваш неполный линейный алгоритм неверен для всех входов.

Вы можете заменить myFilter с bufio.Scanner, который будет обрабатывать буферизацию частичной линии правильно для вас, и []byte или bytes.Buffer аккумулировать выход.

var out bytes.Buffer 
scanner := bufio.NewScanner(stdout) 
for scanner.Scan() { 
    p := scanner.Bytes() 
    lines := bytes.Split(p, cr) 
    out.Write(lines[0]) 
    out.Write(lf) 
    if len(lines) > 1 { 
     out.Write(lines[len(lines)-1]) 
     out.Write(lf) 
    } 
}