stream: fix nested compose error propagation · nodejs/node@3527646

_resolve({ __proto__: null, done: true, cb });

},

destroy(err, cb) {

ac.abort();

ac.abort(err);

// If the source async iterator is waiting for the next write/final

// signal, unblock it so the readable side can observe the abort and

// finish destroying.

if (resolve !== null) {

const _resolve = resolve;

resolve = null;

_resolve({ __proto__: null, done: true, cb() {} });

}

cb(err);

},

};

const { Buffer } = require('buffer');

const {

ERR_INVALID_ARG_TYPE,

ERR_STREAM_NULL_VALUES,

} = require('internal/errors').codes;

aggregateTwoErrors,

codes: {

ERR_INVALID_ARG_TYPE,

ERR_STREAM_NULL_VALUES,

},

} = require('internal/errors');

function from(Readable, iterable, opts) {

let iterator;

// TODO(ronag): What options should be allowed?

...opts,

});

const originalDestroy = readable._destroy;

// Flag to protect against _read

// being called before last iteration completion.

};

readable._destroy = function(error, cb) {

PromisePrototypeThen(

close(error),

() => process.nextTick(cb, error), // nextTick is here in case cb throws

(e) => process.nextTick(cb, e || error),

);

originalDestroy.call(this, error, (destroyError) => {

const combinedError = destroyError || error;

PromisePrototypeThen(

close(combinedError),

// nextTick is here in case cb throws

() => process.nextTick(cb, combinedError),

(closeError) => process.nextTick(

cb,

aggregateTwoErrors(combinedError, closeError),

),

);

});

};

async function close(error) {

).then(common.mustCall());

}

{

// Errors from nested `.compose()` calls should propagate instead of hanging.

const stream = Readable.from(['hello'])

.compose(async function *(source) { // eslint-disable-line require-yield

for await (const chunk of source) {

throw new Error(`boom: ${chunk}`);

}

})

.compose(async function *(source) {

yield* source;

});

assert.rejects(

stream.toArray(),

/boom: hello/,

).then(common.mustCall());

}

{

// AbortSignal

const ac = new AbortController();