在 Node 中实现“take N”转换流答案

【问题标题】：Implementing a "take N" transform stream in Node在 Node 中实现“take N”转换流
【发布时间】：2020-08-20 18:36:37
【问题描述】：

我有一个可读流 X 和一个可写流 Z。我无法控制 X 将产生多少字节的数据流。我想将不超过 N 字节的数据从 X 流式传输到 Z。我认为一个新的转换流 Y 将传递它接收到的所有数据，直到它接收到的数据量超过其限制，这可能是一个不错的、干净的解决方案。如何实现这样的流？

【问题讨论】：

标签： node.js stream

【解决方案1】：

你需要使用 node 的流 api。

请注意，无休止的阅读器会触发一些错误，如下面的代码所示：

const { Readable, Writable, Transform } = require('stream')

const source = new Readable({
  objectMode: true,
  read () {
    // endless writer
    this.push({ time: new Date() })
  }
})

let items = 0
const transformation = new Transform({
  objectMode: true,
  transform  (data, encoding, done) {
    if (items > 100) {
      done(null)
      this.end() // we must stop this stream
    } else {
      done(null, { ...data, index: items++ })
    }
  }
})

const destination = new Writable({
  objectMode: true,
  write (data, encoding, done) {
    console.log({ data })
    done()
  }
})

source
  .pipe(transformation)
  .on('error', () => { console.log('Error: it will be emitted because the READER will "write" in the transform stream but we closed it') })
  .pipe(destination)
  .on('finish', () => { console.log('All writes are now complete.') })

【讨论】：