Something went wrong. Try again.
Offload functions to worker threads with shared memory primitives for Node.js.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869import { describe, it } from 'node:test';import assert from 'node:assert/strict';import { workers, channel, int32atomic } from 'moroutine';import { sumStream, uppercaseStream } from './fixtures/stream-input.ts';import { slowSumPeakGap } from './fixtures/backpressure.ts';
describe('channel() input (main -> worker)', () => { it('pipes async iterable to worker', async () => { async function* numbers() { yield 1; yield 2; yield 3; } const run = workers(1); try { const result = await run(sumStream(channel(numbers()))); assert.equal(result, 6); } finally { run[Symbol.dispose](); } });
it('bidirectional: channel input + generator output', async () => { async function* words() { yield 'hello'; yield 'world'; } const run = workers(1); try { const results: string[] = []; for await (const word of run(uppercaseStream(channel(words())))) { results.push(word); } assert.deepEqual(results, ['HELLO', 'WORLD']); } finally { run[Symbol.dispose](); } });
it('accepts highWaterMark on channel()', async () => { async function* numbers() { yield 1; yield 2; yield 3; } const run = workers(1); try { const result = await run(sumStream(channel(numbers(), { highWaterMark: 2 }))); assert.equal(result, 6); } finally { run[Symbol.dispose](); } });
it('honors channel highWaterMark: parent producer stalls for slow worker', async () => { using run = workers(1); const highWaterMark = 4; const emitted = int32atomic(); async function* source() { for (let i = 0; i < 50; i++) { yield i; emitted.add(1); } } const peak = await run(slowSumPeakGap(channel(source(), { highWaterMark }), emitted, 2)); assert.ok(peak <= highWaterMark + 3, `expected peak gap <= ${highWaterMark + 3}, got ${peak}`); });});