stream-to-push-stream/bench/index.js

import { Readable, Writable, Duplex } from 'stream';
import { source, sink, duplex } from '../index.js';

const CHUNK_SIZE = 16384;
const CHUNK_COUNT = 100000;

function createBenchmarkReadable() {
  const data = Buffer.alloc(CHUNK_SIZE, 'x');
  let count = 0;
  return new Readable({
    highWaterMark: 1024 * 1024,
    read() {
      if (count < CHUNK_COUNT) {
        count++;
        this.push(data);
      } else {
        this.push(null);
      }
    }
  });
}

function createBenchmarkWritable() {
  let bytesWritten = 0;
  const w = new Writable({
    highWaterMark: 1024 * 1024,
    write(chunk, encoding, callback) {
      bytesWritten += chunk.length;
      callback();
    }
  });
  return w;
}

async function benchmarkSource() {
  const readable = createBenchmarkReadable();
  const pushSrc = source(readable);
  const received = [];
  let byteCount = 0;

  const start = process.hrtime.bigint();

  await new Promise((resolve, reject) => {
    pushSrc.pipe({
      write(data) {
        byteCount += data.length;
      },
      end(err) {
        if (err) return reject(err);
        resolve();
      },
      get paused() { return false; },
      set paused(v) {}
    });
  });

  const end = process.hrtime.bigint();
  const duration = Number(end - start) / 1e9;
  const throughput = (byteCount / duration / 1024 / 1024).toFixed(2);

  return { duration, throughput, bytes: byteCount };
}

async function benchmarkSink() {
  const writable = createBenchmarkWritable();
  const pushSnk = sink(writable);
  const data = Buffer.alloc(CHUNK_SIZE, 'x');

  const start = process.hrtime.bigint();

  for (let i = 0; i < CHUNK_COUNT; i++) {
    pushSnk.write(data);
  }
  pushSnk.end();

  await new Promise((resolve) => {
    writable.on('finish', resolve);
  });

  const end = process.hrtime.bigint();
  const duration = Number(end - start) / 1e9;
  const bytes = CHUNK_COUNT * CHUNK_SIZE;
  const throughput = (bytes / duration / 1024 / 1024).toFixed(2);

  return { duration, throughput, bytes };
}

async function benchmarkDuplex() {
  const data = Buffer.alloc(CHUNK_SIZE, 'x');
  let count = 0;
  let byteCount = 0;

  const duplexStream = new Duplex({
    highWaterMark: 1024 * 1024,
    read() {
      if (count < CHUNK_COUNT) {
        count++;
        this.push(data);
      } else {
        this.push(null);
      }
    },
    write(chunk, encoding, callback) {
      callback();
    }
  });

  const pushDup = duplex(duplexStream);

  const start = process.hrtime.bigint();

  await new Promise((resolve, reject) => {
    pushDup.pipe({
      write(data) {
        byteCount += data.length;
      },
      end(err) {
        if (err) return reject(err);
        resolve();
      },
      get paused() { return false; },
      set paused(v) {}
    });
  });

  const end = process.hrtime.bigint();
  const duration = Number(end - start) / 1e9;
  const throughput = (byteCount / duration / 1024 / 1024).toFixed(2);

  return { duration, throughput, bytes: byteCount };
}

async function benchmarkObjectMode() {
  const arr = Array.from({ length: CHUNK_COUNT }, (_, i) => ({ id: i, value: Math.random() }));
  let index = 0;

  const readable = new Readable({
    objectMode: true,
    highWaterMark: 1024,
    read() {
      if (index < arr.length) {
        this.push(arr[index++]);
      } else {
        this.push(null);
      }
    }
  });

  const pushSrc = source(readable);
  let itemCount = 0;

  const start = process.hrtime.bigint();

  await new Promise((resolve, reject) => {
    pushSrc.pipe({
      write(data) {
        itemCount++;
      },
      end(err) {
        if (err) return reject(err);
        resolve();
      },
      get paused() { return false; },
      set paused(v) {}
    });
  });

  const end = process.hrtime.bigint();
  const duration = Number(end - start) / 1e9;
  const throughput = (itemCount / duration).toFixed(0);

  return { duration, throughput, items: itemCount };
}

async function runBenchmarks() {
  console.log('stream-to-push-stream benchmarks');
  console.log('================================');
  console.log(`Chunk size: ${CHUNK_SIZE} bytes`);
  console.log(`Chunk count: ${CHUNK_COUNT}`);
  console.log();

  console.log('Running benchmarks...\n');

  console.log('1. Source (Readable -> Push-Source):');
  const sourceResult = await benchmarkSource();
  console.log(`   Duration: ${sourceResult.duration.toFixed(3)}s`);
  console.log(`   Throughput: ${sourceResult.throughput} MB/s`);
  console.log();

  console.log('2. Sink (Writable -> Push-Sink):');
  const sinkResult = await benchmarkSink();
  console.log(`   Duration: ${sinkResult.duration.toFixed(3)}s`);
  console.log(`   Throughput: ${sinkResult.throughput} MB/s`);
  console.log();

  console.log('3. Duplex (Duplex -> Push-Duplex):');
  const duplexResult = await benchmarkDuplex();
  console.log(`   Duration: ${duplexResult.duration.toFixed(3)}s`);
  console.log(`   Throughput: ${duplexResult.throughput} MB/s`);
  console.log();

  console.log('4. Object Mode (Readable Object -> Push-Source):');
  const objectResult = await benchmarkObjectMode();
  console.log(`   Duration: ${objectResult.duration.toFixed(3)}s`);
  console.log(`   Throughput: ${objectResult.throughput} items/s`);
  console.log();

  console.log('Benchmark complete.');
}

runBenchmarks().catch(console.error);