1/**
2 * Parent-side worker support for Bun applications.
3 *
4 * `layerPlatform` provides the `WorkerPlatform` used to communicate with
5 * `globalThis.Worker` instances through Effect's worker protocol. `layer`
6 * combines that platform with a `Spawner` built from a callback that creates a
7 * worker for each worker id. The platform forwards worker messages and errors,
8 * asks workers to close on scope finalization, and terminates them if graceful
9 * shutdown times out.
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, WorkerUnknownError } from "effect/workers/WorkerError"
20
21/**
22 * Provides the Bun `WorkerPlatform` together with a `Worker.Spawner` created
23 * from the supplied worker spawning function.
24 *
25 * @category layers
26 * @since 4.0.0
27 */
28export const layer = (
29 spawn: (id: number) => globalThis.Worker
30): Layer.Layer<Worker.WorkerPlatform | Worker.Spawner> =>
31 Layer.merge(
32 layerPlatform,
33 Layer.succeed(Worker.Spawner)(spawn)
34 )
35
36/**
37 * Provides the Bun `WorkerPlatform`, wiring worker messages and errors into
38 * Effect workers and requesting graceful worker shutdown during scope
39 * finalization before terminating on timeout.
40 *
41 * @category layers
42 * @since 4.0.0
43 */
44export const layerPlatform = Layer.succeed(Worker.WorkerPlatform)(
45 Worker.makePlatform<globalThis.Worker>()({
46 setup({ scope, worker }) {
47 const closeDeferred = Deferred.makeUnsafe<void>()
48 worker.addEventListener("close", () => {
49 Deferred.doneUnsafe(closeDeferred, Exit.void)
50 })
51 return Effect.as(
52 Scope.addFinalizer(
53 scope,
54 Effect.suspend(() => {
55 worker.postMessage([1])
56 return Deferred.await(closeDeferred)
57 }).pipe(
58 Effect.interruptible,
59 Effect.timeout(5000),
60 Effect.catchCause(() => Effect.sync(() => worker.terminate()))
61 )
62 ),
63 worker
64 )
65 },
66 listen({ deferred, emit, port, scope }) {
67 function onMessage(event: MessageEvent) {
68 emit(event.data)
69 }
70 function onError(event: ErrorEvent) {
71 Deferred.doneUnsafe(
72 deferred,
73 new WorkerError({
74 reason: new WorkerUnknownError({
75 message: "An error event was emitted",
76 cause: event.error ?? event.message
77 })
78 })
79 )
80 }
81 port.addEventListener("message", onMessage)
82 port.addEventListener("error", onError)
83 return Scope.addFinalizer(
84 scope,
85 Effect.sync(() => {
86 port.removeEventListener("message", onMessage)
87 port.removeEventListener("error", onError)
88 })
89 )
90 }
91 })
92)