Pipe files with backpressure

Scoped v2 example

import { createReadStream, createWriteStream } from 'node:fs';
import { once } from 'node:events';
import { ConvertBuddy } from '@diakrio/converter';

const input = createReadStream('events.csv');
const output = createWriteStream('events.ndjson');

const controller = new ConvertBuddy({ profile: true }).stream(input, {
  inputFormat: 'csv',
  outputFormat: 'ndjson',
  onBytes: async (chunk) => {
    if (!output.write(chunk)) await once(output, 'drain');
  }
});

try {
  const stats = await controller.done;
  output.end();
  await once(output, 'finish');
  console.table({
    mbPerSecond: stats.wallThroughputMbPerSec.toFixed(2),
    recordsPerSecond: Math.round(stats.recordsPerSec),
    records: stats.recordsOut,
    peakEngineBufferMiB: (stats.maxBufferSize / 2 ** 20).toFixed(2),
    processRssMiB: (process.memoryUsage().rss / 2 ** 20).toFixed(2)
  });
} catch (error) {
  output.destroy();
  throw error;
}

After the scoped package is released, save as convert.mjs, then run node convert.mjs.

Measure both layers

Diakrio reports engine timings and its peak partial-record buffer. Node'sprocess.memoryUsage() reports process-wide runtime memory. Keep both numbers: they answer different questions and make benchmark claims reproducible.