For issue #44, make the Zipper use MessageChannel and not have a hard dependency on Worker

Author: codedread

archive/compress.js

@@ -11,7 +11,7 @@
 // NOTE: THIS IS A VERY HACKY WORK-IN-PROGRESS! THE API IS NOT FROZEN! USE AT YOUR OWN RISK!
 
 /**
- * @typedef FileInfo An object that is sent to the worker to represent a file to zip.
+ * @typedef FileInfo An object that is sent to the implementation to represent a file to zip.
  * @property {string} fileName The name of the file. TODO: Includes the path?
  * @property {number} lastModTime The number of ms since the Unix epoch (1970-01-01 at midnight).
  * @property {ArrayBuffer} fileData The bytes of the file.
@@ -42,7 +42,6 @@ export const ZipCompressionMethod = {
 
 /**
  * @typedef CompressorOptions
- * @property {string} pathToBitJS A string indicating where the BitJS files are located.
  * @property {ZipCompressionMethod} zipCompressionMethod
  * @property {DeflateCompressionMethod=} deflateCompressionMethod Only present if
  *     zipCompressionMethod is set to DEFLATE.
@@ -60,36 +59,52 @@ export const CompressStatus = {
   ERROR: 'error',
 };
 
+/**
+ * Connects the MessagePort to the compressor implementation (e.g. zip.js). If Workers exist
+ * (e.g. web browsers or deno), imports the implementation inside a Web Worker. Otherwise, it
+ * dynamically imports the implementation inside the current JS context.
+ * The MessagePort is used for communication between host and implementation.
+ * @param {string} implFilename The compressor implementation filename relative to this path
+ *     (e.g. './zip.js').
+ * @param {MessagePort} implPort The MessagePort to connect to the compressor implementation.
+ * @returns {Promise<void>} The Promise resolves once the ports are connected.
+ */
+const connectPortFn = async (implFilename, implPort) => {
+  if (typeof Worker === 'undefined') {
+    return import(`${implFilename}`).then(implModule => implModule.connect(implPort));
+  }
+  return new Promise((resolve, reject) => {
+    const workerScriptPath = new URL(`./webworker-wrapper.js`, import.meta.url).href;
+    const worker = new Worker(workerScriptPath, { type: 'module' });
+    worker.postMessage({ implSrc: implFilename }, [ implPort ]);
+    resolve();
+  });
+};
+
 /**
  * A thing that zips files.
  * NOTE: THIS IS A VERY HACKY WORK-IN-PROGRESS! THE API IS NOT FROZEN! USE AT YOUR OWN RISK!
  * TODO: Make a streaming / event-driven API.
  */
 export class Zipper {
+  /**
+   * The client-side port that sends messages to, and receives messages from the
+   * decompressor implementation.
+   * @type {MessagePort}
+   * @private
+   */
+  port_;
+
   /**
    * @param {CompressorOptions} options
    */
   constructor(options) {
-    /**
-     * The path to the BitJS files.
-     * @type {string}
-     * @private
-     */
-    this.pathToBitJS = options.pathToBitJS || '/';
-
     /**
      * @type {ZipCompressionMethod}
      * @private
      */
     this.zipCompressionMethod = options.zipCompressionMethod || ZipCompressionMethod.STORE;
 
-    /**
-     * Private web worker initialized during start().
-     * @type {Worker}
-     * @private
-     */
-    this.worker_ = null;
-
     /**
      * @type {CompressStatus}
      * @private
@@ -109,14 +124,14 @@ export class Zipper {
    * @param {boolean} isLastFile 
    */
   appendFiles(files, isLastFile) {
-    if (!this.worker_) {
-      throw `Worker not initialized. Did you forget to call start() ?`;
+    if (!this.port_) {
+      throw `Port not initialized. Did you forget to call start() ?`;
     }
     if (![CompressStatus.READY, CompressStatus.WORKING].includes(this.compressState)) {
       throw `Zipper not in the right state: ${this.compressState}`;
     }
 
-    this.worker_.postMessage({ files, isLastFile });
+    this.port_.postMessage({ files, isLastFile });
   }
 
   /**
@@ -128,18 +143,19 @@ export class Zipper {
    * @returns {Promise<Uint8Array>} A Promise that will contain the entire zipped archive as an array
    *     of bytes.
    */
-  start(files, isLastFile) {
+  async start(files, isLastFile) {
+    const messageChannel = new MessageChannel();
+    this.port_ = messageChannel.port1;
+    await connectPortFn('./zip.js', messageChannel.port2);
     return new Promise((resolve, reject) => {
-      // TODO: Only use Worker if it exists (like decompress).
-      // TODO: Remove need for pathToBitJS (like decompress).
-      this.worker_ = new Worker(this.pathToBitJS + `archive/zip.js`);
-      this.worker_.onerror = (evt) => {
-        console.log('Worker error: message = ' + evt.message);
-        throw evt.message;
+      this.port_.onerror = (evt) => {
+        console.log('Impl error: message = ' + evt.message);
+        reject(evt.message);
       };
-      this.worker_.onmessage = (evt) => {
+
+      this.port_.onmessage = (evt) => {
         if (typeof evt.data == 'string') {
-          // Just log any strings the worker pumps our way.
+          // Just log any strings the implementation pumps our way.
           console.log(evt.data);
         } else {
           switch (evt.data.type) {

archive/decompress.js

@@ -13,6 +13,8 @@ import { UnarchiveAppendEvent, UnarchiveErrorEvent, UnarchiveEvent, UnarchiveEve
          UnarchiveProgressEvent, UnarchiveStartEvent } from './events.js';
 import { findMimeType } from '../file/sniffer.js';
 
+// Exported as a convenience (and also because this module used to contain these).
+// TODO(bitjs): Remove this export in a future release?
 export {
   UnarchiveAppendEvent,
   UnarchiveErrorEvent,
@@ -57,7 +59,7 @@ const connectPortFn = async (implFilename, implPort) => {
   }
 
   return new Promise((resolve, reject) => {
-    const workerScriptPath = new URL(`./unarchiver-webworker.js`, import.meta.url).href;
+    const workerScriptPath = new URL(`./webworker-wrapper.js`, import.meta.url).href;
     const worker = new Worker(workerScriptPath, { type: 'module' });
     worker.postMessage({ implSrc: implFilename }, [implPort]);
     resolve();

archive/unarchiver-webworker.js

@@ -0,0 +1,21 @@
+/**
+ * webworker-wrapper.js
+ *
+ * Licensed under the MIT License
+ *
+ * Copyright(c) 2023 Google Inc.
+ */
+
+/**
+ * A WebWorker wrapper for a decompress/compress implementation. Upon creation and being sent its
+ * first message, it dynamically imports the decompressor / compressor implementation and connects
+ * the message port. All other communication takes place over the MessageChannel.
+ */
+
+/** @type {MessagePort} */
+let implPort;
+
+onmessage = async (evt) => {
+  const module = await import(evt.data.implSrc);
+  module.connect(evt.ports[0]);
+};

archive/webworker-wrapper.js

@@ -11,37 +11,42 @@
  * DEFLATE format: http://tools.ietf.org/html/rfc1951
  */
 
-// This file expects to be invoked as a Worker (see onmessage below).
 import { ByteBuffer } from '../io/bytebuffer.js';
 
+/** @type {MessagePort} */
+let hostPort;
+
 /**
- * The client sends messages to this Worker containing files to archive in order. The client
- * indicates to the Worker when the last file has been sent to be compressed.
+ * The client sends a set of CompressFilesMessage to the MessagePort containing files to archive in
+ * order. The client sets isLastFile to true to indicate to the implementation when the last file
+ * has been sent to be compressed.
  *
- * The Worker emits an event to indicate compression has started: { type: 'start' }
- * As the files compress, bytes are sent back in order: { type: 'compress', bytes: Uint8Array }
- * After the last file compresses, the Worker indicates finish by: { type 'finish' }
+ * The impl posts an event to the port indicating compression has started: { type: 'start' }.
+ * As each file compresses, bytes are sent back in order: { type: 'compress', bytes: Uint8Array }.
+ * After the last file compresses, we indicate finish by: { type 'finish' }
  *
- * Clients should append the bytes to a single buffer in the order they were received.
+ * The client should append the bytes to a single buffer in the order they were received.
  */
 
+// TODO(bitjs): De-dupe this typedef and the one in compress.js.
 /**
- * @typedef FileInfo An object that is sent to this worker by the client to represent a file.
+ * @typedef FileInfo An object that is sent by the client to represent a file.
  * @property {string} fileName The name of this file. TODO: Includes the path?
  * @property {number} lastModTime The number of ms since the Unix epoch (1970-01-01 at midnight).
  * @property {Uint8Array} fileData The raw bytes of the file.
  */
 
-// TODO: Support DEFLATE.
-// TODO: Support options that can let client choose levels of compression/performance.
-
+// TODO(bitjs): Figure out where this typedef should live.
 /**
- * Ideally these constants should be defined in a common isomorphic ES module. Unfortunately, the
- * state of JavaScript is such that modules cannot be shared easily across browsers, worker threads,
- * NodeJS environments, etc yet. Thus, these constants, as well as logic that should be extracted to
- * common modules and shared with unzip.js are not yet easily possible.
+ * @typedef CompressFilesMessage A message the client sends to the implementation.
+ * @property {FileInfo[]} files A set of files to add to the zip file.
+ * @property {boolean} isLastFile Indicates this is the last set of files to add to the zip file.
  */
 
+// TODO: Support DEFLATE.
+// TODO: Support options that can let client choose levels of compression/performance.
+
+// TODO(bitjs): These constants should be defined in a common isomorphic ES module.
 const zLocalFileHeaderSignature = 0x04034b50;
 const zCentralFileHeaderSignature = 0x02014b50;
 const zEndOfCentralDirSignature = 0x06054b50;
@@ -231,39 +236,52 @@ function writeCentralFileDirectory() {
 }
 
 /**
- * @param {{data: {isLastFile?: boolean, files: FileInfo[]}}} evt The event for the Worker
- *     to process. It is an error to send any more events to the Worker if a previous event had
- *     isLastFile is set to true.
+ * @param {{data: CompressFilesMessage}} evt The event for the implementation to process. It is an
+ *     error to send any more events after a previous event had isLastFile is set to true.
  */
-onmessage = function(evt) {
+const onmessage = function(evt) {
   if (state === CompressorState.FINISHED) {
-    throw `The zip worker was sent a message after last file received.`;
+    throw `The zip implementation was sent a message after last file received.`;
   }
 
   if (state === CompressorState.NOT_STARTED) {
-    postMessage({ type: 'start' });
+    hostPort.postMessage({ type: 'start' });
   }
 
   state = CompressorState.COMPRESSING;
 
-  /** @type {FileInfo[]} */
-  const filesToCompress = evt.data.files;
+  const msg = evt.data;
+  const filesToCompress = msg.files;
   while (filesToCompress.length > 0) {
     const fileInfo = filesToCompress.shift();
     const fileBuffer = zipOneFile(fileInfo);
     filesCompressed.push(fileInfo);
     numBytesWritten += fileBuffer.data.byteLength;
-    this.postMessage({ type: 'compress', bytes: fileBuffer.data }, [ fileBuffer.data.buffer ]);
+    hostPort.postMessage({ type: 'compress', bytes: fileBuffer.data }, [ fileBuffer.data.buffer ]);
   }
 
   if (evt.data.isLastFile) {
     const centralBuffer = writeCentralFileDirectory();
     numBytesWritten += centralBuffer.data.byteLength;
-    this.postMessage({ type: 'compress', bytes: centralBuffer.data }, [ centralBuffer.data.buffer ]);
+    hostPort.postMessage({ type: 'compress', bytes: centralBuffer.data },
+        [ centralBuffer.data.buffer ]);
 
     state = CompressorState.FINISHED;
-    this.postMessage({ type: 'finish' });
+    hostPort.postMessage({ type: 'finish' });
   } else {
     state = CompressorState.WAITING;
   }
 };
+
+
+/**
+ * Connect the host to the zip implementation with the given MessagePort.
+ * @param {MessagePort} port
+ */
+export function connect(port) {
+  if (hostPort) {
+    throw `hostPort already connected`;
+  }
+  hostPort = port;
+  port.onmessage = onmessage;
+}