node.js - Nodejs 流 API 和 AWS S3

Question

我对 NodeJS 流 API 感到非常困惑。我正在尝试建立一个管道，其中有几个相互连接的流。它看起来像这样：

S3ReadStream -> ParseTransform -> PassThrough -> CustomFilter -> Serializer -> PassThrough

我挣扎的是错误/完成处理。通常我尝试使用stream.pipeline()而不是readable.pipe(writable).

例如，当我查看 S3 node.js SDK 的源代码时，.upload我发现它managed_upload.js以这种方式处理事件：error导致清理、readable导致耗尽和end导致提交。这对我来说看起来很直观。 https://github.com/aws/aws-sdk-js/blob/v2.391.0/lib/s3/managed_upload.js#L185

问题是，当我使用.pipeline并error发生时，ParseStream错误不会传播到其他流（destroy 被调用而没有错误导致 plain close）。

在这里可以看出opts.error从未设置： https ://github.com/nodejs/node/blob/v10.15.3/lib/internal/streams/end-of-stream.js#L78

虽然在这里可以看出，该销毁器总是被调用而没有错误 https://github.com/nodejs/node/blob/v10.15.3/lib/internal/streams/pipeline.js#L81

更重要的是，.pipeline似乎没有调用end最后一个可写流，如下面的代码所示：

const { PassThrough, pipeline } = require('stream')
const a = new PassThrough().on('end', console.log.bind(console, 'a'))
const b = new PassThrough().on('end', console.log.bind(console, 'b'))
const c = new PassThrough().on('end', console.log.bind(console, 'c'))
pipeline(a, b, c, () => {})
a.end()

可能只有我一个人，但是在阅读了有关它的官方文档后，这与我所经历的行为或我在代码中看到的行为不符。

是否有一种可靠且简单的方法如何通过管道传播错误，即每当管道中的任何地方发生错误时，所有流都会出现错误，从而尝试“清理”或“恢复”？是否有一个库可以构建不包含大量隐藏陷阱的“直观”管道，或者我是否需要通过仔细处理所有事件来照顾每个流？

标签： node.jsamazon-s3node-streams