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