Add multi connection support

config/epp-config-example.json

@@ -27,7 +27,8 @@
                 "DNSSEC": {
                     "xmlns": "urn:ietf:params:xml:ns:secDNS-1.1"
                 }
-            }
+            },
+            "connections":[10, 100]
         },
         "registry-test2": {
             "ssl": true,
@@ -57,7 +58,8 @@
                 "keyvalue": {
                     "xmlns": "http://schema.ispapi.net/epp/xml/keyvalue-1.0"
                 }
-            }
+            },
+            "connections":[10, 100]
         }
     },
     "rabbitmq": {

lib/dispatcher.js

@@ -65,9 +65,11 @@ Dispatcher.prototype.processMessage = function processMessage(m) {
                 loginTransactionId).then(
                     function(data) {
                         logger.log("Got login data: ", data.toString());
+                        process.send(currentState.loggedIn);
                     },
                     function(error) {
                         logger.error("Unable to login: ", error);
+                        process.send(currentState.loggedIn);
                     });
             },
             2000);

lib/epp-factory.js

@@ -20,6 +20,9 @@ var logger = new (winston.Logger)({
 function EppFactory() {}
 
 EppFactory.generate = function(registry, config) {
+    if (!registry) {
+        return;
+    }
     var epp = new EPP(registry, config);
     config.extensionClasses.forEach(function(extensionClass) {
         var extension = extensionClass.extension;

lib/node-epp-server.js

@@ -4,11 +4,10 @@ var cp = require('child_process');
 var program = require('commander');
 var ipfilter =  require('ipfilter');
 var moment = require('moment');
-var Listener = require('./listener.js');
-var EventDispatcher = require('./event-dispatcher.js');
 var path = require('path');
 var nconf = require('nconf');
 var winston = require('winston');
+var Pool = require('generic-pool').Pool;
 
 nconf.env().file({
     "file": path.resolve(__dirname, '..', 'config/epp-config.json')
@@ -35,68 +34,124 @@ logger.log("Initialised with registries: ", registries);
 var availableProcesses = {};
 for (var accred in registries) {
     var registry = registries[accred];
-    var eppProcess = cp.fork(__dirname + '/worker.js');
-    eppProcess.send({
-        "registry": registry
+    var registryConfig = nconf.get('registries')[registry];
+    var pool = new Pool({
+        name     : registry,
+        create   : function(callback) {
+            var client = cp.fork(__dirname + '/worker.js');
+            client.once('message', function(loggedIn) {
+                if (!loggedIn) {
+                    client.kill();
+                    return callback("Login failed.", null);
+                }
+                // parameter order: err, resource
+                callback(null, client);
+            });
+            client.send({
+                "registry": registry
+            });
+        },
+        destroy: function(client) {
+            setTimeout(function() {
+                client.kill();
+            }, 2000);
+            client.send({
+                "command": "logout",
+                "data": {"kill": true}
+            });
+        },
+        validateAsync: function(client, callback) {
+            if (!client.connected) {
+                return callback(false);
+            }
+            return callback(true);
+        },
+        returnToHead: true,
+        max      : registryConfig['connections'][1],
+        min      : registryConfig['connections'][0],
+        // specifies how long a resource can stay idle in pool before being removed
+        idleTimeoutMillis : 30000,
+         // if true, logs via console.log - can also be a function
+        log : true
     });
-    availableProcesses[registry] = eppProcess;
+    availableProcesses[registry] = pool;
 }
-process.on('SIGINT', function() {
-    var logoutResponse = function(data) {
-        logger.log("Got reply from logout: ", data);
-    };
-    for (var registry in availableProcesses) {
-        var childProc = availableProcesses[registry];
-        childProc.send({
-            "command": "logout",
-            "data": {"kill": true}
-        });
-        childProc.once('message', logoutResponse);
-    }
-    process.exit(0);
-});
-
 
 // Wire up event/listener to keep track of available worker process. This is
 // to avoid individual connection workers from getting swamped.
-var eventDispatch =  new EventDispatcher();
-var listener = new Listener(eventDispatch, availableProcesses);
-eventDispatch.on('queueChild', listener.queueChild);
-eventDispatch.on('childFree', listener.childFree);
-
-
 var app = express();
 app.use(bodyParser.json());
 var ips = nconf.get('whitelisted_ips');
 app.use(ipfilter(ips, {"mode": "allow"}));
 
 app.post('/command/:registry/:command', function(req, res) {
     var registry = req.params.registry;
-    if (! (registry in availableProcesses)) {
-        res.send(400, {
+    if (!(registry in availableProcesses)) {
+        res.status(400).send({
             "error": "Unknown registry"
         });
         return;
     }
+
     var queryData = req.body;
+    var pool = availableProcesses[registry];
+    pool.acquire(function(err, client) {
+        if (err) {
+            logger.error(err);
+            res.status(400).send({
+                "error": "Session overload"
+            });
+            return;
+        }
+
+        var registryConfig = nconf.get('registries')[registry];
+        var t = setTimeout(function() {
+            pool.destroy(client);
+            res.status(400).send({
+                "error": "Timeout"
+            });
+            return;
+        }, registryConfig['max_runtime']);
 
-    var a = moment();
-    var processChild = function (childProcess) {
-        childProcess.once('message', function(m) {
+        var a = moment();
+        client.once('message', function(m) {
+            clearTimeout(t);
             var b = moment();
             var diff = b.diff(a, 'milliseconds');
             logger.info('Request elapsed time: '+ diff.toString() + ' ms');
+            m['runtime'] = diff/1000;
+            pool.release(client);
             res.send(m);
-            eventDispatch.childFree(registry);
         });
-        childProcess.send({
+
+        client.send({
             "command": req.params.command,
             "data": queryData
         });
-    };
-    listener.pushChildQueue(processChild);
-    eventDispatch.queueChild(registry);
+    });
+});
+
+process.on('SIGINT', function(err) {
+    if (err) {
+        console.log(err.stack);
+    }
+
+    for (var registry in availableProcesses) {
+        var pool = availableProcesses[registry];
+        pool.drain(function() {
+            pool.destroyAllNow();
+        });
+    }
+
+    setTimeout(function() {
+        process.exit(0);
+    }, 5000);
 });
 
-app.listen(program.listen);
+process.on('uncaughtException', function(err) {
+    console.error(err);
+    console.error(err.stack);
+});
+process.stdin.resume(); //so the program will not close instantly
 
+app.listen(program.listen);

lib/worker.js

@@ -2,3 +2,14 @@ var Dispatcher = require('./dispatcher.js');
 var dispatch = new Dispatcher();
 
 process.on('message', dispatch.processMessage);
+process.on('SIGINT', function(err) {
+    console.log("Try to logout.");
+    setTimeout(function() {
+        process.exit(0);
+    }, 2000);
+    process.send({
+        "command": "logout",
+        "data": {"kill": true}
+    });
+});
+

package.json

@@ -33,6 +33,7 @@
     "q": "*",
     "xml2json": "*",
     "winston": "*",
+    "generic-pool": "*",
     "crypto": "*"
   },
   "devDependencies": {