Стримы, async iterators и pipelines

Когда результатов много и приходят они со временем, нужна абстракция, которая помнит о потреблении и о памяти. В JavaScript таких абстракций три: async generator, Node Stream, Web Stream. Они решают одну задачу, но с разной семантикой. Этот урок про то, что выбрать в каждом случае и почему.

Полезно держать рядом доклад Маттео Коллины про backpressure и доку WHATWG Streams. По ним же сверяемся, когда внутри Node-овых стримов происходит что-то странное.

1. Зачем вообще стрим

Главная причина, память. Файл на 5 GB не лезет в Buffer.from. Но читать его построчно и сразу обрабатывать получается даже на 50 MB оперативки.

Вторая причина, backpressure. Если продюсер быстрее, чем потребитель, без давления память закончится. С давлением, продюсер тормозится автоматически.

Третья причина, читаемость. Pipeline read | parse | filter | write, это та же идея, что unix pipe, и читается так же.

2. Async iterator: pull-протокол

Самая простая форма. Объект с методом [Symbol.asyncIterator](), возвращающим объект с методом next(): Promise<{ value, done }>. Ровно тот же протокол, что у синхронного итератора, но next отдаёт промис.

Async generator (async function*), это сахар, который превращается в state machine с next/return/throw.

async function* readLines(file: string): AsyncIterable<string> {
  const handle = await fs.open(file, 'r');
  try {
    let buffer = '';
    for await (const chunk of handle.createReadStream({ encoding: 'utf8' })) {
      buffer += chunk;
      const lines = buffer.split('\n');
      buffer = lines.pop() ?? '';
      for (const line of lines) yield line;
    }
    if (buffer.length > 0) yield buffer;
  } finally {
    await handle.close();
  }
}

for await (const line of readLines('users.jsonl')) {
  process(line);
}

Что важно подметить:

• Pull. Следующий await происходит, когда потребитель попросил через for await. Файл читается чанками по мере необходимости.
• finally, гарантия закрытия ресурса. Если потребитель прервал через break, движок зовёт iterator.return(), и finally отработает.
• Ошибки, обычные throw. Одна ошибка останавливает итерацию.

Async generator, лучший выбор для in-memory трансформаций.

Ленивость async-итератора проще всего прочувствовать на срезе: сколько значений попросил потребитель, столько источник и отдал. Ниже редактор с такой задачей, тесты считают обращения к источнику. Гоняются прямо в браузере, ставить ничего не нужно.

3. Node Stream: четыре класса

Node-стримы старше Web Streams. Их API многословнее, но возможностей больше.

Класс	Роль
Readable	источник данных
Writable	приёмник данных
Duplex	оба сразу, но независимые потоки внутрь и наружу
Transform	специальный Duplex, “прочитал, превратил, записал”

Самый часто пишется Transform. Создавать его удобно через Transform.from(asyncFn) (Node 18+) или конструктор:

import { Transform } from 'node:stream';

const upper = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk.toString().toUpperCase());
  },
});

Readable тоже удобнее всего создавать из async iterable:

import { Readable } from 'node:stream';

async function* source() {
  for (let i = 0; i < 1_000; i += 1) yield `line ${i}\n`;
}

const stream = Readable.from(source());

Writable обычно встроенный (fs.createWriteStream, process.stdout), но при необходимости пишется так:

import { Writable } from 'node:stream';

const sink = new Writable({
  write(chunk, encoding, callback) {
    console.log('got', chunk.toString());
    callback();
  },
});

4. Backpressure в Node streams

Каждый стрим имеет внутренний буфер с порогом highWaterMark (по умолчанию 16 KB для байтов, 16 объектов для objectMode). Когда буфер заполнен:

• Writable.write(chunk) возвращает false. Это сигнал “медленнее”.
• Readable останавливает чтение из источника.

Соблюдать backpressure руками выглядит так:

import { once } from 'node:events';

const writeAll = async (writable, source) => {
  for await (const chunk of source) {
    if (writable.write(chunk) === false) {
      await once(writable, 'drain');
    }
  }
  writable.end();
};

Это рабочий код, но он многословный. На практике используют pipe/pipeline, которые сами это делают.

producer

flowing

buffer

0/8

consumer

0

0 produced0 consumed0 paused0 tps

Подними producer выше consumer, буфер заполняется. Когда длина достигает hwm, producer переходит в paused: это и есть backpressure. Опустится буфер ниже половины hwm, producer возвращается в flowing. Это упрощённая модель: в Node hwm, это порог для возврата false у write(), в Web Streams похожая логика реализована через возврат промиса из pull.

5. pipe vs pipeline

pipe старый, pipeline новый. Оба соединяют Readable с Writable (или цепочку с трансформациями посередине).

// плохо: ошибка в одном из стримов оставит другие висеть
src.pipe(transform).pipe(dst);

// хорошо: pipeline закроет всё корректно
import { pipeline } from 'node:stream/promises';

await pipeline(src, transform, dst);

Что pipeline делает, чего нет у pipe:

• При ошибке в любом стриме закрывает все остальные через destroy(err).
• Возвращает промис (в node:stream/promises), интегрируется с await/try/catch.
• Поддерживает { signal: AbortSignal } для отмены.

pipe остался для совместимости. В новом коде, всегда pipeline.