1import * as common from '../common/index.mjs'; 2import { setTimeout } from 'timers/promises'; 3import { Readable } from 'stream'; 4import assert from 'assert'; 5 6 7function oneTo5() { 8 return Readable.from([1, 2, 3, 4, 5]); 9} 10 11function oneTo5Async() { 12 return oneTo5().map(async (x) => { 13 await Promise.resolve(); 14 return x; 15 }); 16} 17{ 18 // Some, find, and every work with a synchronous stream and predicate 19 assert.strictEqual(await oneTo5().some((x) => x > 3), true); 20 assert.strictEqual(await oneTo5().every((x) => x > 3), false); 21 assert.strictEqual(await oneTo5().find((x) => x > 3), 4); 22 assert.strictEqual(await oneTo5().some((x) => x > 6), false); 23 assert.strictEqual(await oneTo5().every((x) => x < 6), true); 24 assert.strictEqual(await oneTo5().find((x) => x > 6), undefined); 25 assert.strictEqual(await Readable.from([]).some(() => true), false); 26 assert.strictEqual(await Readable.from([]).every(() => true), true); 27 assert.strictEqual(await Readable.from([]).find(() => true), undefined); 28} 29 30{ 31 // Some, find, and every work with an asynchronous stream and synchronous predicate 32 assert.strictEqual(await oneTo5Async().some((x) => x > 3), true); 33 assert.strictEqual(await oneTo5Async().every((x) => x > 3), false); 34 assert.strictEqual(await oneTo5Async().find((x) => x > 3), 4); 35 assert.strictEqual(await oneTo5Async().some((x) => x > 6), false); 36 assert.strictEqual(await oneTo5Async().every((x) => x < 6), true); 37 assert.strictEqual(await oneTo5Async().find((x) => x > 6), undefined); 38} 39 40{ 41 // Some, find, and every work on synchronous streams with an asynchronous predicate 42 assert.strictEqual(await oneTo5().some(async (x) => x > 3), true); 43 assert.strictEqual(await oneTo5().every(async (x) => x > 3), false); 44 assert.strictEqual(await oneTo5().find(async (x) => x > 3), 4); 45 assert.strictEqual(await oneTo5().some(async (x) => x > 6), false); 46 assert.strictEqual(await oneTo5().every(async (x) => x < 6), true); 47 assert.strictEqual(await oneTo5().find(async (x) => x > 6), undefined); 48} 49 50{ 51 // Some, find, and every work on asynchronous streams with an asynchronous predicate 52 assert.strictEqual(await oneTo5Async().some(async (x) => x > 3), true); 53 assert.strictEqual(await oneTo5Async().every(async (x) => x > 3), false); 54 assert.strictEqual(await oneTo5Async().find(async (x) => x > 3), 4); 55 assert.strictEqual(await oneTo5Async().some(async (x) => x > 6), false); 56 assert.strictEqual(await oneTo5Async().every(async (x) => x < 6), true); 57 assert.strictEqual(await oneTo5Async().find(async (x) => x > 6), undefined); 58} 59 60{ 61 async function checkDestroyed(stream) { 62 await setTimeout(); 63 assert.strictEqual(stream.destroyed, true); 64 } 65 66 { 67 // Some, find, and every short circuit 68 const someStream = oneTo5(); 69 await someStream.some(common.mustCall((x) => x > 2, 3)); 70 await checkDestroyed(someStream); 71 72 const everyStream = oneTo5(); 73 await everyStream.every(common.mustCall((x) => x < 3, 3)); 74 await checkDestroyed(everyStream); 75 76 const findStream = oneTo5(); 77 await findStream.find(common.mustCall((x) => x > 1, 2)); 78 await checkDestroyed(findStream); 79 80 // When short circuit isn't possible the whole stream is iterated 81 await oneTo5().some(common.mustCall(() => false, 5)); 82 await oneTo5().every(common.mustCall(() => true, 5)); 83 await oneTo5().find(common.mustCall(() => false, 5)); 84 } 85 86 { 87 // Some, find, and every short circuit async stream/predicate 88 const someStream = oneTo5Async(); 89 await someStream.some(common.mustCall(async (x) => x > 2, 3)); 90 await checkDestroyed(someStream); 91 92 const everyStream = oneTo5Async(); 93 await everyStream.every(common.mustCall(async (x) => x < 3, 3)); 94 await checkDestroyed(everyStream); 95 96 const findStream = oneTo5Async(); 97 await findStream.find(common.mustCall(async (x) => x > 1, 2)); 98 await checkDestroyed(findStream); 99 100 // When short circuit isn't possible the whole stream is iterated 101 await oneTo5Async().some(common.mustCall(async () => false, 5)); 102 await oneTo5Async().every(common.mustCall(async () => true, 5)); 103 await oneTo5Async().find(common.mustCall(async () => false, 5)); 104 } 105} 106 107{ 108 // Concurrency doesn't affect which value is found. 109 const found = await Readable.from([1, 2]).find(async (val) => { 110 if (val === 1) { 111 await setTimeout(100); 112 } 113 return true; 114 }, { concurrency: 2 }); 115 assert.strictEqual(found, 1); 116} 117 118{ 119 // Support for AbortSignal 120 for (const op of ['some', 'every', 'find']) { 121 { 122 const ac = new AbortController(); 123 assert.rejects(Readable.from([1, 2, 3])[op]( 124 () => new Promise(() => { }), 125 { signal: ac.signal } 126 ), { 127 name: 'AbortError', 128 }, `${op} should abort correctly with sync abort`).then(common.mustCall()); 129 ac.abort(); 130 } 131 { 132 // Support for pre-aborted AbortSignal 133 assert.rejects(Readable.from([1, 2, 3])[op]( 134 () => new Promise(() => { }), 135 { signal: AbortSignal.abort() } 136 ), { 137 name: 'AbortError', 138 }, `${op} should abort with pre-aborted abort controller`).then(common.mustCall()); 139 } 140 } 141} 142{ 143 // Error cases 144 for (const op of ['some', 'every', 'find']) { 145 assert.rejects(async () => { 146 await Readable.from([1])[op](1); 147 }, /ERR_INVALID_ARG_TYPE/, `${op} should throw for invalid function`).then(common.mustCall()); 148 assert.rejects(async () => { 149 await Readable.from([1])[op]((x) => x, { 150 concurrency: 'Foo' 151 }); 152 }, /ERR_OUT_OF_RANGE/, `${op} should throw for invalid concurrency`).then(common.mustCall()); 153 assert.rejects(async () => { 154 await Readable.from([1])[op]((x) => x, 1); 155 }, /ERR_INVALID_ARG_TYPE/, `${op} should throw for invalid concurrency`).then(common.mustCall()); 156 assert.rejects(async () => { 157 await Readable.from([1])[op]((x) => x, { 158 signal: true 159 }); 160 }, /ERR_INVALID_ARG_TYPE/, `${op} should throw for invalid signal`).then(common.mustCall()); 161 } 162} 163{ 164 for (const op of ['some', 'every', 'find']) { 165 const stream = oneTo5(); 166 Object.defineProperty(stream, 'map', { 167 value: common.mustNotCall(), 168 }); 169 // Check that map isn't getting called. 170 stream[op](() => {}); 171 } 172} 173