stream: Add option to `Readable.take` operator to not close the stream

[rluvaton]

What is the problem this feature will solve?

I will be able to do this:

const csvParsedStream = fs
	.createReadStream('file.csv')
	.compose(csvParse({ columns: false }));

const [columns] = await csvParsedStream
	.take(1)
	.toArray();

// This will now be empty and no data as take already consumed the stream
const parsed = await csvParsedStream
   .map((row) => parseRowByColumns(row, columns))
   .toArray();

Another example (I know I can use [first, ...rest] this is just an example):

const a = Readable.from([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
const [first] = await a.take(1).toArray();

console.log(first);
// [1]

const rest = await a.toArray();
console.log(rest)
// []

What is the feature you are proposing to solve the problem?

Adding closeStream option to the take operator that with default value true that I could disable closing the stream

const csvParsedStream = fs
	.createReadStream('file.csv')
	.compose(csvParse({ columns: false }));

const [columns] = await csvParsedStream
	.take(1, { closeStream: false })  // Right now this would close the stream, but we give it an option to not
	.toArray();

const parsed = await csvParsedStream.map((row) => parseRowByColumns(row, columns)).toArray();

What alternatives have you considered?

Get the first value from stream as async itarator and rest

let columns;
for await (const c of csvParsedStream.iterator<string[]>({ destroyOnReturn: false })) {
	columns = c;

	break;
}

const parsed = await csvParsedStream
    .map((row) => parseRowByColumns(row, columns))
    .toArray();

[rluvaton]

I'll implement this if we think it's a good addition.

Cc. @benjamingr @mcollina @ronag

[benjamingr]

What does a .NET IEnumerable do for example?

[rluvaton]

Take multiple times just returns the same data:

IEnumerable<int> myMethod() {
    yield return 1;
    yield return 2;
    yield return 3;
}

IEnumerable<int> values = myMethod();

foreach (int value in values.Take(1)) {
    Console.WriteLine(value); // 1
}

foreach (int value in values.Take(1)) {
    Console.WriteLine(value); // 1
}

[benjamingr]

@Linkgoron @ronag wdyt?

[ronag]

I think Ienumerable is not relevant. It's Observable operators (async) we should look at.

[rluvaton]

I think Ienumerable is not relevant. It's Observable operators (async) we should look at.

Same in rxjs:

import { of, take, toArray, firstValueFrom } from 'rxjs';

async function run() {
    const intervalCount = of(0, 1, 2);
    const data = await firstValueFrom(
        intervalCount.pipe(
            take(1),
            toArray()
        )
    );
    console.log(data)
    const data2 = await firstValueFrom(
        intervalCount.pipe(
            take(1),
            toArray()
        )
    );
    console.log(data2)
}

run()

// logs:
// [0]
// [0]

[rluvaton]

Although if you do this:

It logs 0, 1

import { of, take, toArray, firstValueFrom, Subject } from 'rxjs';

const sub = new Subject();

async function run() {
    const data = await firstValueFrom(
        sub.pipe(
            take(1),
            toArray()
        )
    );
    console.log(data)
    const data2 = await firstValueFrom(
        sub.pipe(
            take(1),
            toArray()
        )
    );
    console.log(data2)
}

let i = 0;
run()
setInterval(() => sub.next(i++), 100)

Basically, RxJs pipes don't change the original producer and most of the time don't wait for subscribers to start emitting, and no backpressure from what I know...

[benjamingr]

The issue is that in Rx you compose the observable (operation) and not the observer.

In node the operators are on the iterator and not the iterable.

[rluvaton]

Any bottom line? @ronag WDYT?

[rluvaton]

This is how I think it should be implemented, any thoughts?:

• stream: allow stream to stay open after take #47023