parser.stream() и Transform

Есть два способа подключить разборщик к потоку: итератор для обычного кода и Transform для конвейеров pipeline.

Асинхронный итератор#

parser.stream(source: Readable, opts?: { signal?: AbortSignal }): AsyncIterable<Record>

Источник читается лениво, порция за порцией. Подробности про память и остановку — в разделе потоки.

Transform#

import { pipeline } from 'node:stream/promises';
import { createTransform } from 'sverlo';

await pipeline(
  createReadStream('app.log'),
  createTransform({ format: 'json-lines', query: 'level>=error' }),
  async function* (records) {
    for await (const rec of records) yield JSON.stringify(rec) + '\n';
  },
  createWriteStream('errors.ndjson')
);

createTransform принимает те же параметры, что и createParser, плюс необязательный query. На вход идут байты или строки, на выходе объекты Record, поток работает в object mode.

События и статистика#

transform.on('skip', (e) => metrics.inc('skipped', { code: e.code }));

Ошибки в конвейере#

В режиме onError: 'throw' исключение уничтожает поток обычным способом, и pipeline завершится с ошибкой. В остальных режимах ошибки разбора до pipeline не доходят, их нужно собирать через 'skip' или parser.errors.