stream: add map method to Readable:

[benjamingr]

This is very much work in progress when meeting @ronag at NodeTLV.

This is based on the ongoing standards work in https://github.com/tc39/proposal-iterator-helpers in order to make Node.js more compatible with the language in the future.

In addition, people often really just want to map a stream and this provides a simpler API than compose or pipeline for simple cases.

This is still missing docs, I want to bikeshed the API and see we have consensus first.

[benjamingr]

This is still missing:

• docs (I want this to be experimental, but this bit is straightforward).
• consensus regarding whether it should be an instance methods (I like better) or static (may be more compatible possibly).
• consensus regarding its inclusion.

In terms of API - I think this is what people are usually looking for when transforming a stream.

Also cc @mcollina @nodejs/streams to make sure this isn't "under the radar".

[mcollina]

I think it's better to lift the implementation of mapIterator in https://github.com/mcollina/hwp/blob/main/index.js where also @ronag contributed.

[benjamingr]

I don't understand the need for the double queueing here but I am happy to adjust the implementation. The actual API sounds good?

[mcollina]

I don't understand the need for the double queueing here but I am happy to adjust the implementation. The actual API sounds good?

The implementation in this PR processes chunks one by one, however this is problematic given it's an async function. If we do not support some level of parallel processing it's not going to be really useful because it would be too slow.

[ronag]

Should we use hwp map?

[benjamingr]

The implementation in this PR processes chunks one by one, however this is problematic given it's an async function. If we do not support some level of parallel processing it's not going to be really useful because it would be too slow.

That sounds reasonable. So we agree on the API itself but it would be better if it mapd concurrently.

I'm fine with hwp map as that implementation (can adjust the PR + add tests) :)

I think we should slightly adjust the behavior:

• Return a stream and not an async iterator (or results array)
• Name the hwm parameter something more indicative than n (like concurrency or maxConcurrency)?

WDYT?

[jasnell]

That sounds reasonable. So we agree on the API itself but it would be better if it mapd concurrently.

I think concurrency should be an option, not necessarily the default given that it would introduce non-deterministic ordering of the results.

Options that would appear to make immediate sense here include:

• concurrency
• signal (AbortSignal)

const ac = new AbortController();
Readable.from([1, 2, 3, 4, 5]).map((x) => x + x, { concurrency: 5, signal: ac.signal });

[mcollina]

I think concurrency should be an option, not necessarily the default given that it would introduce non-deterministic ordering of the results.

That is incorrect. The implementation I recommended keep the ordering.

[ronag]

@benjamingr I did some work on the PR. I hope you don't mind.

• Added concurrency and highWaterMark options
• Output is in order
• Bypasses a lot of intermediate buffering by creating a Readable directly

[benjamingr]

@ronag I don't, feel free, I am still in the conference aftermath here ^^

I'll try and push some docs :)

[ronag]

Suggested change

	}
	}
	stream.destroy(err);

Should destroy back propagate to source stream?

[benjamingr]

@devsnek does .return propagate in proposal-iterator-helpers?

That is, in that proposal:

const iter = getIterator();
const mapped = iter.map(x => x * 2);
mapped.return(); // is `.return` called on iter?

[benjamingr]

@ronag did you resolve because you checked with that? Because I wanna make sure we align

[ronag]

I resolved it because I added a destroy. It was before I saw your comment.