如何将 exec 函数的标准输出通过管道传输到另一个读取器？

Question

我正在尝试将标准输出从 mongodump 流式传输到 s3。我已经正确掌握了 S3 任意长度流的语法，但我不明白如何耦合这两个函数。在开始上传到 S3 之前，我不想处理整个 mongodump 命令。这是我目前所拥有的：

dumpCmd := exec.Command("mongodump", "--host", "<host>", "--port", "<port>", "--archive")
dumpCmd.Stdout = os.Stdout


uploader := s3manager.NewUploader(session.New(&aws.Config{Region: aws.String("us-east-1")}))
result, err := uploader.Upload(&s3manager.UploadInput{
    Body:   dumpCmd.Stdout,
    Bucket: aws.String("myBucket"),
    Key:    aws.String("myKey"),
})
if err != nil {
    log.Fatalln("Failed to upload", err)
}

log.Println("Successfully uploaded to", result.Location)

不幸的是，dumpCmd.Stdout 是写入器，而不是读取器，我不知道如何将写入器的输出通过管道传输到读取器。

使用管道后，我现在遇到了一个新错误，这可能属于一个新问题：

dumpCmd := exec.Command("mongodump", "--host", "<host>", "--port", "<port>", "--archive")
body, err := dumpCmd.StdoutPipe()
if err != nil {
    // handle error
}

if err := dumpCmd.Start(); err != nil {
    // handle error
}

uploader := s3manager.NewUploader(session.New(&aws.Config{Region: aws.String("us-east-1")}))
result, err := uploader.Upload(&s3manager.UploadInput{
    Body:   body,
    Bucket: aws.String("myBucket"),
    Key:    aws.String("myKey"),
})
if err != nil {
    log.Fatalln("Failed to upload", err)
}

if err := dumpCmd.Wait(); err != nil {
    // handle error
}

log.Println("Successfully uploaded to", result.Location)

错误：

2016/03/10 12:39:18 Failed to upload MultipartUpload: upload multipart failed
    upload id: QOWW4jBHH4PKjs1Tloc8dlCTtFN94vDHIJIWJChsrjxLZggScZbRUhM4FU9V.xOnIg9uYnBWqOA1x1xqStfA1p8vdAOHNyUp4gOO5b1gbuXvUitQyLdfFhKg9MnyxsV1
caused by: RequestError: send request failed
caused by: Put https://myBucket/myKey?partNumber=1&uploadId=QOWW4jBHH4PKjs1Tloc8dlCTtFN94vDHIJIWJChsrjxLZggScZbRUhM4FU9V.xOnIg9uYnBWqOA1x1xqStfA1p8vdAOHNyUp4gOO5b1gbuXvUitQyLdfFhKg9MnyxsV1: read |0: illegal seek
exit status 1

Answer 1

使用pipe 将命令的输出连接到上传的输入。

这里的棘手问题是上传者seeks on the body，但是管道不支持seek。为了避免这种情况，在管道周围创建了一个包装器，以对上传者隐藏 Seek 方法。

dumpCmd := exec.Command("mongodump", "--host", "<host>", "--port", "<port>", "--archive")
body, err := dumpCmd.StdoutPipe()
if err != nil {
    // handle error
}

if err := dumpCmd.Start(); err != nil {
    // handle error
}

// Wrap the pipe to hide the seek methods from the uploader
bodyWrap := struct {
    io.Reader
}{body}

uploader := s3manager.NewUploader(session.New(&aws.Config{Region: aws.String("us-east-1")}))
result, err := uploader.Upload(&s3manager.UploadInput{
    Body:   bodyWrap,
    Bucket: aws.String("net-openwhere-mongodb-snapshots-dev"),
    Key:    aws.String("myKey"),
})
if err != nil {
    log.Fatalln("Failed to upload", err)
}

if err := dumpCmd.Wait(); err != nil {
   // handle error
}

log.Println("Successfully uploaded to", result.Location)

Answer 2

可以使用io.Copy方法（godoc），定义如下：

函数复制

func Copy(dst Writer, src Reader)（写入int64，err错误）

将副本从 src 复制到 dst 直到在 src 上达到 EOF 或 发生错误。它返回复制的字节数和第一个 复制时遇到错误（如果有）。

成功的复制返回 err == nil，而不是 err == EOF。因为复制是 定义为从 src 读取直到 EOF，它不处理来自 Read 的 EOF 作为要报告的错误。

如果 src 实现了 WriterTo 接口，则复制由 调用 src.WriteTo(dst)。否则，如果 dst 实现 ReaderFrom 接口，拷贝是通过调用dst.ReadFrom(src)实现的。