push-take-until/bench/index.js

import { bench, run } from './benchlib.js';
import takeUntil from '../index.js';

function createSource() {
  let index = 0;
  let sink = null;
  return {
    pipe(s) { sink = s; return s; },
    resume() {
      while (sink && !sink.paused && index < 10000) {
        sink.write(index++);
      }
    },
    write() {},
    end() {},
    abort() {},
    get sink() { return sink; },
    get paused() { return sink ? sink.paused : true; }
  };
}

function createCollectSink() {
  const items = [];
  return {
    items,
    write(data) { items.push(data); },
    end() {},
    abort() {},
    paused: false,
    ended: false
  };
}

function createSignalStream() {
  let sink = null;
  return {
    pipe(s) { sink = s; return s; },
    write() {},
    end() { if (sink) sink.end(); },
    abort() {},
    get paused() { return false; },
    trigger() { if (sink) sink.end(); }
  };
}

async function benchmark() {
  await bench('takeUntil with early signal (10000 items)', () => {
    const sink = createCollectSink();
    const signal = createSignalStream();
    const stream = takeUntil(signal);
    stream.pipe(sink);
    const source = createSource();
    source.pipe(stream);
    source.resume();
    signal.trigger();
    return sink.items.length;
  }, { iterations: 20, warmup: 2, itemsPerIter: 10000, unit: 'items/s' });

  await bench('takeUntil with late signal (10000 items)', () => {
    const sink = createCollectSink();
    const signal = createSignalStream();
    const stream = takeUntil(signal);
    stream.pipe(sink);
    const source = createSource();
    source.pipe(stream);
    source.resume();
    return sink.items.length;
  }, { iterations: 20, warmup: 2, itemsPerIter: 10000, unit: 'items/s' });

  await bench('takeUntil immediate signal (0 items)', () => {
    const sink = createCollectSink();
    const signal = createSignalStream();
    const stream = takeUntil(signal);
    stream.pipe(sink);
    const source = createSource();
    source.pipe(stream);
    signal.trigger();
    return sink.items.length;
  }, { iterations: 100, warmup: 5 });

  await run({ output: './benchmark-results.json' });
}

benchmark().catch(console.error);