stream: Readable.concat util method

[tugrul]

There are many npm modules for this functionality and I think it is a common requirements because I see 452k weekly downloads of multistream npm package.

[VoltrexKeyva]

Suggested change

	* `streams` {Readable[]|Iterable[]|AsyncIterable[]} List of readable streams or iterator of readable stream generator
	* `streams` {Readable[]|Iterable[]|AsyncIterable[]} List of readable streams or
	iterator of readable stream generator.

Every line must be 80 characters or less in length and this line should end with a period.

[VoltrexKeyva]

Suggested change

	By default, `Readable.concat()` will set `options.objectMode` to `false`, unless
	this is explicitly opted out by setting `options.objectMode` to `true`.
	By default, `Readable.concat()` will set `options.objectMode` to `false`,
	unless this is explicitly opted out by setting `options.objectMode` to `true`.

Every line must be 80 characters or less in length.

[VoltrexKeyva]

Suggested change

	A utility method to concatenate multiple readable streams to single one. The main readable stream will consume the
	list of readable streams in order
	A utility method to concatenate multiple readable streams to single one. The
	main readable stream will consume the list of readable streams in order.

Every line must be 80 characters or less in length and this line must end with a period.

[VoltrexKeyva]

Suggested change

	return Readable.from(streamSeriesIterator(streams), {objectMode: false, ...opts});
	return Readable.from(streamSeriesIterator(streams),
	{ objectMode: false, ...opts });

Lint.

[aduh95]

To make Readable.concat.length equal to 1:

Suggested change

	Readable.concat = function(streams, opts) {
	Readable.concat = function(streams, opts = undefined) {

[aduh95]

Would that have the same effect?

Suggested change

	for await (const chunk of stream) {
	yield chunk;
	}
	yield* stream;

[tugrul]

This works fine and looks better. Thanks

[tniessen]

cc @nodejs/streams

[mcollina]

Thanks for the PR!

Would you mind to add a few unit test for this feature?

I think if one of the "waiting" streams errors all should be destroyed, similar to how pipeline works.

[ronag]

IbjectMode should be true

[ronag]

This doesn’t support old streams. See pipeline for how we fix that.

[tugrul]

@mcollina I quickly lookup and only we need destroy remaining readable streams if it is an Array when error occurs on previous readable readings. This control not necessary for iterables because stream will initialize lazy. We do not have control of writable stream in this scenario. Following solutions can solve the problem.

async function * streamSeriesIterator(streams) {
  let index = 0;

  try {
    for await (const stream of streams) {
      index++;
      yield* stream;
    }
  } finally {
    if (streams instanceof Array) {
      streams.slice(index).forEach(stream => stream.destroy());
    }
  }
}

or

async function * streamSeriesIterator(streams) {
  let index = 0;

  try {
    for await (const stream of streams) {
      index++;
      yield* stream;
    }
  } catch (err) {
    if (streams instanceof Array) {
      streams.slice(index).forEach(stream => stream.destroy(err));
    }
    
    throw err;
  }
}

[jasnell]

As an alternative suggestion... consider adding this utility to the streams/consumers module and making it work with any stream.Readable, ReadableStream, and async iterable...

import { concat, arrayBuffer } from 'node:stream/consumers';

const readable1 = new ReadableStream();
const readable2 = new ReadableStream();
const readable3 = new stream.Readable();

for await (const chunk of concat(readable1, readable2, readable3)) {
  console.log(chunk);
}

// Let's say you want all of the contents in a single ArrayBuffer...

await arrayBuffer(concat(readable1, readable2, readable3));

The return value would be just an async iterator. If someone wanted a stream.Readable from the results, it would be as simple as... stream.Readable.from(concat(readable1, readable2, readable3)) ... getting a ReadableStream from it is a bit more complicated but also straightforward...

const readable = new ReadableStream({
  async pull(c) {
    for await (const chunk of concat(readable1, readable2, readable3))
      c.enqueue(chunk);
    c.close();
  }
});

[mcollina]

@tugrul unfortunately #40075 (comment) does not work for node streams. They can error at any time even if they are not being consumed, you need to add on('error') handlers.

[Ethan-Arrowood]

Hello, I'm doing some backlog hygiene on all PRs related to networking and streaming modules. There hasn't been any activity here for quite a while so I'm going to mark it stalled. It will auto-close in ~30 days. This is not a rejection; if you've like to pick it back up, rebase onto latest main and give me a ping.

Thank you!

[github-actions]

This issue/PR was marked as stalled, it will be automatically closed in 30 days. If it should remain open, please leave a comment explaining why it should remain open.

[codecov]

Codecov Report

❌ Patch coverage is 93.73434% with 25 lines in your changes missing coverage. Please review.
✅ Project coverage is 90.14%. Comparing base (4a5eb1c) to head (d019ad6).
⚠️ Report is 9 commits behind head on main.

Files with missing lines	Patch %	Lines
lib/stream/consumers/concat.js	93.71%	25 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #40075      +/-   ##
==========================================
- Coverage   90.15%   90.14%   -0.02%     
==========================================
  Files         744      745       +1     
  Lines      242517   242917     +400     
  Branches    45688    45782      +94     
==========================================
+ Hits       218642   218968     +326     
- Misses      15358    15431      +73     
- Partials     8517     8518       +1     

Files with missing lines	Coverage Δ
lib/stream/consumers.js	100.00% <100.00%> (ø)
lib/stream/consumers/concat.js	93.71% <93.71%> (ø)

... and 31 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.