add multi process support for portal

Author: lijinxia

Diff for: source/portal/index.js

@@ -248,79 +248,115 @@ var rpcPublic = {
   }
 };
 
-amqper.connect(config.rabbit, function () {
-  amqper.asRpcClient(function(rpcClnt) {
-    rpcClient = rpcClnt;
-    log.info('portal initializing as rpc client ok');
-    joinCluster(function(id) {
-      log.info('portal join cluster ok, with rpcID:', id);
-        amqper.asRpcServer(id, rpcPublic, function(rpcSvr) {
-          log.info('portal initializing as rpc server ok');
-            amqper.asMonitor(function (data) {
-              if (data.reason === 'abnormal' || data.reason === 'error' || data.reason === 'quit') {
-                if (portal !== undefined) {
-                  if (data.message.purpose === 'conference') {
-                    return portal.getParticipantsByController(data.message.type, data.message.id)
-                      .then(function (impactedParticipants) {
-                        impactedParticipants.forEach(function(participantId) {
-                          log.error('Fault on conference controller(type:', data.message.type, 'id:', data.message.id, ') of participant', participantId, 'was detected, drop it.');
-                          socketio_server && socketio_server.drop(participantId);
+// Mutiple process setup
+var cluster = require("cluster");
+var numCPUs = 4;
+var rpcID;
+
+if (cluster.isMaster) {
+  // Master Process
+  amqper.connect(config.rabbit, function () {
+    amqper.asRpcClient(function(rpcClnt) {
+      rpcClient = rpcClnt;
+      log.info('portal initializing as rpc client ok');
+      joinCluster(function(id) {
+        rpcID = id;
+        log.info('portal join cluster ok, with rpcID:', id);
+          amqper.asRpcServer(id, rpcPublic, function(rpcSvr) {
+            log.info('portal initializing as rpc server ok');
+              amqper.asMonitor(function (data) {
+                if (data.reason === 'abnormal' || data.reason === 'error' || data.reason === 'quit') {
+                  if (portal !== undefined) {
+                    if (data.message.purpose === 'conference') {
+                      return portal.getParticipantsByController(data.message.type, data.message.id)
+                        .then(function (impactedParticipants) {
+                          impactedParticipants.forEach(function(participantId) {
+                            log.error('Fault on conference controller(type:', data.message.type, 'id:', data.message.id, ') of participant', participantId, 'was detected, drop it.');
+                            socketio_server && socketio_server.drop(participantId);
+                          });
                         });
-                      });
+                    }
                   }
                 }
-              }
-            }, function (monitor) {
-              log.info(id + ' as monitor ready');
-              getTokenKey(id, function(tokenKey) {
-                startServers(id, tokenKey);
-              }, function() {
-                log.error('portal getting token failed.');
+              }, function (monitor) {
+                log.info(id + ' as monitor ready');
+                for (var i = 0; i < numCPUs; i++) {
+                  cluster.fork();
+                }
+            
+                cluster.on('exit', function(worker, code, signal) {
+                    log.info(`Worker ${worker.process.pid} died`);
+                });  
+              }, function(reason) {
+                log.error('portal initializing as monitor failed, reason:', reason);
                 stopServers();
                 process.exit();
               });
-            }, function(reason) {
-              log.error('portal initializing as rpc client failed, reason:', reason);
-              stopServers();
-              process.exit();
-            });
-      }, function(reason) {
-        log.error('portal initializing as rpc client failed, reason:', reason);
-        stopServers();
-        process.exit();
+        }, function(reason) {
+          log.error('portal initializing as rpc server failed, reason:', reason);
+          stopServers();
+          process.exit();
+        });
       });
+    }, function(reason) {
+      log.error('portal initializing as rpc client failed, reason:', reason);
+      stopServers();
+      process.exit();
     });
   }, function(reason) {
-    log.error('portal initializing as rpc client failed, reason:', reason);
-    stopServers();
-    process.exit();
+      log.error('portal connect to rabbitMQ server failed, reason:', reason);
+      process.exit();
   });
-}, function(reason) {
-    log.error('portal connect to rabbitMQ server failed, reason:', reason);
-    process.exit();
-});
 
-['SIGINT', 'SIGTERM'].map(function (sig) {
-  process.on(sig, async function () {
-    log.warn('Exiting on', sig);
+  ['SIGINT', 'SIGTERM'].map(function (sig) {
+    process.on(sig, async function () {
+      log.warn('Master exiting on', sig);
+      stopServers();
+      amqper.disconnect();
+      process.exit();
+    });
+  });
+  
+  process.on('SIGPIPE', function () {
+    log.warn('SIGPIPE!!');
+  });
+  
+  process.on('exit', function () {
+    log.info('Master Process exit');
+  });
+  
+  process.on('unhandledRejection', (reason) => {
+      log.info('Reason: ' + reason);
+  });
+  
+  process.on('SIGUSR2', function () {
+    logger.reconfigure();
+  });
+}else{
+  // Worker Process
+  getTokenKey(rpcID, function(tokenKey) {
+    startServers(rpcID, tokenKey);
+  }, function() {
+    log.error('portal getting token failed.');
     stopServers();
-    amqper.disconnect();
     process.exit();
   });
-});
 
-process.on('SIGPIPE', function () {
-  log.warn('SIGPIPE!!');
-});
+  log.info(`Worker ${process.pid} started`);
 
-process.on('exit', function () {
-  log.info('Process exit');
-});
+  ['SIGINT', 'SIGTERM'].map(function (sig) {
+    process.on(sig, async function () {
+      log.warn('Worker exiting on', sig);
+      stopServers();
+      process.exit();
+    });
+  });
 
-process.on('unhandledRejection', (reason) => {
-    log.info('Reason: ' + reason);
-});
+  process.on('exit', function () {
+    log.info('Process exit');
+  });
 
-process.on('SIGUSR2', function () {
-  logger.reconfigure();
-});
+  process.on('SIGUSR2', function() {
+      logger.reconfigure();
+  });
+}