fork-openpgpjs/src/worker/async_proxy.js

// GPG4Browsers - An OpenPGP implementation in javascript
// Copyright (C) 2011 Recurity Labs GmbH
//
// This library is free software; you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public
// License as published by the Free Software Foundation; either
// version 3.0 of the License, or (at your option) any later version.
//
// This library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
// Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public
// License along with this library; if not, write to the Free Software
// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA

/**
 * @fileoverview Provides functions for maintaining browser workers
 * @see module:openpgp.initWorker
 * @see module:openpgp.getWorker
 * @see module:openpgp.destroyWorker
 * @see module:worker/worker
 * @requires util
 * @requires config
 * @requires crypto
 * @requires packet
 * @module worker/async_proxy
 */

import util from '../util.js';
import config from '../config';
import crypto from '../crypto';
import packet from '../packet';


/**
 * Creates a new async proxy
 * @constructor
 */
function AsyncProxy() {}

/**
 * Initializes the proxy and loads the web worker
 * @param {String} path            The path to the worker or 'openpgp.worker.js' by default
 * @param {Number} n               number of workers to initialize if path given
 * @param {Object} config          config The worker configuration
 * @param {Array<Object>} worker   alternative to path parameter: web worker initialized with 'openpgp.worker.js'
 */
AsyncProxy.prototype.init = async function({ path = 'openpgp.worker.js', n = 1, workers = [], config } = {}) {
  /**
   * Message handling
   */
  const handleMessage = workerId => event => {
    const msg = event.data;
    switch (msg.event) {
      case 'method-return':
        if (msg.err) {
          // fail
          const err = new Error(msg.err);
          // add worker stack
          err.workerStack = msg.stack;
          this.tasks[msg.id].reject(err);
        } else {
          // success
          this.tasks[msg.id].resolve(msg.data);
        }
        delete this.tasks[msg.id];
        this.workers[workerId].requests--;
        break;
      case 'request-seed':
        this.seedRandom(workerId, msg.amount);
        break;
      default:
        throw new Error('Unknown Worker Event.');
    }
  };

  if (workers.length) {
    this.workers = workers;
  }
  else {
    this.workers = [];
    while (this.workers.length < n) {
      this.workers.push(new Worker(path));
    }
  }

  // Cannot rely on task order being maintained, use object keyed by request ID to track tasks
  this.tasks = {};
  this.currentID = 0;

  let workerId = 0;
  await Promise.all(this.workers.map(worker => new Promise(async (resolve, reject) => {
    worker.requests = 0;
    worker.onmessage = handleMessage(workerId++);
    worker.onerror = e => {
      reject(new Error('Unhandled error in openpgp worker: ' + e.message + ' (' + e.filename + ':' + e.lineno + ')'));
    };

    await this.callWorker(worker, 'configure', config);
    resolve();
  })));
};

/**
 * Get new request ID
 * @returns {integer}          New unique request ID
*/
AsyncProxy.prototype.getID = function() {
  return this.currentID++;
};

/**
 * Send message to worker with random data
 * @param  {Integer} size Number of bytes to send
 * @async
 */
AsyncProxy.prototype.seedRandom = async function(workerId, size) {
  const buf = await crypto.random.getRandomBytes(size);
  this.workers[workerId].postMessage({ event: 'seed-random', buf }, util.getTransferables(buf, true));
};

/**
 * Clear key caches
 * @async
 */
AsyncProxy.prototype.clearKeyCache = async function() {
  await Promise.all(this.workers.map(worker => this.callWorker(worker, 'clear-key-cache')));
};

/**
 * Terminates the workers
 */
AsyncProxy.prototype.terminate = function() {
  this.workers.forEach(worker => {
    worker.terminate();
  });
};

/**
 * Generic proxy function that handles all commands from the public api.
 * @param  {String} method    the public api function to be delegated to the worker thread
 * @param  {Object} options   the api function's options
 * @returns {Promise}          see the corresponding public api functions for their return types
 * @async
 */
AsyncProxy.prototype.delegate = async function (method, options) {
  const requests = this.workers.map(worker => worker.requests);
  const minRequests = Math.min(...requests);
  let workerId = 0;
  for (; workerId < this.workers.length; workerId++) {
    if (this.workers[workerId].requests === minRequests) {
      break;
    }
  }

  const data = { options: packet.clone.clonePackets(options) };
  const transferables = util.getTransferables(data, config.zero_copy);
  const worker = this.workers[workerId];
  worker.requests++;
  const result = await this.callWorker(worker, method, data.options, transferables);
  return packet.clone.parseClonedPackets(util.restoreStreams(result, options.streaming), method);
};

AsyncProxy.prototype.callWorker = function(worker, method, options, transferables) {
  return new Promise((resolve, reject) => {
    const id = this.getID();

    worker.postMessage({ id, method, options }, transferables);

    this.tasks[id] = { resolve, reject };
  });
};

export default AsyncProxy;