通过 SSH 流式传输 Stdout 和 Stderr，操作流，然后打印到本地 Stdout 和 Stderr

Question

我正在远程机器上通过 SSH 执行一系列操作，我正在流式传输其 stdout 和 stderr，然后由写入器使用它，写入本地 stdout 和 stderr 以及字节缓冲区。

就在编写器使用它之前，我想对其执行一系列字符串操作，然后写入我的屏幕和缓冲区。到目前为止，一切都很好而且花花公子。

我的问题是现在它不再是流了，它挂起，然后将整个 glob 输出为一个块。我希望它是实时的，所以我将频道放在我的 goroutine 中，但没有任何改进。以下是我的功能，如果您能找出原因，或者可能是实现此目的的更好方法，请告诉我。

//发送

func handleStdStream(filters []string, replaceFilters map[string]string, pipe io.Reader, readers chan io.Reader) {
        if filters != nil {
                // filters exist
                // read first 8 bytes
                res := readPipe(8, pipe)

                // get each line from the resulting streamed output
                for _, str := range strings.Split(res, "\n") {
                        if str != "" {
                                out := lineFilterAndReplace(str, filters, replaceFilters)

                                // instantiate an io.Reader obj from the given string
                                outReader := strings.NewReader(out)

                                readers <- outReader
                        }
                }
        } else {
                // filters dont exist
                if len(replaceFilters) > 0 {
                        res := readPipe(8, pipe)

                        for _, str := range strings.Split(res, "\n") {
                                if str != "" {
                                        out := lineReplace(str, replaceFilters)

                                        // instantiate an io.Reader obj from the given string
                                        outReader := strings.NewReader(out)

                                        readers <- outReader
                                }
                        }
                } else {
                        readers <- pipe
                }
        }
}

// 接收

    outReaders := make(chan io.Reader)

    go handleStdStream(outFilters, replaceFilters, stdoutIn, outReaders)

    go func() {
            for {
                    pipe := <-outReaders

                    _, errStdout = io.Copy(outWriter, pipe)
            }

            // _, errStdout = io.Copy(outWriter, stdoutIn)
    }()

Answer 1

我认为您不需要通道或 goroutine 来完成此操作。 Writer 和 Reader 接口已经在流式传输；你连续从Reader 啜饮字节，直到遇到EOF 或错误，然后连续将字节传递给Writer，直到完成或出现错误。就其本身而言，处理流不需要任何并发性，因此在单个 goroutine 中顺序执行此操作非常合适。

您不应忽略错误返回。如果函数或方法返回错误值，则需要检查它。对于 I/O，您通常需要在 Reader 返回错误时停止读取，并且通常需要在 Writer 返回错误时停止写入。对于Reader，您还必须检查特殊的“错误”值io.EOF。

我认为使用 bufio 包中的 Scanner 比尝试自己进行缓冲/拆分要好。默认情况下，Scanner 在换行符处拆分输入（Unix 样式的 LF 或 DOS 样式的 CRLF）。它还消除了检查io.EOF 的需要，前提是您只通过Scanner 与Reader 交互。

考虑以下版本的handleStdStream：

func handleStdStream(filters []string, replaceFilters map[string]string, pipe io.Reader, w io.Writer) error {
    scanner := bufio.NewScanner(pipe)
    for scanner.Scan() {
        str := scanner.Text()
        if str == "" {
            continue
        }
        out := ""
        if len(filters) != 0 {
            out = lineFilterAndReplace(str, filters, replaceFilters)
        } else {
            out = lineReplace(str, replaceFilters)
        }
        if _, err := w.Write([]byte(out)); err != nil {
            return err
        }
    }
    if err := scanner.Err(); err != nil {
        return err
    }
    return nil
}

你会这样使用它：

err := handleStdStream(filters, replaceFilters, pipe, outWriter)
if err != nil {
    // do something, like printing the error to a log or stderr
}