Async Iteration over Streams

Every Readable implements Symbol.asyncIterator, so for await (const chunk of stream) reads it with backpressure handled: the loop body finishes before the next chunk is pulled, and a slow body slows the source. Leaving the loop — break, return or a thrown error — destroys the stream and closes the underlying handle, as the ENOENT case below shows; pass { destroyOnReturn: false } to stream.iterator() to resume reading later.

Iterating, and composing with the stream helper methodsJavaScript

const stats = Readable.from(['1,emea,10.5', '2,apac,3.25', '3,emea,7'])
  .map((line) => line.split(',')).filter(([, region]) => region === 'emea')
  .map(([id, , amount]) => `${id}: ${Number(amount).toFixed(2)}`);
console.log(await stats.toArray());
const slow = Readable.from([1, 2, 3]).map(   // three awaits at once, input order kept
  async n => { await new Promise(r => setTimeout(r, 20)); return n * 2; }, { concurrency: 3 });
console.log(await slow.toArray());
const stream = createReadStream('missing.csv');
try { for await (const chunk of stream) console.log(chunk.length); }
catch (err) { console.log(err.code, '| destroyed:', stream.destroyed); }

Output

[ '1: 10.50', '3: 7.00' ]
[ 2, 4, 6 ]
ENOENT | destroyed: true

map, filter, take, drop, flatMap, forEach, some, every, find, reduce and toArray mirror the array methods but stay lazy: readable.take(1).toArray() on a 108 MB file reads one chunk and stops. The concurrency option on map, filter and forEach is the reason to prefer them over a hand-written loop — several awaits run in flight, output order preserved.

MENU

Async Iteration over Streams