| 1 | import * as NodeWorker from "@effect/platform-node/NodeWorker" |
| 2 | import { assert, describe, it } from "@effect/vitest" |
| 3 | import * as Deferred from "effect/Deferred" |
| 4 | import * as Effect from "effect/Effect" |
| 5 | import * as Queue from "effect/Queue" |
| 6 | import * as Worker from "effect/workers/Worker" |
| 7 | import { Worker as NativeWorker } from "node:worker_threads" |
| 8 | |
| 9 | const fixture = new URL("./fixtures/worker-runner.ts", import.meta.url) |
| 10 | |
| 11 | describe("NodeWorkerRunner", () => { |
| 12 | it.live("sendUnsafe preserves reply payloads", () => |
| 13 | Effect.scoped( |
| 14 | Effect.gen(function*() { |
| 15 | const replies = yield* Queue.unbounded<unknown>() |
| 16 | const ready = yield* Deferred.make<void>() |
| 17 | const native = new NativeWorker(fixture, { execArgv: [] }) |
| 18 | const worker = yield* Worker.WorkerPlatform.use((platform) => platform.spawn<unknown, void>(0)).pipe( |
| 19 | Effect.provide(NodeWorker.layer(() => native)) |
| 20 | ) |
| 21 | yield* worker.run((reply) => Queue.offer(replies, reply), { |
| 22 | onSpawn: Deferred.succeed(ready, undefined) |
| 23 | }).pipe(Effect.forkScoped) |
| 24 | yield* Deferred.await(ready) |
| 25 | |
| 26 | yield* worker.send(undefined) |
| 27 | |
| 28 | assert.deepStrictEqual(yield* Queue.take(replies), { value: 42 }) |
| 29 | }) |
| 30 | ), { timeout: 20_000 }) |
| 31 | }) |