Merge pull request #417 from embark-framework/features/process-wrapper

Add Process Wrapper and Launcher

lib/constants.json

@@ -16,11 +16,18 @@
     "contractFilesChanged": "contractFilesChanged",
     "contractConfigChanged": "contractConfigChanged"
   },
+  "process": {
+    "log": "log",
+    "events": {
+      "on": "on",
+      "request": "request",
+      "response": "response"
+    }
+  },
   "pipeline": {
     "init": "init",
     "build": "build",
     "initiated": "initiated",
-    "built": "built",
-    "log": "log"
+    "built": "built"
   }
 }

lib/i18n/locales/en.json

@@ -96,6 +96,7 @@
 	"help": "help",
 	"quit": "quit",
 	"Error Compiling/Building contracts: ": "Error Compiling/Building contracts: ",
+	"file not found, creating it...": "file not found, creating it..."
 	"{{className}} has code associated to it but it's configured as an instanceOf {{parentContractName}}": "{{className}} has code associated to it but it's configured as an instanceOf {{parentContractName}}",
 	"downloading {{packageName}} {{version}}....": "downloading {{packageName}} {{version}}....",
 	"Swarm node is offline...": "Swarm node is offline...",

lib/pipeline/pipeline.js

@@ -1,6 +1,6 @@
 const fs = require('../core/fs.js');
 const async = require('async');
-const child_process = require('child_process');
+const ProcessLauncher = require('../process/processLauncher');
 const utils = require('../utils/utils.js');
 const constants = require('../constants');
 
@@ -69,22 +69,18 @@ class Pipeline {
                 // JS files
                 async.waterfall([
                   function runWebpack(next) {
-                    const webpackProcess = child_process.fork(utils.joinPath(__dirname, 'webpackProcess.js'));
+                    const webpackProcess = new ProcessLauncher({
+                      modulePath: utils.joinPath(__dirname, 'webpackProcess.js'),
+                      logger: self.logger,
+                      events: self.events,
+                      normalizeInput: self.normalizeInput
+                    });
                     webpackProcess.send({action: constants.pipeline.init, options: {}});
                     webpackProcess.send({action: constants.pipeline.build, file, importsList});
 
-                    webpackProcess.on('message', function (msg) {
-                      if (msg.result === constants.pipeline.built) {
+                    webpackProcess.subscribeTo('result', constants.pipeline.built, (msg) => {
                       webpackProcess.disconnect();
                       return next(msg.error);
-                      }
-
-                      if (msg.result === constants.pipeline.log) {
-                        if (self.logger[msg.type]) {
-                          return self.logger[msg.type](self.normalizeInput(msg.message));
-                        }
-                        self.logger.debug(self.normalizeInput(msg.message));
-                      }
                     });
                   },
 

lib/pipeline/webpackProcess.js

@@ -4,42 +4,11 @@ const utils = require('../utils/utils');
 const fs = require('../core/fs');
 const constants = require('../constants');
 const HardSourceWebpackPlugin = require('hard-source-webpack-plugin');
-
-// Override process.chdir so that we have a partial-implementation PWD for Windows
-const realChdir = process.chdir;
-process.chdir = (...args) => {
-  if (!process.env.PWD) {
-    process.env.PWD = process.cwd();
-  }
-  realChdir(...args);
-};
+const ProcessWrapper = require('../process/processWrapper');
 
 let webpackProcess;
 
-class WebpackProcess {
-  constructor(_options) {
-    this.interceptLogs();
-  }
-
-  interceptLogs() {
-    const context = {};
-    context.console = console;
-
-    context.console.log = this.log.bind(this, 'log');
-    context.console.warn = this.log.bind(this, 'warn');
-    context.console.info = this.log.bind(this, 'info');
-    context.console.debug = this.log.bind(this, 'debug');
-    context.console.trace = this.log.bind(this, 'trace');
-    context.console.dir = this.log.bind(this, 'dir');
-  }
-
-  log(type, ...messages) {
-    if (messages[0].indexOf('hard-source')) {
-      return;
-    }
-    process.send({result: constants.pipeline.log, message: messages, type});
-  }
-
+class WebpackProcess extends ProcessWrapper {
   build(file, importsList, callback) {
     const self = this;
     let realCwd;
@@ -150,7 +119,3 @@ process.on('message', (msg) => {
     });
   }
 });
-
-process.on('exit', () => {
-  process.exit(0);
-});

lib/process/eventsWrapper.js

@@ -0,0 +1,55 @@
+const uuid = require('uuid/v1');
+const constants = require('../constants');
+
+class Events {
+
+  /**
+   * Constructs an event wrapper for processes.
+   * Handles sending an event message to the parent process and waiting for its response
+   * No need to create an instance of eventsWrapper for your own process, just extend ProcessWrapper
+   *  Then, you an use `this.events.[on|request]` with the usual parameters you would use
+   */
+  constructor() {
+    this.subscribedEvents = {};
+    this.listenToParentProcess();
+  }
+
+  listenToParentProcess() {
+    process.on('message', (msg) => {
+      if (!msg.event || msg.event !== constants.process.events.response) {
+        return;
+      }
+      if (!this.subscribedEvents[msg.eventId]) {
+        return;
+      }
+      this.subscribedEvents[msg.eventId](msg.result);
+    });
+  }
+
+  sendEvent() {
+    const eventType = arguments[0];
+    const requestName = arguments[1];
+
+    let args = [].slice.call(arguments, 2);
+    const eventId = uuid();
+    this.subscribedEvents[eventId] = args[args.length - 1];
+    args = args.splice(0, args.length - 2);
+
+    process.send({
+      event: eventType,
+      requestName,
+      args,
+      eventId: eventId
+    });
+  }
+
+  on() {
+    this.sendEvent(constants.process.events.on, ...arguments);
+  }
+
+  request() {
+    this.sendEvent(constants.process.events.request, ...arguments);
+  }
+}
+
+module.exports = Events;

lib/process/processLauncher.js

@@ -0,0 +1,177 @@
+const child_process = require('child_process');
+const constants = require('../constants');
+
+
+class ProcessLauncher {
+
+  /**
+   * Constructor of ProcessLauncher. Forks the module and sets up the message handling
+   * @param {Object}    options   Options tp start the process
+   *        * modulePath      {String}    Absolute path to the module to fork
+   *        * logger          {Object}    Logger
+   *        * normalizeInput  {Function}  Function to normalize logs
+   *        * events          {Function}  Events Emitter instance
+   * @return {ProcessLauncher}    The ProcessLauncher instance
+   */
+  constructor(options) {
+    this.process = child_process.fork(options.modulePath);
+    this.logger = options.logger;
+    this.normalizeInput = options.normalizeInput;
+    this.events = options.events;
+
+    this.subscriptions = {};
+    this._subscribeToMessages();
+  }
+
+  // Subscribes to messages from the child process and delegates to the right methods
+  _subscribeToMessages() {
+    const self = this;
+    this.process.on('message', (msg) => {
+      if (msg.result === constants.process.log) {
+        return self._handleLog(msg);
+      }
+      if (msg.event) {
+        return this._handleEvent(msg);
+      }
+      this._checkSubscriptions(msg);
+    });
+  }
+
+  // Translates logs from the child process to the logger
+  _handleLog(msg) {
+    if (this.logger[msg.type]) {
+      return this.logger[msg.type](this.normalizeInput(msg.message));
+    }
+    this.logger.debug(this.normalizeInput(msg.message));
+  }
+
+  // Handle event calls from the child process
+  _handleEvent(msg) {
+    const self = this;
+    if (!self.events[msg.event]) {
+      self.logger.warn('Unknown event method called: ' + msg.event);
+      return;
+    }
+    if (!msg.args || !Array.isArray(msg.args)) {
+      msg.args = [];
+    }
+    // Add callback in the args
+    msg.args.push((result) => {
+      self.process.send({
+        event: constants.process.events.response,
+        result,
+        eventId: msg.eventId
+      });
+    });
+    self.events[msg.event](msg.requestName, ...msg.args);
+  }
+
+  // Looks at the subscriptions to see if there is a callback to call
+  _checkSubscriptions(msg) {
+    const messageKeys = Object.keys(msg);
+    const subscriptionsKeys = Object.keys(this.subscriptions);
+    let subscriptions;
+    let messageKey;
+    // Find if the message contains a key that we are subscribed to
+    messageKeys.some(_messageKey => {
+      return subscriptionsKeys.some(subscriptionKey => {
+        if (_messageKey === subscriptionKey) {
+          subscriptions = this.subscriptions[subscriptionKey];
+          messageKey = _messageKey;
+          return true;
+        }
+        return false;
+      });
+    });
+
+    if (subscriptions) {
+      let subscription;
+      // Find if we are subscribed to one of the values
+      subscriptions.some(sub => {
+        if (msg[messageKey] === sub.value) {
+          subscription = sub;
+          return true;
+        }
+        return false;
+      });
+
+      if (subscription) {
+        // We are subscribed to that message, call the callback
+        subscription.callback(msg);
+      }
+    }
+  }
+
+  /**
+   * Subscribe to a message using a key-value pair
+   * @param {String}    key       Message key to subscribe to
+   * @param {String}    value     Value that the above key must have for the callback to be called
+   * @param {Function}  callback  callback(response)
+   * @return {void}
+   */
+  subscribeTo(key, value, callback) {
+    if (this.subscriptions[key]) {
+      this.subscriptions[key].push({value, callback});
+      return;
+    }
+    this.subscriptions[key] = [{value, callback}];
+  }
+
+  /**
+   * Unsubscribes from a previously subscribed key-value pair (or key if no value)
+   * @param {String}  key     Message key to unsubscribe
+   * @param {String}  value   [Optional] Value of the key to unsubscribe
+ *                            If there is no value, unsubscribes from all the values of that key
+   * @return {void}
+   */
+  unsubscribeTo(key, value) {
+    if (!value) {
+      this.subscriptions[key] = [];
+    }
+    if (this.subscriptions[key]) {
+      this.subscriptions[key].filter((val, index) => {
+        if (val.value === value) {
+          this.subscriptions[key].splice(index, 1);
+        }
+      });
+    }
+  }
+
+  /**
+   * Unsubscribes from all subscriptions
+   * @return {void}
+   */
+  unsubscribeToAll() {
+    this.subscriptions = {};
+  }
+
+  /**
+   * Sends a message to the child process. Same as ChildProcess.send()
+   * @params {Object}   message     Message to send
+   * For other parameters, see:
+   *  https://nodejs.org/api/child_process.html#child_process_subprocess_send_message_sendhandle_options_callback
+   * @return {void}
+   */
+  send() {
+    this.process.send(...arguments);
+  }
+
+  /**
+   * Disconnects the child process. It will exit on its own
+   * @return {void}
+   */
+  disconnect() {
+    this.process.disconnect();
+  }
+
+  /**
+   * Kills the child process
+   *  https://nodejs.org/api/child_process.html#child_process_subprocess_kill_signal
+   * @return {void}
+   */
+  kill() {
+    this.process.kill(...arguments);
+  }
+}
+
+module.exports = ProcessLauncher;

lib/process/processWrapper.js

@@ -0,0 +1,54 @@
+const constants = require('../constants');
+const Events = require('./eventsWrapper');
+
+// Override process.chdir so that we have a partial-implementation PWD for Windows
+const realChdir = process.chdir;
+process.chdir = (...args) => {
+  if (!process.env.PWD) {
+    process.env.PWD = process.cwd();
+  }
+  realChdir(...args);
+};
+
+class ProcessWrapper {
+
+  /**
+   * Class from which process extend. Should not be instantiated alone.
+   * Manages the log interception so that all console.* get sent back to the parent process
+   * Also creates an Events instance. To use it, just do `this.events.[on|request]`
+   *
+   * @param {Options}     _options    Nothing for now
+   */
+  constructor(_options) {
+    this.interceptLogs();
+    this.events = new Events();
+  }
+
+  interceptLogs() {
+    const context = {};
+    context.console = console;
+
+    context.console.log = this._log.bind(this, 'log');
+    context.console.warn = this._log.bind(this, 'warn');
+    context.console.info = this._log.bind(this, 'info');
+    context.console.debug = this._log.bind(this, 'debug');
+    context.console.trace = this._log.bind(this, 'trace');
+    context.console.dir = this._log.bind(this, 'dir');
+  }
+
+  _log(type, ...messages) {
+    const isHardSource = messages.some(message => {
+      return (typeof message === 'string' && message.indexOf('hard-source') > -1);
+    });
+    if (isHardSource) {
+      return;
+    }
+    process.send({result: constants.process.log, message: messages, type});
+  }
+}
+
+process.on('exit', () => {
+  process.exit(0);
+});
+
+module.exports = ProcessWrapper;

package-lock.json

@@ -23,7 +23,7 @@
       "resolved": "https://registry.npmjs.org/@types/fs-extra/-/fs-extra-5.0.2.tgz",
       "integrity": "sha512-Q3FWsbdmkQd1ib11A4XNWQvRD//5KpPoGawA8aB2DR7pWKoW9XQv3+dGxD/Z1eVFze23Okdo27ZQytVFlweKvQ==",
       "requires": {
-        "@types/node": "10.0.9"
+        "@types/node": "10.1.0"
       }
     },
     "@types/lockfile": {
@@ -32,16 +32,16 @@
       "integrity": "sha512-pD6JuijPmrfi84qF3/TzGQ7zi0QIX+d7ZdetD6jUA6cp+IsCzAquXZfi5viesew+pfpOTIdAVKuh1SHA7KeKzg=="
     },
     "@types/node": {
-      "version": "10.0.9",
-      "resolved": "https://registry.npmjs.org/@types/node/-/node-10.0.9.tgz",
-      "integrity": "sha512-ekJ3mXJcJP+Nn5kC6eCmWPND/fHx/Ga12Lz0IJgTfGz1ge7OCIR5xcDY5tcYgnyM1kWsVDRH2bguxkGcNj39AQ=="
+      "version": "10.1.0",
+      "resolved": "https://registry.npmjs.org/@types/node/-/node-10.1.0.tgz",
+      "integrity": "sha512-sELcX/cJHwRp8kn4hYSvBxKGJ+ubl3MvS8VJQe5gz/sp7CifYxsiCxIJ35wMIYyGVMgfO2AzRa8UcVReAcJRlw=="
     },
     "@types/node-fetch": {
       "version": "1.6.9",
       "resolved": "https://registry.npmjs.org/@types/node-fetch/-/node-fetch-1.6.9.tgz",
       "integrity": "sha512-n2r6WLoY7+uuPT7pnEtKJCmPUGyJ+cbyBR8Avnu4+m1nzz7DwBVuyIvvlBzCZ/nrpC7rIgb3D6pNavL7rFEa9g==",
       "requires": {
-        "@types/node": "10.0.9"
+        "@types/node": "10.1.0"
       }
     },
     "@types/semver": {
@@ -54,7 +54,7 @@
       "resolved": "https://registry.npmjs.org/@types/tar/-/tar-4.0.0.tgz",
       "integrity": "sha512-YybbEHNngcHlIWVCYsoj7Oo1JU9JqONuAlt1LlTH/lmL8BMhbzdFUgReY87a05rY1j8mfK47Del+TCkaLAXwLw==",
       "requires": {
-        "@types/node": "10.0.9"
+        "@types/node": "10.1.0"
       }
     },
     "@types/url-join": {

package.json

@@ -67,6 +67,7 @@
     "underscore": "^1.9.0",
     "underscore.string": "^3.3.4",
     "url-loader": "^0.6.2",
+    "uuid": "^3.2.1",
     "viz.js": "^1.8.1",
     "web3": "1.0.0-beta.34",
     "web3-provider-engine": "^14.0.5",

test/processLauncher.js

@@ -0,0 +1,139 @@
+/*global describe, it, before, beforeEach*/
+const assert = require('assert');
+const sinon = require('sinon');
+const TestLogger = require('../lib/tests/test_logger.js');
+const ProcessLauncher = require('../lib/process/processLauncher');
+
+describe('ProcessWrapper', () => {
+  let processLauncher;
+
+  before(() => {
+    sinon.stub(ProcessLauncher.prototype, '_subscribeToMessages');
+    processLauncher = new ProcessLauncher({
+      logger: new TestLogger({})
+    });
+  });
+
+  describe('subscribeTo', () => {
+
+    beforeEach(() => {
+      processLauncher.subscriptions = {};
+    });
+
+    it('should create an array for the key value', function () {
+      processLauncher.subscribeTo('test', 'value', 'myCallback');
+      assert.deepEqual(processLauncher.subscriptions, {
+        "test": [
+          {
+            "callback": "myCallback",
+            "value": "value"
+          }
+        ]
+      });
+    });
+
+    it('should add another value to the key', () => {
+      processLauncher.subscribeTo('test', 'value', 'myCallback');
+      processLauncher.subscribeTo('test', 'value2', 'myCallback2');
+      assert.deepEqual(processLauncher.subscriptions, {
+        "test": [
+          {
+            "callback": "myCallback",
+            "value": "value"
+          },
+          {
+            "callback": "myCallback2",
+            "value": "value2"
+          }
+        ]
+      });
+    });
+  });
+
+  describe('unsubscribeTo', () => {
+    it('should remove the value for the key', () => {
+      processLauncher.subscriptions = {
+        "test": [
+          {
+            "callback": "myCallback",
+            "value": "value"
+          },
+          {
+            "callback": "myCallback2",
+            "value": "value2"
+          }
+        ]
+      };
+
+      processLauncher.unsubscribeTo('test', 'value2');
+      assert.deepEqual(processLauncher.subscriptions, {
+        "test": [
+          {
+            "callback": "myCallback",
+            "value": "value"
+          }
+        ]
+      });
+    });
+
+    it('should remove the whole key', () => {
+      processLauncher.subscriptions = {
+        "test": [
+          {
+            "callback": "myCallback",
+            "value": "value"
+          }
+        ]
+      };
+
+      processLauncher.unsubscribeTo('test');
+      assert.deepEqual(processLauncher.subscriptions, {test: []});
+    });
+  });
+
+  describe('unsubscribeToAll', () => {
+    it('clears every subscriptions', () => {
+      processLauncher.subscriptions = {
+        "test": [
+          {
+            "callback": "myCallback",
+            "value": "value"
+          }
+        ]
+      };
+
+      processLauncher.unsubscribeToAll();
+      assert.deepEqual(processLauncher.subscriptions, {});
+    });
+  });
+
+  describe('_checkSubscriptions', function () {
+    it('should not do anything if not in subscription', function () {
+      const callback = sinon.stub();
+      processLauncher.subscriptions = {
+        "test": [
+          {
+            "callback": callback,
+            "value": "value"
+          }
+        ]
+      };
+      processLauncher._checkSubscriptions({does: 'nothing', for: 'real'});
+      assert.strictEqual(callback.callCount, 0);
+    });
+
+    it('should call the callback', function () {
+      const callback = sinon.stub();
+      processLauncher.subscriptions = {
+        "test": [
+          {
+            "callback": callback,
+            "value": "value"
+          }
+        ]
+      };
+      processLauncher._checkSubscriptions({test: 'value'});
+      assert.strictEqual(callback.callCount, 1);
+    });
+  });
+});