// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license. /* eslint-disable @typescript-eslint/no-explicit-any */ import * as dispatch from "./dispatch.ts"; import { sendAsync, sendSync } from "./dispatch_json.ts"; import { log, createResolvable, Resolvable } from "./util.ts"; import { TextDecoder, TextEncoder } from "./text_encoding.ts"; import { window } from "./window.ts"; import { blobURLMap } from "./url.ts"; import { blobBytesWeakMap } from "./blob.ts"; import { EventTarget } from "./event_target.ts"; const encoder = new TextEncoder(); const decoder = new TextDecoder(); export function encodeMessage(data: any): Uint8Array { const dataJson = JSON.stringify(data); return encoder.encode(dataJson); } export function decodeMessage(dataIntArray: Uint8Array): any { const dataJson = decoder.decode(dataIntArray); return JSON.parse(dataJson); } function createWorker( specifier: string, includeDenoNamespace: boolean, hasSourceCode: boolean, sourceCode: Uint8Array ): { id: number; loaded: boolean } { return sendSync(dispatch.OP_CREATE_WORKER, { specifier, includeDenoNamespace, hasSourceCode, sourceCode: new TextDecoder().decode(sourceCode) }); } async function hostGetWorkerLoaded(id: number): Promise { return await sendAsync(dispatch.OP_HOST_GET_WORKER_LOADED, { id }); } async function hostPollWorker(id: number): Promise { return await sendAsync(dispatch.OP_HOST_POLL_WORKER, { id }); } function hostCloseWorker(id: number): void { sendSync(dispatch.OP_HOST_CLOSE_WORKER, { id }); } function hostResumeWorker(id: number): void { sendSync(dispatch.OP_HOST_RESUME_WORKER, { id }); } function hostPostMessage(id: number, data: any): void { const dataIntArray = encodeMessage(data); sendSync(dispatch.OP_HOST_POST_MESSAGE, { id }, dataIntArray); } async function hostGetMessage(id: number): Promise { const res = await sendAsync(dispatch.OP_HOST_GET_MESSAGE, { id }); if (res.data != null) { return decodeMessage(new Uint8Array(res.data)); } else { return null; } } // Stuff for workers export const onmessage: (e: { data: any }) => void = (): void => {}; export const onerror: (e: { data: any }) => void = (): void => {}; export function postMessage(data: any): void { const dataIntArray = encodeMessage(data); sendSync(dispatch.OP_WORKER_POST_MESSAGE, {}, dataIntArray); } export async function getMessage(): Promise { log("getMessage"); const res = await sendAsync(dispatch.OP_WORKER_GET_MESSAGE); if (res.data != null) { return decodeMessage(new Uint8Array(res.data)); } else { return null; } } export let isClosing = false; export function workerClose(): void { isClosing = true; } export async function workerMain(): Promise { log("workerMain"); while (!isClosing) { const data = await getMessage(); if (data == null) { log("workerMain got null message. quitting."); break; } let result: void | Promise; const event = { data }; try { result = window.onmessage(event); if (result && "then" in result) { await result; } if (!window["onmessage"]) { break; } } catch (e) { if (window["onerror"]) { const result = window.onerror( e.message, e.fileName, e.lineNumber, e.columnNumber, e ); if (result === true) { continue; } } throw e; } } } export interface Worker { onerror?: (e: any) => void; onmessage?: (e: { data: any }) => void; onmessageerror?: () => void; postMessage(data: any): void; // TODO(bartlomieju): remove this closed: Promise; } // TODO(kevinkassimo): Maybe implement reasonable web worker options? // eslint-disable-next-line @typescript-eslint/no-empty-interface export interface WorkerOptions {} /** Extended Deno Worker initialization options. * `noDenoNamespace` hides global `window.Deno` namespace for * spawned worker and nested workers spawned by it (default: false). */ export interface DenoWorkerOptions extends WorkerOptions { noDenoNamespace?: boolean; } export class WorkerImpl extends EventTarget implements Worker { private readonly id: number; private isClosing = false; private messageBuffer: any[] = []; private ready = false; private readonly isClosedPromise: Resolvable; public onerror?: (e: any) => void; public onmessage?: (data: any) => void; public onmessageerror?: () => void; constructor(specifier: string, options?: DenoWorkerOptions) { super(); let hasSourceCode = false; let sourceCode = new Uint8Array(); let includeDenoNamespace = true; if (options && options.noDenoNamespace) { includeDenoNamespace = false; } // Handle blob URL. if (specifier.startsWith("blob:")) { hasSourceCode = true; const b = blobURLMap.get(specifier); if (!b) { throw new Error("No Blob associated with the given URL is found"); } const blobBytes = blobBytesWeakMap.get(b!); if (!blobBytes) { throw new Error("Invalid Blob"); } sourceCode = blobBytes!; } const { id, loaded } = createWorker( specifier, includeDenoNamespace, hasSourceCode, sourceCode ); this.id = id; this.ready = loaded; this.isClosedPromise = createResolvable(); this.poll(); } get closed(): Promise { return this.isClosedPromise; } private handleError(e: any): boolean { const event = new window.Event("error", { cancelable: true }); event.message = e.message; event.lineNumber = e.lineNumber ? e.lineNumber + 1 : null; event.columnNumber = e.columnNumber ? e.columnNumber + 1 : null; event.fileName = e.fileName; event.error = null; let handled = false; if (this.onerror) { this.onerror(event); if (event.defaultPrevented) { handled = true; } } return handled; } async poll(): Promise { // If worker has not been immediately executed // then let's await it's readiness if (!this.ready) { const result = await hostGetWorkerLoaded(this.id); if (result.error) { if (!this.handleError(result.error)) { throw new Error(result.error.message); } return; } } // drain messages for (const data of this.messageBuffer) { hostPostMessage(this.id, data); } this.messageBuffer = []; this.ready = true; this.run(); while (true) { const result = await hostPollWorker(this.id); if (result.error) { if (!this.handleError(result.error)) { throw Error(result.error.message); } else { hostResumeWorker(this.id); } } else { this.isClosing = true; hostCloseWorker(this.id); this.isClosedPromise.resolve(); break; } } } postMessage(data: any): void { if (!this.ready) { this.messageBuffer.push(data); return; } hostPostMessage(this.id, data); } private async run(): Promise { while (!this.isClosing) { const data = await hostGetMessage(this.id); if (data == null) { log("worker got null message. quitting."); break; } if (this.onmessage) { const event = { data }; this.onmessage(event); } } } }