Offload functions to worker threads with shared memory primitives for Node.js.
feat: assign() for pinning tasks to specific workers
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
+57
+23
src/assign.ts
+23
src/assign.ts
···
1
+
import { Task } from './task.ts';
2
+
import { StreamTask } from './stream-task.ts';
3
+
import type { WorkerHandle } from './runner.ts';
4
+
5
+
/**
6
+
* Returns a copy of the task pinned to a specific worker.
7
+
* The original task is unchanged.
8
+
* @param worker - The worker handle to pin the task to.
9
+
* @param task - The task or streaming task to assign.
10
+
* @returns A new task with the same id and args, pinned to the given worker.
11
+
*/
12
+
export function assign<T>(worker: WorkerHandle, task: Task<T>): Task<T>;
13
+
export function assign<T>(worker: WorkerHandle, task: StreamTask<T>): StreamTask<T>;
14
+
export function assign(worker: WorkerHandle, task: Task<any> | StreamTask<any>): Task<any> | StreamTask<any> {
15
+
if (task instanceof StreamTask) {
16
+
const copy = new StreamTask(task.id, task.args);
17
+
copy.worker = worker;
18
+
return copy;
19
+
}
20
+
const copy = new Task(task.id, task.args);
21
+
copy.worker = worker;
22
+
return copy;
23
+
}
+1
src/index.ts
+1
src/index.ts
···
6
6
export type { ChannelOptions } from './channel.ts';
7
7
export { workers } from './worker-pool.ts';
8
8
export { transfer } from './transfer.ts';
9
+
export { assign } from './assign.ts';
9
10
export type { Runner, WorkerHandle, WorkerOptions } from './runner.ts';
10
11
export {
11
12
shared,
+33
test/assign.test.ts
+33
test/assign.test.ts
···
1
+
import { describe, it } from 'node:test';
2
+
import assert from 'node:assert/strict';
3
+
import { mo, workers, assign } from 'moroutine';
4
+
5
+
const double = mo(import.meta, (n: number): number => n * 2);
6
+
7
+
describe('assign()', () => {
8
+
it('returns a new task with worker set', () => {
9
+
const run = workers(1);
10
+
try {
11
+
const task = double(5);
12
+
const assigned = assign(run.workers[0], task);
13
+
assert.notStrictEqual(assigned, task);
14
+
assert.equal(assigned.id, task.id);
15
+
assert.deepEqual(assigned.args, task.args);
16
+
assert.notEqual(assigned.uid, task.uid);
17
+
assert.equal(assigned.worker, run.workers[0]);
18
+
} finally {
19
+
run[Symbol.dispose]();
20
+
}
21
+
});
22
+
23
+
it('does not modify the original task', () => {
24
+
const run = workers(1);
25
+
try {
26
+
const task = double(5);
27
+
assign(run.workers[0], task);
28
+
assert.equal(task.worker, undefined);
29
+
} finally {
30
+
run[Symbol.dispose]();
31
+
}
32
+
});
33
+
});