GeezerVison is showing github.com — stream: fix nested compose error propagation · nodejs/node@3607857
File treeExpand file treeCollapse file tree Open diff view settings Expand file treeCollapse file tree Open diff view settings Collapse file Copy file name to clipboardExpand all lines: lib/internal/streams/duplexify.js +11-1Lines changed: 11 additions & 1 deletion | Original file line number | Diff line number | Diff line change |
|---|
@@ -239,7 +239,17 @@ function fromAsyncGen(fn) { | 239 | 239 | _resolve({ __proto__: null, done: true, cb }); | 240 | 240 | }, | 241 | 241 | destroy(err, cb) { | 242 | | - ac.abort(); | | 242 | + ac.abort(err); | | 243 | + | | 244 | + // If the source async iterator is waiting for the next write/final | | 245 | + // signal, unblock it so the readable side can observe the abort and | | 246 | + // finish destroying. | | 247 | + if (resolve !== null) { | | 248 | + const _resolve = resolve; | | 249 | + resolve = null; | | 250 | + _resolve({ __proto__: null, done: true, cb() {} }); | | 251 | + } | | 252 | + | 243 | 253 | cb(err); | 244 | 254 | }, | 245 | 255 | }; | |
Collapse file Copy file name to clipboardExpand all lines: lib/internal/streams/from.js +19-8Lines changed: 19 additions & 8 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 8 | 8 | const { Buffer } = require('buffer'); | 9 | 9 |
| 10 | 10 | const { | 11 | | - ERR_INVALID_ARG_TYPE, | 12 | | - ERR_STREAM_NULL_VALUES, | 13 | | -} = require('internal/errors').codes; | | 11 | + aggregateTwoErrors, | | 12 | + codes: { | | 13 | + ERR_INVALID_ARG_TYPE, | | 14 | + ERR_STREAM_NULL_VALUES, | | 15 | + }, | | 16 | +} = require('internal/errors'); | 14 | 17 |
| 15 | 18 | function from(Readable, iterable, opts) { | 16 | 19 | let iterator; | @@ -43,6 +46,7 @@ function from(Readable, iterable, opts) { | 43 | 46 | // TODO(ronag): What options should be allowed? | 44 | 47 | ...opts, | 45 | 48 | }); | | 49 | + const originalDestroy = readable._destroy; | 46 | 50 |
| 47 | 51 | // Flag to protect against _read | 48 | 52 | // being called before last iteration completion. | @@ -64,11 +68,18 @@ function from(Readable, iterable, opts) { | 64 | 68 | }; | 65 | 69 |
| 66 | 70 | readable._destroy = function(error, cb) { | 67 | | - PromisePrototypeThen( | 68 | | - close(error), | 69 | | - () => process.nextTick(cb, error), // nextTick is here in case cb throws | 70 | | - (e) => process.nextTick(cb, e || error), | 71 | | - ); | | 71 | + originalDestroy.call(this, error, (destroyError) => { | | 72 | + const combinedError = destroyError || error; | | 73 | + PromisePrototypeThen( | | 74 | + close(combinedError), | | 75 | + // nextTick is here in case cb throws | | 76 | + () => process.nextTick(cb, combinedError), | | 77 | + (closeError) => process.nextTick( | | 78 | + cb, | | 79 | + aggregateTwoErrors(combinedError, closeError), | | 80 | + ), | | 81 | + ); | | 82 | + }); | 72 | 83 | }; | 73 | 84 |
| 74 | 85 | async function close(error) { | |
Collapse file Copy file name to clipboardExpand all lines: test/parallel/test-stream-readable-compose.js +18Lines changed: 18 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -116,6 +116,24 @@ const assert = require('assert'); | 116 | 116 | ).then(common.mustCall()); | 117 | 117 | } | 118 | 118 |
| | 119 | +{ | | 120 | + // Errors from nested `.compose()` calls should propagate instead of hanging. | | 121 | + const stream = Readable.from(['hello']) | | 122 | + .compose(async function *(source) { // eslint-disable-line require-yield | | 123 | + for await (const chunk of source) { | | 124 | + throw new Error(`boom: ${chunk}`); | | 125 | + } | | 126 | + }) | | 127 | + .compose(async function *(source) { | | 128 | + yield* source; | | 129 | + }); | | 130 | + | | 131 | + assert.rejects( | | 132 | + stream.toArray(), | | 133 | + /boom: hello/, | | 134 | + ).then(common.mustCall()); | | 135 | +} | | 136 | + | 119 | 137 | { | 120 | 138 | // AbortSignal | 121 | 139 | const ac = new AbortController(); | |
|