Как записывать stdout в буфер и считывать оттуда построчно?

Question

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

Дано:
сторонний SDK некоторой библиотеки на Go, которая запускает внешний процесс. В ней есть возможность записывать stdout, получаемый при выводе оного процесса в виде многострочного текста.
SetStdout(writer io.Writer)

Мне необходимо реализовать функционал, который позволит отправлять в брокер сообщения по 20 строк, которые были записаны при помощи того writer. И, таким образом, отправлять сообщения в брокер, пока весь вывод не закончится (стриминг).
Насколько я понимаю, мне нужно реализовать интерфейс io.Writer, записывая в какой-то буфер, а в другой функции, в горутине, считывать оттуда построчно и по мере наполнения буфера из 20 строк, отправлять их в брокер.

Надеюсь, объяснил достаточно понятно.

Подскажите на примере простого кода как это осуществить.

Answer 1

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

Вы всё верно понимаете, горутину можно не использовать.

Вы можете сделать что-то типа такого

работающий вариант

package main

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

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
}

type BufferedBrokerWriter struct {
    buffer  bytes.Buffer
    scanner *bufio.Scanner
}

func NewBufferedBrokerWriter() *BufferedBrokerWriter {
    bw := &BufferedBrokerWriter{}
    bw.scanner = bufio.NewScanner(&bw.buffer)
    return bw
}

func (bw *BufferedBrokerWriter) Write(p []byte) (int, error) {
    return bw.buffer.Write(p)
}

func (bw *BufferedBrokerWriter) ReadLines(cnt uint) ([]string, error) {
    lines := make([]string, 0, cnt)
    linesCnt := uint(0)

    for bw.scanner.Scan() {
        line := bw.scanner.Text()
        err := bw.scanner.Err()
        if err != nil || line == `` {
            return lines, err
        }

        lines = append(lines, line)
        linesCnt++
        if linesCnt >= cnt {
            return lines, nil
        }
    }

    return nil, io.EOF
}

func main() {
    // подключаетесь к вашему брокеру сообщений,

    bw := NewBufferedBrokerWriter()

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

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

    for {
        lines, err := bw.ReadLines(2)
        if err == io.EOF {
            break
        }
        fmt.Printf("Lines:\n%+v\n\n", lines)
    }
}

если возникнут сложности с реализацией - напишите, помогу

Comments

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

Спасибо за ответ. У меня как раз загвоздка с подсчетом строк в массиве байтов. Поэтому я подумал, что лучше воспользоваться готовым функционалом, типа
bytes, err := reader.ReadBytes('\n')
Именно поэтому подумал, что должен быть один писатель (реализация io.Writer), а другой - читатель (асинхронный?).
Вот поэтому пытаюсь сообразить как это сделать.

[uvelichitel]

type BufferedBrokerWriter struct {
    io.Writer   //А зачем  вот это?
    buffer bytes.Buffer 
}

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

А функция SetStdout(writer io.Writer) точно io.Writer принимает?
Там было бы логичнее io.Reader принимать, можете скинуть название библиотеки, чтобы я посмотрел?

Если там таки именно io.Writer, можно, как вариант сделать что-то типа такого

package main

import (
    "bytes"
    "io"
)

type BufferedBrokerWriter struct {
    buffer bytes.Buffer
}

func NewBufferedBrokerWriter() *BufferedBrokerWriter {
    return &BufferedBrokerWriter{}
}

func (bw *BufferedBrokerWriter) Write(p []byte) (n int, err error) {
    return bw.buffer.Write(p)
}

func (bw *BufferedBrokerWriter) Read(p []byte) (n int, err error) {
    return bw.buffer.Read(p)
}

func main() {
    // подключаетесь к вашему брокеру сообщений,

    bw := NewBufferedBrokerWriter()
    SetStdout(bw)

    scanner := bufio.NewScanner(bw)
    linesCnt := 0
    for scanner.Scan() {
        fmt.Println(`line:`, scanner.Text()) // для примера (чтобы вы увидели, что данные приходят корректно)
        linesCnt++
        if linesCnt >= 20 {
            // пишите в брокер
            linesCnt = 0
        }
    }
}

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

uvelichitel, ой, добавил по привычке :) обычно это имеет смысл при описании интерфейсов, т.е. когда не struct, а interface

type BufferedBrokerWriter interface {
    io.Writer
}

// иначе нужно будет писать вот так
type BufferedBrokerWriter interface {
    Write(p []byte) (n int, err error)
}

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

Евгений Мамонов, , вот ссылка на документацию.

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

Получилось?
Я часа 4 буду не на связи, если не получится реализовать - как вернусь домой, помогу.

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

Евгений Мамонов, не могу так быстро. Там один только вывод результата занимает более 5 минут. Сейчас попробую воспользоваться примером с NewScanner.

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

Евгений Мамонов, не хочу задерживать. Позже отпишусь о результатах.

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

Евгений Мамонов, вот, родил такой пример:
https://play.golang.org/p/9bklxBHgvS3
Все равно в буфере пусто, то ли не то отправляю.

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

Там в буфере пусто потому, что Scan уже вычитал от туда данные, сейчас поправлю ваш пример.

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

Я немного изменил схему, работающий вариант тут https://play.golang.org/p/9lFvdiAWXt2
Посмотрите пожалуйста.
С таким вариантом вам точно будет удобнее работать :)

Думаю если бы у меня было понимание задачи в целом - скорее всего решение было бы значительно проще :)

Обновил текст ответа.
Напишите пожалуйста всё ли у вас получилось?

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

Евгений Мамонов, благодарю за помощь.
Не могли бы вы помочь модифицировать код так, чтобы можно было, не дожидаясь окончания полного вывода программы, одновременно отправлять строки в брокер?
Я создал новый вопрос:
Как читать с stdout и, не дожидаясь окончания, перенаправлять вывод?