Implement BlobWorker

Author: andywer

src/master/implementation.browser.ts

@@ -1,6 +1,6 @@
 // tslint:disable max-classes-per-file
 
-import { ThreadsWorkerOptions, WorkerImplementation } from "../types/master"
+import { ImplementationExport, ThreadsWorkerOptions } from "../types/master"
 import { getBundleURL } from "./get-bundle-url.browser"
 
 export const defaultPoolSize = typeof navigator !== "undefined" && navigator.hardwareConcurrency
@@ -17,7 +17,7 @@ function createSourceBlobURL(code: string): string {
   return URL.createObjectURL(blob)
 }
 
-function selectWorkerImplementation(): typeof WorkerImplementation {
+function selectWorkerImplementation(): ImplementationExport {
   if (typeof Worker === "undefined") {
     // Might happen on Safari, for instance
     // The idea is to only fail if the constructor is actually used
@@ -28,7 +28,7 @@ function selectWorkerImplementation(): typeof WorkerImplementation {
     } as any
   }
 
-  return class WebWorker extends Worker {
+  class WebWorker extends Worker {
     constructor(url: string | URL, options?: ThreadsWorkerOptions) {
       if (typeof url === "string" && options && options._baseURL) {
         url = new URL(url, options._baseURL)
@@ -44,11 +44,28 @@ function selectWorkerImplementation(): typeof WorkerImplementation {
       super(url, options)
     }
   }
+
+  class BlobWorker extends WebWorker {
+    constructor(blob: Blob, options?: ThreadsWorkerOptions) {
+      const url = window.URL.createObjectURL(blob)
+      super(url, options)
+    }
+
+    public static fromText(source: string, options?: ThreadsWorkerOptions): WebWorker {
+      const blob = new window.Blob([source], { type: "text/javascript" })
+      return new BlobWorker(blob, options)
+    }
+  }
+
+  return {
+    blob: BlobWorker,
+    default: WebWorker
+  }
 }
 
-let implementation: typeof WorkerImplementation
+let implementation: ImplementationExport
 
-export function getWorkerImplementation(): typeof WorkerImplementation {
+export function getWorkerImplementation(): ImplementationExport {
   if (!implementation) {
     implementation = selectWorkerImplementation()
   }

src/master/implementation.node.ts

@@ -5,7 +5,11 @@ import getCallsites, { CallSite } from "callsites"
 import EventEmitter from "events"
 import { cpus } from 'os'
 import * as path from "path"
-import { ThreadsWorkerOptions, WorkerImplementation } from "../types/master"
+import {
+  ImplementationExport,
+  ThreadsWorkerOptions,
+  WorkerImplementation
+} from "../types/master"
 
 interface WorkerGlobalScope {
   addEventListener(eventName: string, listener: (event: Event) => void): void
@@ -84,7 +88,7 @@ function resolveScriptPath(scriptPath: string, baseURL?: string | undefined) {
   return workerFilePath
 }
 
-function initWorkerThreadsWorker(): typeof WorkerImplementation {
+function initWorkerThreadsWorker(): ImplementationExport {
   // Webpack hack
   const NativeWorker = typeof __non_webpack_require__ === "function"
     ? __non_webpack_require__("worker_threads").Worker
@@ -95,10 +99,16 @@ function initWorkerThreadsWorker(): typeof WorkerImplementation {
   class Worker extends NativeWorker {
     private mappedEventListeners: WeakMap<EventListener, EventListener>
 
-    constructor(scriptPath: string, options?: ThreadsWorkerOptions) {
-      const resolvedScriptPath = resolveScriptPath(scriptPath, (options || {})._baseURL)
+    constructor(scriptPath: string, options?: ThreadsWorkerOptions & { fromSource: boolean }) {
+      const resolvedScriptPath = options && options.fromSource
+        ? null
+        : resolveScriptPath(scriptPath, (options || {})._baseURL)
 
-      if (resolvedScriptPath.match(/\.tsx?$/i) && detectTsNode()) {
+      if (!resolvedScriptPath) {
+        // `options.fromSource` is true
+        const sourceCode = scriptPath
+        super(sourceCode, { ...options, eval: true })
+      } else if (resolvedScriptPath.match(/\.tsx?$/i) && detectTsNode()) {
         super(createTsNodeModule(resolvedScriptPath), { ...options, eval: true })
       } else if (resolvedScriptPath.match(/\.asar[\/\\]/)) {
         // See <https://github.com/andywer/threads-plugin/issues/17>
@@ -138,25 +148,44 @@ function initWorkerThreadsWorker(): typeof WorkerImplementation {
   process.on("SIGINT", () => terminateWorkersAndMaster())
   process.on("SIGTERM", () => terminateWorkersAndMaster())
 
-  return Worker as any
+  class BlobWorker extends Worker {
+    constructor(blob: Uint8Array, options?: ThreadsWorkerOptions) {
+      super(Buffer.from(blob).toString("utf-8"), { ...options, fromSource: true })
+    }
+
+    public static fromText(source: string, options?: ThreadsWorkerOptions): WorkerImplementation {
+      return new Worker(source, { ...options, fromSource: true }) as any
+    }
+  }
+
+  return {
+    blob: BlobWorker as any,
+    default: Worker as any
+  }
 }
 
-function initTinyWorker(): typeof WorkerImplementation {
+function initTinyWorker(): ImplementationExport {
   const TinyWorker = require("tiny-worker")
 
   let allWorkers: Array<typeof TinyWorker> = []
 
   class Worker extends TinyWorker {
     private emitter: EventEmitter
 
-    constructor(scriptPath: string) {
+    constructor(scriptPath: string, options?: ThreadsWorkerOptions & { fromSource?: boolean }) {
       // Need to apply a work-around for Windows or it will choke upon the absolute path
       // (`Error [ERR_INVALID_PROTOCOL]: Protocol 'c:' not supported`)
-      const resolvedScriptPath = process.platform === "win32"
-        ? `file:///${resolveScriptPath(scriptPath).replace(/\\/g, "/")}`
-        : resolveScriptPath(scriptPath)
-
-      if (resolvedScriptPath.match(/\.tsx?$/i) && detectTsNode()) {
+      const resolvedScriptPath = options && options.fromSource
+        ? null
+        : process.platform === "win32"
+          ? `file:///${resolveScriptPath(scriptPath).replace(/\\/g, "/")}`
+          : resolveScriptPath(scriptPath)
+
+      if (!resolvedScriptPath) {
+        // `options.fromSource` is true
+        const sourceCode = scriptPath
+        super(new Function(sourceCode), [], { esm: true })
+      } else if (resolvedScriptPath.match(/\.tsx?$/i) && detectTsNode()) {
         super(new Function(createTsNodeModule(resolveScriptPath(scriptPath))), [], { esm: true })
       } else if (resolvedScriptPath.match(/\.asar[\/\\]/)) {
         // See <https://github.com/andywer/threads-plugin/issues/17>
@@ -171,12 +200,15 @@ function initTinyWorker(): typeof WorkerImplementation {
       this.onerror = (error: Error) => this.emitter.emit("error", error)
       this.onmessage = (message: MessageEvent) => this.emitter.emit("message", message)
     }
+
     public addEventListener(eventName: WorkerEventName, listener: EventListener) {
       this.emitter.addListener(eventName, listener)
     }
+
     public removeEventListener(eventName: WorkerEventName, listener: EventListener) {
       this.emitter.removeListener(eventName, listener)
     }
+
     public terminate() {
       allWorkers = allWorkers.filter(worker => worker !== this)
       return super.terminate()
@@ -197,13 +229,26 @@ function initTinyWorker(): typeof WorkerImplementation {
   process.on("SIGINT", () => terminateWorkersAndMaster())
   process.on("SIGTERM", () => terminateWorkersAndMaster())
 
-  return Worker as any
+  class BlobWorker extends Worker {
+    constructor(blob: Uint8Array, options?: ThreadsWorkerOptions) {
+      super(Buffer.from(blob).toString("utf-8"), { ...options, fromSource: true })
+    }
+
+    public static fromText(source: string, options?: ThreadsWorkerOptions): WorkerImplementation {
+      return new Worker(source, { ...options, fromSource: true }) as any
+    }
+  }
+
+  return {
+    blob: BlobWorker as any,
+    default: Worker as any
+  }
 }
 
-let implementation: typeof WorkerImplementation
+let implementation: ImplementationExport
 let isTinyWorker: boolean
 
-function selectWorkerImplementation(): typeof WorkerImplementation {
+function selectWorkerImplementation(): ImplementationExport {
   try {
     isTinyWorker = false
     return initWorkerThreadsWorker()
@@ -215,7 +260,7 @@ function selectWorkerImplementation(): typeof WorkerImplementation {
   }
 }
 
-export function getWorkerImplementation(): typeof WorkerImplementation {
+export function getWorkerImplementation(): ImplementationExport {
   if (!implementation) {
     implementation = selectWorkerImplementation()
   }

src/master/index.ts

@@ -1,3 +1,5 @@
+// tslint:disable no-duplicate-imports
+import type { BlobWorker as BlobWorkerClass } from "../types/master"
 import { Worker as WorkerType } from "../types/master"
 import { getWorkerImplementation, isWorkerRuntime } from "./implementation"
 
@@ -7,7 +9,11 @@ export { spawn } from "./spawn"
 export { Thread } from "./thread"
 export { isWorkerRuntime }
 
+export type BlobWorker = typeof BlobWorkerClass
 export type Worker = WorkerType
 
+/** Separate class to spawn workers from source code blobs or strings. */
+export const BlobWorker = getWorkerImplementation().blob
+
 /** Worker implementation. Either web worker or a node.js Worker class. */
-export const Worker = getWorkerImplementation()
+export const Worker = getWorkerImplementation().default

src/types/master.ts

@@ -1,4 +1,5 @@
 /// <reference lib="dom" />
+// tslint:disable max-classes-per-file
 
 // Cannot use `compilerOptions.esModuleInterop` and default import syntax
 // See <https://github.com/microsoft/TypeScript/issues/28009>
@@ -89,6 +90,17 @@ export declare class WorkerImplementation extends EventTarget implements Worker
   public terminate(): void
 }
 
+/** Class to spawn workers from a blob or source string. */
+export declare class BlobWorker extends WorkerImplementation {
+  constructor(blob: Blob, options?: ThreadsWorkerOptions)
+  public static fromText(source: string, options?: ThreadsWorkerOptions): WorkerImplementation
+}
+
+export interface ImplementationExport {
+  blob: typeof BlobWorker
+  default: typeof WorkerImplementation
+}
+
 /** Event as emitted by worker thread. Subscribe to using `Thread.events(thread)`. */
 export enum WorkerEventType {
   internalError = "internalError",