1/**
2 * Parent-side Node.js support for Effect workers.
3 *
4 * `layerPlatform` installs the `WorkerPlatform` used by a Node program that
5 * owns workers. It supports both `node:worker_threads` workers and IPC-enabled
6 * child processes, routing messages through Effect's worker protocol. `layer`
7 * combines that platform with a `Spawner` callback, and the platform asks
8 * workers to close on scope finalization before forcefully terminating them on
9 * timeout.
10 *
11 * @since 4.0.0
12 */
13import * as Deferred from "effect/Deferred"
14import * as Effect from "effect/Effect"
15import * as Exit from "effect/Exit"
16import * as Layer from "effect/Layer"
17import * as Scope from "effect/Scope"
18import * as Worker from "effect/workers/Worker"
19import { WorkerError, WorkerReceiveError } from "effect/workers/WorkerError"
20import type * as ChildProcess from "node:child_process"
21import type * as WorkerThreads from "node:worker_threads"
22
23/**
24 * Provides the Node `WorkerPlatform` for `worker_threads` workers and child
25 * process workers, wiring messages, errors, and exits into Effect workers and
26 * terminating the worker if graceful shutdown times out.
27 *
28 * @category layers
29 * @since 4.0.0
30 */
31export const layerPlatform: Layer.Layer<Worker.WorkerPlatform> = Layer.succeed(Worker.WorkerPlatform)(
32 Worker.makePlatform<WorkerThreads.Worker | ChildProcess.ChildProcess>()({
33 setup({ scope, worker }) {
34 const exitDeferred = Deferred.makeUnsafe<void, WorkerError>()
35 const thing = "postMessage" in worker ?
36 {
37 postMessage(msg: any, t?: any) {
38 worker.postMessage(msg, t)
39 },
40 kill: () => worker.terminate(),
41 worker
42 } :
43 {
44 postMessage(msg: any, _?: any) {
45 worker.send(msg)
46 },
47 kill: () => worker.kill("SIGKILL"),
48 worker
49 }
50 worker.on("exit", () => {
51 Deferred.doneUnsafe(exitDeferred, Exit.void)
52 })
53 return Effect.as(
54 Scope.addFinalizer(
55 scope,
56 Effect.suspend(() => {
57 thing.postMessage([1])
58 return Deferred.await(exitDeferred)
59 }).pipe(
60 Effect.timeout(5000),
61 Effect.catchCause(() => Effect.promise(() => Promise.resolve(thing.kill())))
62 )
63 ),
64 thing
65 )
66 },
67 listen({ deferred, emit, port }) {
68 port.worker.on("message", (message) => {
69 emit(message)
70 })
71 port.worker.on("messageerror", (cause) => {
72 Deferred.doneUnsafe(
73 deferred,
74 new WorkerError({
75 reason: new WorkerReceiveError({
76 message: "An messageerror event was emitted",
77 cause
78 })
79 })
80 )
81 })
82 port.worker.on("error", (cause) => {
83 Deferred.doneUnsafe(
84 deferred,
85 new WorkerError({
86 reason: new WorkerReceiveError({
87 message: "An error event was emitted",
88 cause
89 })
90 })
91 )
92 })
93 port.worker.on("exit", (code) => {
94 Deferred.doneUnsafe(
95 deferred,
96 new WorkerError({
97 reason: new WorkerReceiveError({
98 message: "The worker has exited with code: " + code
99 })
100 })
101 )
102 })
103 return Effect.void
104 }
105 })
106)
107
108/**
109 * Provides the Node `WorkerPlatform` together with a `Worker.Spawner` created
110 * from the supplied worker or child-process spawning function.
111 *
112 * @category layers
113 * @since 4.0.0
114 */
115export const layer = (
116 spawn: (id: number) => WorkerThreads.Worker | ChildProcess.ChildProcess
117): Layer.Layer<Worker.WorkerPlatform | Worker.Spawner> =>
118 Layer.merge(
119 Worker.layerSpawner(spawn),
120 layerPlatform
121 )