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);