diff --git a/lib/internal/streams/operators.js b/lib/internal/streams/operators.js index 6db2df0e3646e0..a5cbe1af428fae 100644 --- a/lib/internal/streams/operators.js +++ b/lib/internal/streams/operators.js @@ -29,6 +29,7 @@ const { validateFunction, } = require('internal/validators'); const { kWeakHandler, kResistStopPropagation } = require('internal/event_target'); +const destroyImpl = require('internal/streams/destroy'); const { finished } = require('internal/streams/end-of-stream'); const kEmpty = Symbol('kEmpty'); @@ -177,6 +178,7 @@ function map(fn, options) { resume(); resume = null; } + destroyImpl.destroyer(stream, null); } }.call(this); } diff --git a/test/parallel/test-stream-some-find-every.mjs b/test/parallel/test-stream-some-find-every.mjs index 0617102bc471e3..c18e418ddce592 100644 --- a/test/parallel/test-stream-some-find-every.mjs +++ b/test/parallel/test-stream-some-find-every.mjs @@ -1,6 +1,6 @@ import * as common from '../common/index.mjs'; import { setTimeout } from 'timers/promises'; -import { Readable } from 'stream'; +import { PassThrough, Readable } from 'stream'; import assert from 'assert'; @@ -77,6 +77,36 @@ function oneTo5Async() { await findStream.find(common.mustCall((x) => x > 1, 2)); await checkDestroyed(findStream); + const openSomeStream = new PassThrough({ objectMode: true }); + openSomeStream.write(1); + openSomeStream.write(2); + openSomeStream.write(3); + assert.strictEqual( + await openSomeStream.some(common.mustCall((x) => x > 2, 3)), + true, + ); + await checkDestroyed(openSomeStream); + + const openEveryStream = new PassThrough({ objectMode: true }); + openEveryStream.write(1); + openEveryStream.write(2); + openEveryStream.write(3); + assert.strictEqual( + await openEveryStream.every(common.mustCall((x) => x < 3, 3)), + false, + ); + await checkDestroyed(openEveryStream); + + const openFindStream = new PassThrough({ objectMode: true }); + openFindStream.write(1); + openFindStream.write(2); + openFindStream.write(3); + assert.strictEqual( + await openFindStream.find(common.mustCall((x) => x > 1, 2)), + 2, + ); + await checkDestroyed(openFindStream); + // When short circuit isn't possible the whole stream is iterated await oneTo5().some(common.mustCall(() => false, 5)); await oneTo5().every(common.mustCall(() => true, 5));