Something went wrong. Try again.
Offload functions to worker threads with shared memory primitives for Node.js.
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091import { describe, it } from 'node:test';import { setTimeout as sleep } from 'node:timers/promises';import assert from 'node:assert/strict';import { workers, int32atomic } from 'moroutine';import { countUp, failAfter } from './fixtures/stream-gen.ts';import { countingGen } from './fixtures/backpressure.ts';
describe('streaming moroutines (worker -> main)', () => { it('iterates yielded values from worker', async () => { const run = workers(1); try { const results: number[] = []; for await (const value of run(countUp(5))) { results.push(value); } assert.deepEqual(results, [0, 1, 2, 3, 4]); } finally { run[Symbol.dispose](); } });
it('supports early break', async () => { const run = workers(1); try { const results: number[] = []; for await (const value of run(countUp(100))) { results.push(value); if (results.length >= 3) break; } assert.deepEqual(results, [0, 1, 2]); } finally { run[Symbol.dispose](); } });
it('propagates errors from worker generator', async () => { const run = workers(1); try { const results: number[] = []; await assert.rejects( async () => { for await (const value of run(failAfter(3))) { results.push(value); } }, { message: 'intentional error' }, ); assert.deepEqual(results, [0, 1, 2]); } finally { run[Symbol.dispose](); } });
it('works on dedicated worker without pool', async () => { const results: number[] = []; for await (const value of countUp(3)) { results.push(value); } assert.deepEqual(results, [0, 1, 2]); });
it('accepts highWaterMark option', async () => { const run = workers(1); try { const results: number[] = []; for await (const value of run(countUp(5), { highWaterMark: 2 })) { results.push(value); } assert.deepEqual(results, [0, 1, 2, 3, 4]); } finally { run[Symbol.dispose](); } });
it('honors highWaterMark: slow consumer stalls the producer', async () => { using run = workers(1); const highWaterMark = 4; const emitted = int32atomic(); let consumed = 0; let peak = 0; for await (const _ of run(countingGen(50, emitted), { highWaterMark })) { consumed++; const gap = emitted.load() - consumed; if (gap > peak) peak = gap; await sleep(2); } // With highWater=4, worst-case observable gap is ~highWater + 2 slop. assert.ok(peak <= highWaterMark + 3, `expected peak gap <= ${highWaterMark + 3}, got ${peak}`); });});