GeezerVison is showing github.com — stream: make async iterator .next() always resolve · nodejs/node@b809fa8
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/async_iterator.js +10-5Lines changed: 10 additions & 5 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -39,6 +39,11 @@ function onReadable(iter) { | 39 | 39 | function wrapForNext(lastPromise, iter) { | 40 | 40 | return (resolve, reject) => { | 41 | 41 | lastPromise.then(() => { | | 42 | + if (iter[kEnded]) { | | 43 | + resolve(createIterResult(undefined, true)); | | 44 | + return; | | 45 | + } | | 46 | + | 42 | 47 | iter[kHandlePromise](resolve, reject); | 43 | 48 | }, reject); | 44 | 49 | }; | @@ -61,7 +66,7 @@ const ReadableStreamAsyncIteratorPrototype = Object.setPrototypeOf({ | 61 | 66 | } | 62 | 67 |
| 63 | 68 | if (this[kEnded]) { | 64 | | - return Promise.resolve(createIterResult(null, true)); | | 69 | + return Promise.resolve(createIterResult(undefined, true)); | 65 | 70 | } | 66 | 71 |
| 67 | 72 | if (this[kStream].destroyed) { | @@ -74,7 +79,7 @@ const ReadableStreamAsyncIteratorPrototype = Object.setPrototypeOf({ | 74 | 79 | if (this[kError]) { | 75 | 80 | reject(this[kError]); | 76 | 81 | } else { | 77 | | - resolve(createIterResult(null, true)); | | 82 | + resolve(createIterResult(undefined, true)); | 78 | 83 | } | 79 | 84 | }); | 80 | 85 | }); | @@ -115,7 +120,7 @@ const ReadableStreamAsyncIteratorPrototype = Object.setPrototypeOf({ | 115 | 120 | reject(err); | 116 | 121 | return; | 117 | 122 | } | 118 | | - resolve(createIterResult(null, true)); | | 123 | + resolve(createIterResult(undefined, true)); | 119 | 124 | }); | 120 | 125 | }); | 121 | 126 | }, | @@ -131,7 +136,6 @@ const createReadableStreamAsyncIterator = (stream) => { | 131 | 136 | value: stream._readableState.endEmitted, | 132 | 137 | writable: true | 133 | 138 | }, | 134 | | - [kLastPromise]: { value: null, writable: true }, | 135 | 139 | // the function passed to new Promise | 136 | 140 | // is cached so we avoid allocating a new | 137 | 141 | // closure at every run | @@ -151,6 +155,7 @@ const createReadableStreamAsyncIterator = (stream) => { | 151 | 155 | writable: true, | 152 | 156 | }, | 153 | 157 | }); | | 158 | + iterator[kLastPromise] = null; | 154 | 159 |
| 155 | 160 | finished(stream, (err) => { | 156 | 161 | if (err && err.code !== 'ERR_STREAM_PREMATURE_CLOSE') { | @@ -172,7 +177,7 @@ const createReadableStreamAsyncIterator = (stream) => { | 172 | 177 | iterator[kLastPromise] = null; | 173 | 178 | iterator[kLastResolve] = null; | 174 | 179 | iterator[kLastReject] = null; | 175 | | - resolve(createIterResult(null, true)); | | 180 | + resolve(createIterResult(undefined, true)); | 176 | 181 | } | 177 | 182 | iterator[kEnded] = true; | 178 | 183 | }); | |
Collapse file Copy file name to clipboardExpand all lines: test/parallel/test-stream-readable-async-iterators.js +65Lines changed: 65 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -393,6 +393,71 @@ async function tests() { | 393 | 393 | r.destroy(null); | 394 | 394 | } | 395 | 395 | })(); | | 396 | + | | 397 | + await (async () => { | | 398 | + console.log('all next promises must be resolved on end'); | | 399 | + const r = new Readable({ | | 400 | + objectMode: true, | | 401 | + read() { | | 402 | + } | | 403 | + }); | | 404 | + | | 405 | + const b = r[Symbol.asyncIterator](); | | 406 | + const c = b.next(); | | 407 | + const d = b.next(); | | 408 | + r.push(null); | | 409 | + assert.deepStrictEqual(await c, { done: true, value: undefined }); | | 410 | + assert.deepStrictEqual(await d, { done: true, value: undefined }); | | 411 | + })(); | | 412 | + | | 413 | + await (async () => { | | 414 | + console.log('all next promises must be resolved on destroy'); | | 415 | + const r = new Readable({ | | 416 | + objectMode: true, | | 417 | + read() { | | 418 | + } | | 419 | + }); | | 420 | + | | 421 | + const b = r[Symbol.asyncIterator](); | | 422 | + const c = b.next(); | | 423 | + const d = b.next(); | | 424 | + r.destroy(); | | 425 | + assert.deepStrictEqual(await c, { done: true, value: undefined }); | | 426 | + assert.deepStrictEqual(await d, { done: true, value: undefined }); | | 427 | + })(); | | 428 | + | | 429 | + await (async () => { | | 430 | + console.log('all next promises must be resolved on destroy with error'); | | 431 | + const r = new Readable({ | | 432 | + objectMode: true, | | 433 | + read() { | | 434 | + } | | 435 | + }); | | 436 | + | | 437 | + const b = r[Symbol.asyncIterator](); | | 438 | + const c = b.next(); | | 439 | + const d = b.next(); | | 440 | + const err = new Error('kaboom'); | | 441 | + r.destroy(err); | | 442 | + | | 443 | + await Promise.all([(async () => { | | 444 | + let e; | | 445 | + try { | | 446 | + await c; | | 447 | + } catch (_e) { | | 448 | + e = _e; | | 449 | + } | | 450 | + assert.strictEqual(e, err); | | 451 | + })(), (async () => { | | 452 | + let e; | | 453 | + try { | | 454 | + await d; | | 455 | + } catch (_e) { | | 456 | + e = _e; | | 457 | + } | | 458 | + assert.strictEqual(e, err); | | 459 | + })()]); | | 460 | + })(); | 396 | 461 | } | 397 | 462 |
| 398 | 463 | // to avoid missing some tests if a promise does not resolve | |
|