| 1 | import * as DenoWorkerRunner from "@effect/platform-deno/DenoWorkerRunner" |
| 2 | import { assert, describe, it } from "@effect/vitest" |
| 3 | import * as Effect from "effect/Effect" |
| 4 | import * as Fiber from "effect/Fiber" |
| 5 | import * as Queue from "effect/Queue" |
| 6 | |
| 7 | type Listener = EventListenerOrEventListenerObject |
| 8 | |
| 9 | const makePort = () => { |
| 10 | const listeners = new Map<string, Set<Listener>>() |
| 11 | const added: Array<readonly [string, Listener]> = [] |
| 12 | const removed: Array<readonly [string, Listener]> = [] |
| 13 | const port = { |
| 14 | addEventListener(type: string, listener: Listener | null) { |
| 15 | if (listener === null) return |
| 16 | added.push([type, listener]) |
| 17 | const eventListeners = listeners.get(type) ?? new Set() |
| 18 | eventListeners.add(listener) |
| 19 | listeners.set(type, eventListeners) |
| 20 | }, |
| 21 | removeEventListener(type: string, listener: Listener | null) { |
| 22 | if (listener === null) return |
| 23 | removed.push([type, listener]) |
| 24 | listeners.get(type)?.delete(listener) |
| 25 | }, |
| 26 | postMessage() {}, |
| 27 | start() {}, |
| 28 | close() {} |
| 29 | } as unknown as MessagePort |
| 30 | |
| 31 | return { |
| 32 | added, |
| 33 | removed, |
| 34 | port, |
| 35 | emit(type: string, data: unknown) { |
| 36 | const event = { data } as MessageEvent |
| 37 | for (const listener of listeners.get(type) ?? []) { |
| 38 | if (typeof listener === "function") { |
| 39 | listener(event) |
| 40 | } else { |
| 41 | listener.handleEvent(event) |
| 42 | } |
| 43 | } |
| 44 | } |
| 45 | } |
| 46 | } |
| 47 | |
| 48 | const makeSharedWorker = () => { |
| 49 | const worker = { |
| 50 | onconnect: null as ((event: MessageEvent) => void) | null, |
| 51 | addEventListener() {}, |
| 52 | removeEventListener() {}, |
| 53 | close() {} |
| 54 | } |
| 55 | return { |
| 56 | worker: worker as unknown as MessagePort, |
| 57 | connect(port: MessagePort) { |
| 58 | worker.onconnect?.({ ports: [port] } as unknown as MessageEvent) |
| 59 | } |
| 60 | } |
| 61 | } |
| 62 | |
| 63 | describe("DenoWorkerRunner", () => { |
| 64 | it.effect("removes the registered messageerror listener", () => |
| 65 | Effect.gen(function*() { |
| 66 | const fake = makePort() |
| 67 | const runner = yield* DenoWorkerRunner.make(fake.port).start() |
| 68 | const fiber = yield* Effect.forkChild(runner.run<void, never, never>(() => {})) |
| 69 | yield* Effect.yieldNow |
| 70 | |
| 71 | fake.emit("message", [1]) |
| 72 | yield* Fiber.join(fiber) |
| 73 | |
| 74 | const added = fake.added.find(([type]) => type === "messageerror") |
| 75 | const removed = fake.removed.find(([type]) => type === "messageerror") |
| 76 | assert.isDefined(added) |
| 77 | assert.isDefined(removed) |
| 78 | assert.strictEqual(removed[1], added[1]) |
| 79 | })) |
| 80 | |
| 81 | it.effect("emits disconnects for closed SharedWorker ports", () => |
| 82 | Effect.gen(function*() { |
| 83 | const sharedWorker = makeSharedWorker() |
| 84 | const first = makePort() |
| 85 | const second = makePort() |
| 86 | const runner = yield* DenoWorkerRunner.make(sharedWorker.worker).start() |
| 87 | const fiber = yield* Effect.forkChild(runner.run<void, never, never>(() => {})) |
| 88 | yield* Effect.yieldNow |
| 89 | |
| 90 | sharedWorker.connect(first.port) |
| 91 | sharedWorker.connect(second.port) |
| 92 | first.emit("message", [1]) |
| 93 | |
| 94 | const disconnects = runner.disconnects |
| 95 | assert.isDefined(disconnects) |
| 96 | assert.strictEqual(yield* Queue.take(disconnects), 0) |
| 97 | |
| 98 | second.emit("message", [1]) |
| 99 | yield* Fiber.join(fiber) |
| 100 | })) |
| 101 | }) |