Как читать с stdout и, не дожидаясь окончания, перенаправлять вывод?

Question

[Ромзес Панагиотис]

Задача является логическим продолжением Как записывать stdout в буфер и считывать оттуда построчно?

Суть в том, что в https://play.golang.org/p/aryjjfUZEmt идет ожидание окончания вывода. А пока это не произойдет, ничего не отправляется в брокер (в данном примере, для простоты, на экран).
Мне же необходимо не дожидаться и, по мере поступления вывода, отправлять их в брокер.

У меня основная загвоздка с переводами строк, потому что необходимо при получении массива байтов записать его в буфер, а другим читателем подхватывать его и анализировать переводы строк (при помощи scanner).

Добавлено:
Для примера эффекта, вместо cmd := exec.Command("ls", "/etc") (строка 24 из кода выше) взял из https://gobyexample.com/tickers и скомпилировал код для программы, которая выводит в stdout данные не сразу, а постепенно при помошью таймера. В итоге получается, что пока процесс не завершится, в вышеприведенном примере не будет отправки по строкам.

Добавлено:
Обновил код со своей попыткой.

Comments

[Евгений Мамонов]

т.е. задача такая: забрать все строки из буфера, но не больше определённого числа, верно?

[Ромзес Панагиотис]

Евгений Мамонов,
в брокер должно идти не больше X строк за один заход. А в буфер же может поступить иное количество, если я правильно понимаю.

[Ромзес Панагиотис]

По мере выплевывания вывода программы (как бы в "реальном времени") мы должны записывать в буфер, а при анализе его учитывать количество накопившихся строк.

[Ромзес Панагиотис]

Евгений Мамонов, я обновил свой код в вопросе.

[Евгений Мамонов]

думаю как решить задачу :)

Answer 1

[Ромзес Панагиотис]

Решил вопрос при помощи io.Pipe():
https://play.golang.org/p/fGL2uBKuog8

код

package main

import (
	"bufio"
	"fmt"
	"io"
	"log"
	"os"
	"os/exec"
)

var done chan bool

type Executor struct {
	wr io.Writer
}

func (e *Executor) SetStdout(w io.Writer) {
	e.wr = w
}

func (e *Executor) Exec() error {
	cmd := exec.Command("ls", "/etc")
	cmd.Stdout = e.wr
	err := cmd.Run()
	return err
}

func send(lines []string) {
	w := bufio.NewWriter(os.Stdout)
	for i := 0; i < len(lines); i++ {
		w.WriteString(lines[i] + "\n")
	}
	w.Flush()
}

const cnt = 3

func main() {
	// create a pipe
	reader, writer := io.Pipe()

	e := &Executor{}
	e.SetStdout(writer)

	go func() {
		lines := make([]string, 0, cnt)
		scanner := bufio.NewScanner(reader)
		linesCnt := 0
		for scanner.Scan() {
			line := scanner.Text()
			// fmt.Println(`out:`, line)

			lines = append(lines, line)

			linesCnt++
			if linesCnt >= cnt {
				linesCnt = 0
				send(lines)
				lines = lines[:0] // clear
			}
		}

		if err := scanner.Err(); err != nil {
			fmt.Fprintln(os.Stderr, "reading standard input:", err)
		}

		if len(lines) > 0 {
			send(lines)
		}
	}()

	err := e.Exec()
	if err != nil {
		log.Fatal(err)
	}
	log.Println("Exec finished!")
}

Спасибо Евгений Мамонов за наводку с bufio.NewScanner.