1/**
2 * Runner-side Deno platform for Effect worker handlers using the standard `MessagePort` API.
3 *
4 * @since 4.0.0
5 */
6import * as Cause from "effect/Cause"
7import * as Deferred from "effect/Deferred"
8import * as Effect from "effect/Effect"
9import * as Exit from "effect/Exit"
10import * as Fiber from "effect/Fiber"
11import { identity } from "effect/Function"
12import * as Layer from "effect/Layer"
13import * as Queue from "effect/Queue"
14import * as Scope from "effect/Scope"
15import { WorkerError, WorkerReceiveError } from "effect/workers/WorkerError"
16import * as WorkerRunner from "effect/workers/WorkerRunner"
17
18const cachedPorts = new Set<MessagePort>()
19function globalHandleConnect(event: MessageEvent) {
20 cachedPorts.add(event.ports[0])
21}
22if (typeof self !== "undefined" && "onconnect" in self) {
23 self.onconnect = globalHandleConnect
24}
25
26/**
27 * Creates a worker runner platform over a `MessagePort`.
28 *
29 * @category constructors
30 * @since 4.0.0
31 */
32export const make = (self: MessagePort): WorkerRunner.WorkerRunnerPlatform["Service"] => ({
33 start: Effect.fnUntraced(function*<O = unknown, I = unknown>() {
34 const disconnects = yield* Queue.make<number>()
35 let currentPortId = 0
36
37 const ports = new Map<number, readonly [MessagePort, Scope.Closeable]>()
38 const sendUnsafe = (portId: number, message: O, transfer?: ReadonlyArray<unknown>) =>
39 (ports.get(portId)?.[0] ?? self).postMessage([1, message], {
40 transfer: transfer as any
41 })
42 const send = (portId: number, message: O, transfer?: ReadonlyArray<unknown>) =>
43 Effect.sync(() => sendUnsafe(portId, message, transfer))
44
45 const run = <A, E, R>(
46 handler: (portId: number, message: I) => Effect.Effect<A, E, R> | void
47 ) =>
48 Effect.scopedWith(Effect.fnUntraced(function*(scope) {
49 const closeLatch = Deferred.makeUnsafe<void, WorkerError>()
50 const trackFiber = Fiber.runIn(scope)
51 const services = yield* Effect.context<R>()
52 const runFork = Effect.runForkWith(services)
53 const onExit = (exit: Exit.Exit<any, E>) => {
54 if (exit._tag === "Failure" && !Cause.hasInterruptsOnly(exit.cause)) {
55 runFork(Effect.logError("unhandled error in worker", exit.cause))
56 }
57 }
58
59 function onMessage(portId: number) {
60 return function(event: MessageEvent) {
61 const message = event.data as WorkerRunner.PlatformMessage<I>
62 if (message[0] === 0) {
63 const result = handler(portId, message[1])
64 if (Effect.isEffect(result)) {
65 const fiber = runFork(result)
66 fiber.addObserver(onExit)
67 trackFiber(fiber)
68 }
69 } else {
70 const port = ports.get(portId)
71 if (!port) {
72 return
73 } else if (ports.size === 1) {
74 // let the last port close with the outer scope
75 return Deferred.doneUnsafe(closeLatch, Exit.void)
76 }
77 ports.delete(portId)
78 Queue.offerUnsafe(disconnects, portId)
79 Effect.runFork(Scope.close(port[1], Exit.void))
80 }
81 }
82 }
83 function onMessageError(error: MessageEvent) {
84 Deferred.doneUnsafe(
85 closeLatch,
86 new WorkerError({
87 reason: new WorkerReceiveError({
88 message: "An messageerror event was emitted",
89 cause: error.data
90 })
91 })
92 )
93 }
94 function onError(error: any) {
95 Deferred.doneUnsafe(
96 closeLatch,
97 new WorkerError({
98 reason: new WorkerReceiveError({
99 message: "An error event was emitted",
100 cause: error.data
101 })
102 })
103 )
104 }
105 function handlePort(port: MessagePort) {
106 const portScope = Scope.forkUnsafe(scope)
107 const portId = currentPortId++
108 ports.set(portId, [port, portScope])
109 const onMsg = onMessage(portId)
110 port.addEventListener("message", onMsg)
111 port.addEventListener("messageerror", onMessageError)
112 if ("start" in port) {
113 port.start()
114 }
115 port.postMessage([0])
116 Effect.runSync(Scope.addFinalizer(
117 portScope,
118 Effect.sync(() => {
119 port.removeEventListener("message", onMsg)
120 port.removeEventListener("messageerror", onMessageError)
121 port.close()
122 })
123 ))
124 }
125 self.addEventListener("error", onError)
126 let prevOnConnect: unknown | undefined
127 if ("onconnect" in self) {
128 prevOnConnect = self.onconnect
129 self.onconnect = function(event: MessageEvent) {
130 const port = (event as MessageEvent).ports[0]
131 handlePort(port)
132 }
133 for (const port of cachedPorts) {
134 handlePort(port)
135 }
136 cachedPorts.clear()
137 } else {
138 handlePort(self as any)
139 }
140 yield* Scope.addFinalizer(
141 scope,
142 Effect.sync(() => {
143 self.removeEventListener("error", onError)
144 if ("onconnect" in self) {
145 self.close()
146 self.onconnect = prevOnConnect
147 }
148 })
149 )
150
151 yield* Deferred.await(closeLatch)
152 }))
153
154 return identity<WorkerRunner.WorkerRunner<O, I>>({ run, send, sendUnsafe, disconnects })
155 }) as any
156})
157
158/**
159 * Layer that provides the worker runner platform using the global worker scope.
160 *
161 * @category layers
162 * @since 4.0.0
163 */
164export const layer: Layer.Layer<WorkerRunner.WorkerRunnerPlatform> = Layer.sync(WorkerRunner.WorkerRunnerPlatform)(() =>
165 make(self as unknown as MessagePort)
166)
167
168/**
169 * Layer that provides the worker runner platform using the supplied `MessagePort`.
170 *
171 * @category layers
172 * @since 4.0.0
173 */
174export const layerMessagePort = (port: MessagePort): Layer.Layer<WorkerRunner.WorkerRunnerPlatform> =>
175 Layer.succeed(WorkerRunner.WorkerRunnerPlatform)(make(port))