removing async dep in server

nitsujlangston · nitsujlangston · commit 01e1526cbce6 · 2018-05-17T21:08:08.000-04:00
diff --git a/src/server.ts b/src/server.ts
@@ -8,41 +8,39 @@ import config from './config';
 import cluster = require('cluster');
 import app from './routes';
 import parseArgv from './utils/parseArgv';
-const async = require('async');
 let args = parseArgv([], ['DEBUG']);
 
-async.series(
-  [
-    Storage.start.bind(Storage),
-    Worker.start.bind(Worker),
-    async (cb: CallbackType) => {
-      let p2pServices = [] as Array<P2pService>;
-      for (let chain of Object.keys(config.chains)) {
-        for (let network of Object.keys(config.chains[chain])) {
-          const chainConfig = config.chains[chain][network];
-          const hasChainSource = chainConfig.chainSource !== undefined;
-          if (!hasChainSource || chainConfig.chainSource === 'p2p') {
-            let p2pServiceConfig = Object.assign(
-              config.chains[chain][network],
-              { chain, network }
-            );
-            p2pServices.push(new P2pService(p2pServiceConfig));
-          }
-        }
+const startServices = async () => {
+  await Storage.start({});
+  await Worker.start();
+
+  // TODO this needs to move to a static p2pService method
+  let p2pServices = [] as Array<P2pService>;
+  for (let chain of Object.keys(config.chains)) {
+    for (let network of Object.keys(config.chains[chain])) {
+      const chainConfig = config.chains[chain][network];
+      const hasChainSource = chainConfig.chainSource !== undefined;
+      if (!hasChainSource || chainConfig.chainSource === 'p2p') {
+        let p2pServiceConfig = Object.assign(
+          config.chains[chain][network],
+          { chain, network }
+        );
+        p2pServices.push(new P2pService(p2pServiceConfig));
       }
-      await Promise.all(p2pServices.map(p2pService => p2pService.start())).then(
-        cb
-      );
-    }
-  ],
-  function () {
-    const shouldRunDebug = cluster.isMaster && args.DEBUG;
-    const shouldRunCluster = cluster.isWorker && !args.DEBUG;
-    if (shouldRunDebug || shouldRunCluster) {
-      const server = app.listen(config.port, function () {
-        logger.info(`API server started on port ${config.port}`);
-      });
-      server.timeout = 600000;
     }
   }
-);
+  await Promise.all(p2pServices.map(p2pService => p2pService.start()));
+
+  // TODO this needs to move to an api service
+  const shouldRunDebug = cluster.isMaster && args.DEBUG;
+  const shouldRunCluster = cluster.isWorker && !args.DEBUG;
+  if (shouldRunDebug || shouldRunCluster) {
+    const server = app.listen(config.port, function () {
+      logger.info(`API server started on port ${config.port}`);
+    });
+    // TODO this should be config driven
+    server.timeout = 600000;
+  }
+};
+
+startServices();
diff --git a/src/services/storage.ts b/src/services/storage.ts
@@ -10,34 +10,36 @@ import "../models"
 
 @LoggifyClass
 export class StorageService {
-  start(ready: CallbackType, args: any) {
-    let options = Object.assign({}, config, args);
-    let { dbName, dbHost } = options;
-    const connectUrl = `mongodb://${dbHost}/${dbName}?socketTimeoutMS=3600000&noDelay=true`;
-    let attemptConnect = async () => {
-      return mongoose.connect(connectUrl, {
-        keepAlive: 1,
-        poolSize: config.maxPoolSize,
-        /*
-         *nativeParser: true
-         */
-      });
-    };
-    let attempted = 0;
-    let attemptConnectId = setInterval(async () => {
-      try {
-        let data = await attemptConnect();
-        clearInterval(attemptConnectId);
-        ready(null, data);
-      } catch (err) {
-        logger.error(err);
-        attempted++;
-        if (attempted > 5) {
+  start(args: any): Promise<any> {
+    return new Promise((resolve, reject) => {
+      let options = Object.assign({}, config, args);
+      let { dbName, dbHost } = options;
+      const connectUrl = `mongodb://${dbHost}/${dbName}?socketTimeoutMS=3600000&noDelay=true`;
+      let attemptConnect = async () => {
+        return mongoose.connect(connectUrl, {
+          keepAlive: 1,
+          poolSize: config.maxPoolSize,
+          /*
+           *nativeParser: true
+           */
+        });
+      };
+      let attempted = 0;
+      let attemptConnectId = setInterval(async () => {
+        try {
+          let data = await attemptConnect();
           clearInterval(attemptConnectId);
-          ready(err);
+          resolve(data);
+        } catch (err) {
+          logger.error(err);
+          attempted++;
+          if (attempted > 5) {
+            clearInterval(attemptConnectId);
+            reject(err);
+          }
         }
-      }
-    }, 5000);
+      }, 5000);
+    });
   }
 
   stop() {}
diff --git a/src/services/worker.ts b/src/services/worker.ts
@@ -5,38 +5,36 @@ import config from '../config';
 import { LoggifyClass } from "../decorators/Loggify";
 const cluster = require('cluster');
 const { EventEmitter } = require('events');
-const async = require('async');
 
 @LoggifyClass
 export class WorkerService extends EventEmitter {
-  workers = new Array<{ worker: WorkerType; active: boolean }>();
+  workers = new Array<{ worker: WorkerType; active: boolean, started: Promise<any> }>();
 
-  start(ready: CallbackType) {
-    var self = this;
-    if (cluster.isMaster) {
-      logger.verbose(`Master ${process.pid} is running`);
-      cluster.on('exit', function(worker: WorkerType) {
-        logger.error(`worker ${worker.process.pid} died`);
-      });
-      async.times(
-        config.numWorkers,
-        function(n: any, cb: CallbackType) {
-          var newWorker = cluster.fork();
-          newWorker.on('message', function(msg: any) {
-            self.emit(msg.id, msg);
+  async start() {
+    return new Promise(async resolve => {
+      if (cluster.isMaster) {
+        logger.verbose(`Master ${process.pid} is running`);
+        cluster.on('exit', (worker: WorkerType) => {
+          logger.error(`worker ${worker.process.pid} died`);
+        });
+        for (let worker = 0; worker < config.numWorkers; worker++) {
+          let newWorker = cluster.fork();
+          newWorker.on('message', (msg: any) => {
+            this.emit(msg.id, msg);
+          });
+          let started = new Promise(resolve => {
+            newWorker.on('listening', resolve);
           });
-          self.workers.push({ worker: newWorker, active: false });
-          setTimeout(cb, 3000);
-        },
-        function() {
-          ready();
+          this.workers.push({ worker: newWorker, active: false, started });
         }
-      );
-    }
-    if (cluster.isWorker) {
-      logger.verbose(`Worker ${process.pid} started`);
-      setImmediate(ready);
-    }
+        await Promise.all(this.workers.map(worker => worker.started));
+        resolve();
+      }
+      if (cluster.isWorker) {
+        logger.verbose(`Worker ${process.pid} started`);
+        resolve();
+      }
+    });
   }
 
   stop() {}
