连接两个(或n)个流

Question

• 2个流:

  鉴于可读流 stream1和stream2,什么是地道(简洁)的方式来获取包含流stream1和stream2级联？

  我不能这样做stream1.pipe(outStream); stream2.pipe(outStream),因为那时流内容混杂在一起.

• n个流:

  给定一个发出不确定数量的流的EventEmitter,例如

  eventEmitter.emit('stream', stream1)
  eventEmitter.emit('stream', stream2)
  eventEmitter.emit('stream', stream3)
  ...
  eventEmitter.emit('end')
  

  什么是一种惯用(简洁)的方式来获得所有流连接在一起的流？

Answer 1

现在可以使用异步迭代器轻松完成此操作

async function* concatStreams(readables) {
  for (const readable of readables) {
    for await (const chunk of readable) { yield chunk }
  }
} 

你可以像这样使用它

const fs = require('fs')
const stream = require('stream')

const files = ['file1.txt', 'file2.txt', 'file3.txt'] 
const iterable = await concatStreams(files.map(f => fs.createReadStream(f)))

// convert the async iterable to a readable stream
const mergedStream = stream.Readable.from(iterable)

有关异步迭代器的更多信息：https ://2ality.com/2019/11/nodejs-streams-async-iteration.html

Answer 2

该合并的流包会连接流.自述文件中的示例:

var CombinedStream = require('combined-stream');
var fs = require('fs');

var combinedStream = CombinedStream.create();
combinedStream.append(fs.createReadStream('file1.txt'));
combinedStream.append(fs.createReadStream('file2.txt'));

combinedStream.pipe(fs.createWriteStream('combined.txt'));

我相信你必须立刻附加所有流.如果队列为空,则combinedStream自动结束.见问题#5.

该流流库是一个具有明确的替代.end,但它更受欢迎,想必没有得到很好的测试.它使用Node 0.10的streams2 API(参见本讨论).

Answer 3

在nodejs中,一个简单的reduce操作应该没问题!

const {PassThrough} = require('stream')

let joined = [s0, s1, s2, ...sN].reduce((pt, s, i, a) => {
  s.pipe(pt, {end: false})
  s.once('end', () => a.every(s => s.ended) && pt.emit('end'))
  return pt
}, new PassThrough())

干杯;)

Answer 4

这可以用vanilla nodejs来完成

import { PassThrough } from 'stream'
const merge = (...streams) => {
    let pass = new PassThrough()
    let waiting = streams.length
    for (let stream of streams) {
        pass = stream.pipe(pass, {end: false})
        stream.once('end', () => --waiting === 0 && pass.emit('end'))
    }
    return pass
}

Answer 5

在 vanilla nodejs 中使用 ECMA 15+ 并结合Ivo和Feng的良好答案。

该类PassThrough是一个普通的Transform流，不会以任何方式修改该流。

const { PassThrough } = require('stream');

const concatStreams = (streamArray, streamCounter = streamArray.length) => streamArray
  .reduce((mergedStream, stream) => {
    // pipe each stream of the array into the merged stream
    // prevent the automated 'end' event from firing
    mergedStream = stream.pipe(mergedStream, { end: false });
    // rewrite the 'end' event handler
    // Every time one of the stream ends, the counter is decremented.
    // Once the counter reaches 0, the mergedstream can emit its 'end' event.
    stream.once('end', () => --streamCounter === 0 && mergedStream.emit('end'));
    return mergedStream;
  }, new PassThrough());

可以这样使用：

const mergedStreams = concatStreams([stream1, stream2, stream3]);