1/**
2 * Parent-side Deno platform for Effect workers.
3 *
4 * @since 4.0.0
5 */
6import * as Deferred from "effect/Deferred"
7import * as Effect from "effect/Effect"
8import * as Layer from "effect/Layer"
9import * as Scope from "effect/Scope"
10import * as Worker from "effect/workers/Worker"
11import { WorkerError, WorkerReceiveError } from "effect/workers/WorkerError"
12
13/**
14 * Creates Deno worker layers by combining the default worker platform with a spawner.
15 *
16 * @category layers
17 * @since 4.0.0
18 */
19export const layer = (
20 spawn: (id: number) => globalThis.Worker | MessagePort
21): Layer.Layer<Worker.WorkerPlatform | Worker.Spawner> =>
22 Layer.merge(
23 layerPlatform,
24 Worker.layerSpawner(spawn)
25 )
26
27/**
28 * Layer that provides the Deno worker platform.
29 *
30 * @category layers
31 * @since 4.0.0
32 */
33export const layerPlatform: Layer.Layer<Worker.WorkerPlatform> = Layer.succeed(Worker.WorkerPlatform)(
34 Worker.makePlatform<globalThis.Worker | MessagePort>()({
35 setup({ scope, worker }) {
36 return Effect.as(
37 Scope.addFinalizer(
38 scope,
39 Effect.sync(() => {
40 worker.postMessage([1])
41 })
42 ),
43 worker
44 )
45 },
46 listen({ deferred, emit, port, scope }) {
47 function onMessage(event: MessageEvent) {
48 emit(event.data)
49 }
50 function onError(event: ErrorEvent) {
51 Deferred.doneUnsafe(
52 deferred,
53 new WorkerError({
54 reason: new WorkerReceiveError({
55 message: "An error event was emitted",
56 cause: event.error ?? event.message
57 })
58 })
59 )
60 }
61 port.addEventListener("message", onMessage as any)
62 port.addEventListener("error", onError as any)
63 if ("start" in port) {
64 port.start()
65 }
66 return Scope.addFinalizer(
67 scope,
68 Effect.sync(() => {
69 port.removeEventListener("message", onMessage as any)
70 port.removeEventListener("error", onError as any)
71 })
72 )
73 }
74 })
75)