feat(@embark/blockchain): Restart Ethereum via command

Add support for `service blockchain on/off` in the console.

Add service on/off command for each process started by, ie `service blockchain on/off` and `service whisper on/off`.

`service blockchain off` - Kills the blockchain process, stops the web3 provider, and shuts down the proxy (if used). In the case of `embark blockchain`, the entire process is exited.

`service blockchain on` - Starts the blockchain process *as a child process of the main embark process* then starts the web3 provider. This happens regardless of whether or not it was initially started with `embark blockchain` (see known issues).

## Known issues
1. If the blockchain process was started with `embark blockchain`, and then the blockchain process is killed with the `service blockchain off` command, when the blockchain process is restarted with `service blockchain on`, it will be restarted as a child process of the main embark process. It may be possible to allow for the blockchain process to be restarted in the same process it was originally started in, however this will take more development effort and can be handled in a different PR.
2. Parity has not been tested as it is currently not working in the master branch.
3. This PR adds a generic registration of commands for each process, ie `service whisper on/off`, however the only supported command is the `service blockchain on/off` command. Further support for other commands can be handled in separate PRs.

packages/embark/src/lib/core/processes/processManager.js

@@ -1,5 +1,5 @@
 const ProcessState = {
-  Unstarted: 'unstarted',
+  Stopped: 'stopped',
   Starting: 'starting',
   Running: 'running',
   Stopping: 'stopping'
@@ -12,51 +12,50 @@ class ProcessManager {
     this.plugins = options.plugins;
     this.processes = {};
     this.servicesState = {};
+    this.plugin = this.plugins.createPlugin('processManager', {});
 
     this.events.on("servicesState", (servicesState) => {
       this.servicesState = servicesState;
     });
 
-    this._registerAsPlugin();
+    this._registerApiCalls();
     this._registerEvents();
   }
 
-  _registerAsPlugin() {
-    const self = this;
-    self.plugin = this.plugins.createPlugin('processManager', {});
+  _registerApiCalls() {
 
-    self.plugin.registerAPICall(
+    this.plugin.registerAPICall(
       'get',
       '/embark-api/services',
       (req, res) => {
-        res.send(this._sevicesForApi(this.servicesState));
+        res.send(this._servicesForApi(this.servicesState));
       }
     );
 
-    self.plugin.registerAPICall(
+    this.plugin.registerAPICall(
       'ws',
       '/embark-api/services',
       (ws, _res) => {
         this.events.on('servicesState', (servicesState) => {
-          ws.send(JSON.stringify(this._sevicesForApi(servicesState)), () => undefined);
+          ws.send(JSON.stringify(this._servicesForApi(servicesState)), () => undefined);
         });
       }
     );
 
-    self.plugin.registerAPICall(
+    this.plugin.registerAPICall(
       'get',
       '/embark-api/processes',
       (req, res) => {
         const formatter = (acc, processName) => {
-          acc.push({state: self.processes[processName].state, name: processName});
+          acc.push({state: this.processes[processName].state, name: processName});
           return acc;
         };
-        res.send(Object.keys(self.processes).reduce(formatter, []));
+        res.send(Object.keys(this.processes).reduce(formatter, []));
       }
     );
   }
 
-  _sevicesForApi(servicesState) {
+  _servicesForApi(servicesState) {
     let processList = [];
     for (let serviceName in servicesState) {
       let service = servicesState[serviceName];
@@ -78,19 +77,44 @@ class ProcessManager {
 
       this.processes[name] = {
         name: name,
-        state: ProcessState.Unstarted,
+        state: ProcessState.Stopped,
         cb: launchFn || cb,
         stopFn: stopFn || function noop () {}
       };
+
+      this.plugin.registerConsoleCommand({
+        description: __(`Starts/stops the ${name} process`),
+        matches: [`service ${name} on`, `service ${name} off`],
+        usage: `service ${name} on/off`,
+        process: (cmd, callback) => {
+          const enable = cmd.trim().endsWith('on');
+          this.logger.info(`${enable ? 'Starting' :  'Stopping'} the ${name} process...`);
+          if(enable) {
+            return this.events.request("processes:launch", name, (err) => {
+              if (err) this.logger.info(err); // writes to embark's console
+              const process = self.processes[name];
+              if(process && process.afterLaunchFn) {
+                process.afterLaunchFn.call(process.afterLaunchFn, err);
+              }
+              callback(err, `${name} process started.`); // passes a message back to cockpit console
+            });
+          }
+          this.events.request("processes:stop", name, (err) => {
+            if (err) this.logger.info(err); // writes to embark's console
+            callback(err, `${name} process stopped.`); // passes a message back to cockpit console
+          });
+        }
+      });
     });
 
     self.events.setCommandHandler('processes:launch', (name, cb) => {
       cb = cb || function noop() {};
       let process = self.processes[name];
-      if (process.state !== ProcessState.Unstarted) {
-        return cb();
+      if (process.state !== ProcessState.Stopped) {
+        return cb(__(`The ${name} process is already ${process.state.toLowerCase()}.`));
       }
       process.state = ProcessState.Starting;
+      if(!process.afterLaunchFn) process.afterLaunchFn = cb;
       process.cb.apply(process.cb, [
         (...args) => {
           process.state = ProcessState.Running;
@@ -103,12 +127,12 @@ class ProcessManager {
       let process = self.processes[name];
       cb = cb || function noop() {};
       if (process.state !== ProcessState.Running) {
-        return cb();
+        return cb(__(`The ${name} process is already ${process.state.toLowerCase()}.`));
       }
       process.state = ProcessState.Stopping;
       process.stopFn.apply(process.stopFn, [
         (...args) => {
-          process.state = ProcessState.Unstarted;
+          process.state = ProcessState.Stopped;
           cb.apply(cb, args);
         }
       ]);

packages/embark/src/lib/modules/blockchain_connector/index.js

@@ -242,14 +242,7 @@ class BlockchainConnector {
     this.events.on('check:wentOffline:Ethereum', () => {
       this.logger.warn('Ethereum went offline: stopping web3 provider...');
       this.provider.stop();
-
-      // once the node goes back online, we can restart the provider
-      this.events.once('check:backOnline:Ethereum', () => {
-        this.logger.warn('Ethereum back online: starting web3 provider...');
-        this.provider.startWeb3Provider(() => {
-          this.logger.warn('web3 provider restarted after ethereum node came back online');
-        });
-      });
+      this.isWeb3Ready = false;
     });
 
     this.events.on('blockchain:contracts:event', this._saveEvent.bind(this));
@@ -786,7 +779,12 @@ class BlockchainConnector {
 
   subscribeToContractEvents(callback) {
     this.contractsSubscriptions.forEach((eventEmitter) => {
-      eventEmitter.unsubscribe();
+      const reqMgr = eventEmitter.options.requestManager;
+      // attempting an eth_unsubscribe when not connected throws an
+      // "connection not open on send()" error
+      if(reqMgr && reqMgr.provider && reqMgr.provider.connected) {
+        eventEmitter.unsubscribe();
+      }
     });
     this.contractsSubscriptions = [];
     this.events.request("contracts:list", (_err, contractsList) => {

packages/embark/src/lib/modules/blockchain_process/blockchain.js

@@ -142,6 +142,10 @@ Blockchain.prototype.initStandaloneProcess = function () {
           if (this.ipc.connected) {
             logQueue.forEach(message => { this.ipc.request('blockchain:log', message); });
             logQueue = [];
+            this.ipc.client.on('process:blockchain:stop', () => {
+              this.kill();
+              process.exit(0);
+            });
           }
         });
       }

packages/embark/src/lib/modules/blockchain_process/blockchainProcessLauncher.js

@@ -69,6 +69,14 @@ class BlockchainProcessLauncher {
     });
   }
 
+  stopBlockchainNode(cb) {
+    if(this.blockchainProcess) {
+      this.events.on(constants.blockchain.blockchainExit, cb);
+      this.blockchainProcess.exitCallback = () => {}; // don't show error message as the process was killed on purpose
+      this.blockchainProcess.send('exit');
+    }
+  }
+
 }
 
 module.exports = BlockchainProcessLauncher;

packages/embark/src/lib/modules/blockchain_process/index.js

@@ -15,24 +15,27 @@ class BlockchainModule {
     this.isDev = options.isDev;
     this.ipc = options.ipc;
     this.client = options.client;
+    this.blockchainProcess =  null;
 
     this.registerBlockchainProcess();
   }
 
   registerBlockchainProcess() {
-    const self = this;
-    this.events.request('processes:register', 'blockchain', (cb) => {
-      self.assertNodeConnection(true, (connected) => {
-        if (connected) return cb();
-        self.startBlockchainNode(cb);
-        this.listenToCommands();
-        this.registerConsoleCommands();
-      });
+    this.events.request('processes:register', 'blockchain', {
+      launchFn: (cb) => {
+        this.assertNodeConnection(true, (connected) => {
+          if (connected) return cb();
+          this.startBlockchainNode(cb);
+          this.listenToCommands();
+          this.registerConsoleCommands();
+        });
+      },
+      stopFn: (cb) => { this.stopBlockchainNode(cb); }
     });
 
     if (!this.ipc.isServer()) return;
-    self.ipc.on('blockchain:node', (_message, cb) => {
-      cb(null, utils.buildUrlFromConfig(self.contractsConfig.deployment));
+    this.ipc.on('blockchain:node', (_message, cb) => {
+      cb(null, utils.buildUrlFromConfig(this.contractsConfig.deployment));
     });
   }
 
@@ -97,7 +100,7 @@ class BlockchainModule {
   startBlockchainNode(callback) {
     const self = this;
 
-    let blockchainProcess = new BlockchainProcessLauncher({
+    this.blockchainProcess = new BlockchainProcessLauncher({
       events: self.events,
       logger: self.logger,
       normalizeInput: utils.normalizeInput,
@@ -108,17 +111,38 @@ class BlockchainModule {
       embark: this.embark
     });
 
-    blockchainProcess.startBlockchainNode();
-    self.events.once(constants.blockchain.blockchainReady, () => {
-      self.assertNodeConnection(true, (connected) => {
+    this.blockchainProcess.startBlockchainNode();
+    this.events.once(constants.blockchain.blockchainReady, () => {
+      this.assertNodeConnection(true, (connected) => {
         if (!connected) {
           return callback(__('Blockchain process is ready, but still cannot connect to it. Check your host, port and protocol in your contracts config'));
         }
+        this.events.removeListener(constants.blockchain.blockchainExit, callback);
         callback();
       });
     });
-    self.events.once(constants.blockchain.blockchainExit, () => {
-      callback();
+    this.events.once(constants.blockchain.blockchainExit, callback);
+  }
+
+  stopBlockchainNode(cb) {
+    const message = __(`The blockchain process has been stopped. It can be restarted by running ${"service blockchain on".bold} in the Embark console.`);
+    if (this.ipc.isServer()) {
+      if(!this.ipc.connected) {
+        this.ipc.connect(() => {
+          this.ipc.broadcast('process:blockchain:stop');
+          this.logger.info(message);
+        });
+      }
+      else this.ipc.broadcast('process:blockchain:stop');
+    }
+
+    if(!this.blockchainProcess) {
+      return cb();
+    }
+
+    this.blockchainProcess.stopBlockchainNode(() => {
+      this.logger.info(message);
+      cb();
     });
   }
 

packages/embark/src/lib/modules/console/index.ts

@@ -107,7 +107,11 @@ class Console {
           // Avoid HTML injection in the Cockpit
           response = escapeHtml(response);
         }
-        return res.send({ result: response });
+        const jsonResponse = {result: response};
+        if (res.headersSent) {
+          return res.end(jsonResponse);
+        }
+        return res.send(jsonResponse);
       });
     });
   }