mirror of
https://github.com/denoland/deno.git
synced 2024-12-22 23:34:47 -05:00
79b3bc05d6
* establish basic event loop for workers * make "self.close()" inside worker * remove "runWorkerMessageLoop() - instead manually call global function in Rust when message arrives. This is done in preparation for structured clone * refactor "WorkerChannel" and use distinct structs for internal and external channels; "WorkerChannelsInternal" and "WorkerHandle" * move "State.worker_channels_internal" to "Worker.internal_channels" * add "WorkerEvent" enum for child->host communication; currently "Message(Buf)" and "Error(ErrBox)" variants are supported * add tests for nested workers * add tests for worker throwing error on startup
198 lines
4.8 KiB
TypeScript
198 lines
4.8 KiB
TypeScript
// 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 } from "./util.ts";
|
|
import { TextDecoder, TextEncoder } from "./text_encoding.ts";
|
|
/*
|
|
import { blobURLMap } from "./url.ts";
|
|
import { blobBytesWeakMap } from "./blob.ts";
|
|
*/
|
|
import { Event } from "./event.ts";
|
|
import { EventTarget } from "./event_target.ts";
|
|
|
|
const encoder = new TextEncoder();
|
|
const decoder = new TextDecoder();
|
|
|
|
function encodeMessage(data: any): Uint8Array {
|
|
const dataJson = JSON.stringify(data);
|
|
return encoder.encode(dataJson);
|
|
}
|
|
|
|
function decodeMessage(dataIntArray: Uint8Array): any {
|
|
const dataJson = decoder.decode(dataIntArray);
|
|
return JSON.parse(dataJson);
|
|
}
|
|
|
|
function createWorker(
|
|
specifier: string,
|
|
hasSourceCode: boolean,
|
|
sourceCode: Uint8Array,
|
|
name?: string
|
|
): { id: number } {
|
|
return sendSync(dispatch.OP_CREATE_WORKER, {
|
|
specifier,
|
|
hasSourceCode,
|
|
sourceCode: new TextDecoder().decode(sourceCode),
|
|
name
|
|
});
|
|
}
|
|
|
|
function hostTerminateWorker(id: number): void {
|
|
sendSync(dispatch.OP_HOST_TERMINATE_WORKER, { id });
|
|
}
|
|
|
|
function hostPostMessage(id: number, data: any): void {
|
|
const dataIntArray = encodeMessage(data);
|
|
sendSync(dispatch.OP_HOST_POST_MESSAGE, { id }, dataIntArray);
|
|
}
|
|
|
|
interface WorkerEvent {
|
|
event: "error" | "msg" | "close";
|
|
data?: any;
|
|
error?: any;
|
|
}
|
|
|
|
async function hostGetMessage(id: number): Promise<any> {
|
|
return await sendAsync(dispatch.OP_HOST_GET_MESSAGE, { id });
|
|
}
|
|
|
|
export interface Worker {
|
|
onerror?: (e: any) => void;
|
|
onmessage?: (e: { data: any }) => void;
|
|
onmessageerror?: () => void;
|
|
postMessage(data: any): void;
|
|
terminate(): void;
|
|
}
|
|
|
|
export interface WorkerOptions {
|
|
type?: "classic" | "module";
|
|
name?: string;
|
|
}
|
|
|
|
export class WorkerImpl extends EventTarget implements Worker {
|
|
private readonly id: number;
|
|
private isClosing = false;
|
|
public onerror?: (e: any) => void;
|
|
public onmessage?: (data: any) => void;
|
|
public onmessageerror?: () => void;
|
|
private name: string;
|
|
private terminated = false;
|
|
|
|
constructor(specifier: string, options?: WorkerOptions) {
|
|
super();
|
|
|
|
let type = "classic";
|
|
|
|
if (options?.type) {
|
|
type = options.type;
|
|
}
|
|
|
|
if (type !== "module") {
|
|
throw new Error(
|
|
'Not yet implemented: only "module" type workers are supported'
|
|
);
|
|
}
|
|
|
|
this.name = options?.name ?? "unknown";
|
|
const hasSourceCode = false;
|
|
const sourceCode = new Uint8Array();
|
|
|
|
/* TODO(bartlomieju):
|
|
// 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 } = createWorker(
|
|
specifier,
|
|
hasSourceCode,
|
|
sourceCode,
|
|
options?.name
|
|
);
|
|
this.id = id;
|
|
this.poll();
|
|
}
|
|
|
|
private handleError(e: any): boolean {
|
|
// TODO: this is being handled in a type unsafe way, it should be type safe
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const event = new Event("error", { cancelable: true }) as any;
|
|
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<void> {
|
|
while (!this.terminated) {
|
|
const event = await hostGetMessage(this.id);
|
|
|
|
// If terminate was called then we ignore all messages
|
|
if (this.terminated) {
|
|
return;
|
|
}
|
|
|
|
const type = event.type;
|
|
|
|
if (type === "msg") {
|
|
if (this.onmessage) {
|
|
const message = decodeMessage(new Uint8Array(event.data));
|
|
this.onmessage({ data: message });
|
|
}
|
|
continue;
|
|
}
|
|
|
|
if (type === "error") {
|
|
if (!this.handleError(event.error)) {
|
|
throw Error(event.error.message);
|
|
}
|
|
continue;
|
|
}
|
|
|
|
if (type === "close") {
|
|
log(`Host got "close" message from worker: ${this.name}`);
|
|
this.terminated = true;
|
|
return;
|
|
}
|
|
|
|
throw new Error(`Unknown worker event: "${type}"`);
|
|
}
|
|
}
|
|
|
|
postMessage(data: any): void {
|
|
if (this.terminated) {
|
|
return;
|
|
}
|
|
|
|
hostPostMessage(this.id, data);
|
|
}
|
|
|
|
terminate(): void {
|
|
if (!this.terminated) {
|
|
this.terminated = true;
|
|
hostTerminateWorker(this.id);
|
|
}
|
|
}
|
|
}
|