1/**
2 * Parent-side browser platform for Effect workers.
3 *
4 * `layerPlatform` provides the `WorkerPlatform` used to communicate with a
5 * browser `Worker`, `SharedWorker`, or `MessagePort` through Effect's worker
6 * protocol. `layer` combines that platform with a `Spawner` built from a
7 * callback that creates or returns the worker endpoint for each worker id.
8 *
9 * @since 4.0.0
10 */
11import * as Deferred from "effect/Deferred"
12import * as Effect from "effect/Effect"
13import * as Layer from "effect/Layer"
14import * as Scope from "effect/Scope"
15import * as Worker from "effect/workers/Worker"
16import { WorkerError, WorkerReceiveError } from "effect/workers/WorkerError"
17
18/**
19 * Creates browser worker layers by combining the default `WorkerPlatform` with a spawner for `Worker`, `SharedWorker`, or `MessagePort` instances.
20 *
21 * **When to use**
22 *
23 * Use when you need both the browser `WorkerPlatform` and a `Spawner` from one
24 * layer.
25 *
26 * **Details**
27 *
28 * The `spawn` callback receives the numeric worker id and may return a
29 * `Worker`, `SharedWorker`, or `MessagePort`.
30 *
31 * **Gotchas**
32 *
33 * Scope finalization sends the worker close protocol over the port. Dedicated
34 * workers created by `spawn` are not terminated by this layer.
35 *
36 * @see {@link layerPlatform} for providing only the browser worker platform
37 *
38 * @category layers
39 * @since 4.0.0
40 */
41export const layer = (
42 spawn: (id: number) => Worker | SharedWorker | MessagePort
43): Layer.Layer<Worker.WorkerPlatform | Worker.Spawner> =>
44 Layer.merge(
45 layerPlatform,
46 Worker.layerSpawner(spawn)
47 )
48
49/**
50 * Layer that provides the browser `WorkerPlatform` for `Worker`, `SharedWorker`, and `MessagePort` communication.
51 *
52 * @category layers
53 * @since 4.0.0
54 */
55export const layerPlatform: Layer.Layer<Worker.WorkerPlatform> = Layer.succeed(Worker.WorkerPlatform)(
56 Worker.makePlatform<globalThis.SharedWorker | globalThis.Worker | MessagePort>()({
57 setup({ scope, worker }) {
58 const port = "port" in worker ? worker.port : worker
59 return Effect.as(
60 Scope.addFinalizer(
61 scope,
62 Effect.sync(() => {
63 port.postMessage([1])
64 })
65 ),
66 port
67 )
68 },
69 listen({ deferred, emit, port, scope }) {
70 function onMessage(event: MessageEvent) {
71 emit(event.data)
72 }
73 function onError(event: ErrorEvent) {
74 Deferred.doneUnsafe(
75 deferred,
76 new WorkerError({
77 reason: new WorkerReceiveError({
78 message: "An error event was emitter",
79 cause: event.error ?? event.message
80 })
81 })
82 )
83 }
84 port.addEventListener("message", onMessage as any)
85 port.addEventListener("error", onError as any)
86 if ("start" in port) {
87 port.start()
88 }
89 return Scope.addFinalizer(
90 scope,
91 Effect.sync(() => {
92 port.removeEventListener("message", onMessage as any)
93 port.removeEventListener("error", onError as any)
94 })
95 )
96 }
97 })
98)