moved to zmqrpc for management

History.md

@@ -1,4 +1,5 @@
-# release 0.2.2
+# release 0.2.3
+ * switched to zmqrpc for cloud management
  * fan-out feature added (check feedconfig.yaml)
  * javascript modules can be loaded from job config
  * cloudd with new structures ready on npm

bin/cloudd

@@ -21,36 +21,29 @@ winston.default.transports.console.level = 'info';
 var _versionString = 'cloudd v'+cloudd.version + '\nnode.js cloud engine';
 
 var cmdport = 'tcp://127.0.0.1:12345';
-var zmq = require('zmq');
+var zmqrpc = require('zmqrpc').Client;
 console.log(_versionString);
 var launch='none';
 if(process.argv.length>2) {
   launch = process.argv[2];
 }
 if(launch=='jobs' || launch=='ps' || launch=='tasks' || launch=='submit' || launch=='cron') {
-  var csocket = zmq.socket('req');
-  csocket.identity = 'client' + process.pid;
-  csocket.connect(cmdport);
-  var cmd = {command:launch};
-  if(launch=='submit') {
-    cmd['id']=process.argv[3];
-    cmd['config']=process.argv[4];
-  } else if(launch=='cron') {
-    cmd['cron']=process.argv[3];
-    cmd['id']=process.argv[4];
-    cmd['config']=process.argv[5];
-  }
-  csocket.send(JSON.stringify(cmd));
-  csocket.on('message', function(data) {
-    var objs = JSON.parse(data.toString());
-    switch(launch) {
-      case 'ps'   : _ps(objs); break;
-      case 'tasks': _tasks(objs); break;
-      case 'jobs' : _jobs(objs);break;
-      default     : console.log(objs.result); break;
+  var cloudobj = new zmqrpc(cmdport, function(error, res) {
+    if(error) {
+      console.log('***** Error: ' + error);
+    } else {
+      console.log(res.result);
     }
     process.exit(0);
   });
+  var cmd = {command:launch};
+  switch(launch) {
+    case 'submit': cloudobj.submit(process.argv[3], process.argv[4]); break;
+    case 'cron':   cloudobj.cron(process.argv[3], process.argv[4], process.argv[5]); break;
+    case 'ps':     cloudobj.ps(function(error, result) { _ps(result); process.exit(0); }); break;
+    case 'tasks':  cloudobj.tasks(function(error, result) { _tasks(result); process.exit(0); }); break;
+    case 'jobs':   cloudobj.jobs(function(error, result) { _jobs(result); process.exit(0); }); break;
+  }
 } else if(launch=='server') {
   cloudd.start(cmdport);
 } else {

lib/cloudd.js

@@ -54,56 +54,41 @@ exports.submit = function (jobid, config) {
 };
 
 // ---------- management functions
-var zmq = require('zmq');
-var socket = zmq.socket('rep');
-socket.identity = 'cloudd' + process.pid;
+var zmqrpc = require('zmqrpc').Server;
+
+var CloudManagement = {
+  ps:    function() { return _jobsList.taskQ.runningTasks; },
+  jobs:  function() { return _jobsList.toStringObj(); },
+  tasks: function() { return _jobsList.taskQ.qAsArray(); },
+  submit:function(id, config) { 
+    var res = {result:'job submitted:'+id};
+    try {
+      exports.submit(id, to.load(config));
+    } catch(err) {
+      winston.error(_pkgname+err);
+      res.result=err.message;
+    }
+    return res;
+  },
+  cron: function(cronConfig, id, config) {
+    var res = {result:'cron submitted:'+id};
+    try {
+      exports.submitAt(cronConfig, id, to.load(config));
+    } catch(err) {
+      winston.error(_pkgname+err);
+      res.result=err.message;
+    }
+    return res;
+  }
+};
 
 /* Start the server */
 exports.start = function(port) {
   var cmdport = port || 'tcp://127.0.0.1:12345';
   _jobsList = new job.JobList();
-
-  socket.bind(cmdport, function(err) {
-    if (err) throw err;
-    winston.info(_pkgname+'bound to management port!'+cmdport);
-    // command port message
-    socket.on('message', function(data) {
-      winston.info(_pkgname+'received cmd> ' + data.toString());
-      var obj = JSON.parse(data.toString());
-      switch(obj.command) {
-       case 'ps': socket.send(JSON.stringify(_jobsList.taskQ.runningTasks));break;
-       case 'jobs': socket.send(JSON.stringify(_jobsList.toStringObj()));break;
-       case 'tasks': socket.send(JSON.stringify(_jobsList.taskQ.qAsArray()));break;
-       case 'submit': 
-        var res = {result:'job submitted:'+obj.id};
-        try {
-          exports.submit(obj.id, to.load(obj.config));
-        } catch(err) {
-          winston.error(_pkgname+err);
-          res.result=err.message;
-        }
-        socket.send(JSON.stringify(res));
-        break;
-       case 'cron': 
-        res = {result:'cron submitted:'+obj.id};
-        try {
-          exports.submitAt(obj.cron, obj.id, to.load(obj.config));
-        } catch(err) {
-          winston.error(_pkgname+err);
-          res.result=err.message;
-        }
-        socket.send(JSON.stringify(res));
-        break;
-      default: socket.send('{"result":"command ['+data.toString()+'] not found"}');break;
-      };
-    });
-    // poll for work
-    /*
-    setTimeout(function() {
-      _jobsList.process();
-    }, 100);
-    */
-  });
+  var service = new zmqrpc(cmdport, true);
+  service.context(CloudManagement);
+  service.run();
 };
 
 /* returns current version

package.json

@@ -2,14 +2,14 @@
   "name": "cloudd",
   "description": "node based job execution engine",
   "homepage": "https://github.com/openmason/cloudd",
-  "version": "0.2.2",
+  "version": "0.2.3",
   "author": "el aras <openmason@gmail.com>",
   "dependencies": {
     "underscore": ">= 1.4.3",
     "winston"   : ">= 0.6.2",
     "sprintf"   : ">= 0.1.1",
     "cron"      : ">= 1.0.1",
-    "zmq"       : ">= 2.2.0",
+    "zmqrpc"    : ">= 0.0.6",
     "to"        : ">= 0.2.9",
     "handy"     : ">= 0.0.11"
   },