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

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();

@benjamingr

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

Cc. @benjamingr @mcollina @ronag

What does a .NET IEnumerable do for example?

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
}

@Linkgoron

@Linkgoron @ronag wdyt?

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

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]

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...

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.

@ronag

Any bottom line? @ronag WDYT?

stream: allow stream to stay open after take

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

stream: allow stream to stay open after take #47023

I still think we don't fundamentally understand this well enough. Our iterator helpers are on the iterator (like the tc39 proposal) and not on the iterables so it's hard to come up with good analogies regarding whether or not .take should destroy the stream or not.

On one hand the proposal spec clearly states the underlying iterator should be closedd. On the other hand in models based on iterables and not iterators this wouldn't have been a problem.

We should bring this up with TC39 too.

this is why by default it will close, having an option to not close is a problem?

There has been no activity on this feature request for 5 months. To help maintain relevant open issues, please add the never-stale Issues and PRs exempt from automated stale handling. label or close this issue if it should be closed. If not, the issue will be automatically closed 6 months after the last non-automated comment.
For more information on how the project manages feature requests, please consult the feature request management document.

please leave open

This issue has been marked as stale due to 210 days of inactivity.
It will be automatically closed in 30 days if no further activity occurs. If this is still relevant, please leave a comment or update it to keep it open.

Leave open please

This issue has been marked as stale due to 90 days of inactivity.
It will be automatically closed in 30 days if no further activity occurs. If this is still relevant, please leave a comment or update it to keep it open.

Leave open please

Navigation Menu

Repository navigation

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

What is the problem this feature will solve?

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

What alternatives have you considered?

32 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

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

Description

What is the problem this feature will solve?

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

What alternatives have you considered?

Activity

rluvaton commented on Mar 6, 2023

benjamingr commented on Mar 7, 2023

rluvaton commented on Mar 7, 2023

benjamingr commented on Mar 7, 2023

ronag commented on Mar 7, 2023

rluvaton commented on Mar 7, 2023

rluvaton commented on Mar 7, 2023

benjamingr commented on Mar 7, 2023

rluvaton commented on Mar 8, 2023

rluvaton commented on Mar 9, 2023

benjamingr commented on Mar 9, 2023

rluvaton commented on Mar 9, 2023

32 remaining items

github-actions commented on Nov 29, 2025

dmurvihill commented on Nov 30, 2025

github-actions commented on Jun 28, 2026

dmurvihill commented on Jun 28, 2026

github-actions commented on Sep 28, 2026

dmurvihill commented on Sep 29, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

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