--no commit message

--no commit message

coreservice/scheduler.js

@@ -71,7 +71,7 @@ SchedulerService.prototype._callJob = function(cron)
     'object_type':'job_execute',
     'source' : 'scheduler',
     'jobId' : cron.jobid,
-    'option' : {},
+    'option' : {'exe_level':'secondary'},
     'input_data' : {
       'type' : 'bsdata',
       'value' : {

http-listener/main.js

@@ -5,8 +5,11 @@ var app = express();
 var bodyParser = require('body-parser');
 
 var cfg = ctx.config;
+var ConnCtx = ctx.getLib('lib/conn/connection-context');
 var HttpACL = ctx.getLib('lib/mems/http-acl');
 var EvenPub = ctx.getLib('lib/amqp/event-pub');
+var QueueCaller = ctx.getLib('lib/amqp/queuecaller');
+var EvenSub = ctx.getLib('lib/amqp/event-sub');
 
 const JOBCHANEL = 'bs_job_cmd';
 
@@ -20,13 +23,16 @@ function HTTPListener(cfg)
 {
     this.config = cfg;
     this.httpacl = HttpACL.create({'conn':this.config.memstore.url});
-    this.evp = new EvenPub({'url':this.config.amqp.url,'name':JOBCHANEL});
+    this.jobcaller = new QueueCaller({'url':this.config.amqp.url,'name':'bs_jobs_cmd'});
+    this.evs = new EvenSub({'url':this.config.amqp.url,'name':'bs_trigger_cmd'});
+    //this.evp = new EvenPub({'url':this.config.amqp.url,'name':JOBCHANEL});
 }
 
 HTTPListener.prototype.start = function()
 {
   console.log('Starting HTTP Listener ...\n');
   this.http_start();
+  this._controller_start();
 }
 
 HTTPListener.prototype.http_start = function()
@@ -51,7 +57,7 @@ HTTPListener.prototype.http_start = function()
   var context = require('./lib/http-context');
   app.use(context.middleware({
     'httpacl' : self.httpacl,
-    'evp' : self.evp
+    'jobcaller' : self.jobcaller
   }));
 
   app.use(require('./ws'));
@@ -62,5 +68,39 @@ HTTPListener.prototype.http_start = function()
     console.log('WWW:HTTP START\t\t[OK]');
   });
 
+}
+
+HTTPListener.prototype._controller_start = function ()
+{
+  var self=this;
+  var topic = 'ctl.trigger.#';
+  self.evs.sub(topic,function(err,msg){
+    if(!msg){return;}
+
+    var ctl = msg.data;
+    if(ctl.trigger_type != 'http' && ctl.trigger_type != 'all')
+    {
+      return;
+    }
+
+    if(ctl.cmd == 'reload')
+    {
+      console.log('WWW:Reloading ACL\t[OK]');
+      self.reload();
+    }
+
+  });
+}
 
+HTTPListener.prototype.reload = function ()
+{
+  var self = this;
+
+  self.httpacl.update(function(err){
+    if(!err){
+      console.log('WWW:ACL Update\t\t[OK]');
+    }else{
+      console.log('WWW:ACL Update\t\t[ERR]');
+    }
+  });
 }

http-listener/ws/service-main.js

@@ -5,9 +5,11 @@ var express = require('express');
 var router = express.Router();
 
 var cfg = ctx.config;
+
 var response = ctx.getLib('lib/ws/response');
 var request = ctx.getLib('lib/ws/request');
 
+
 var process_get = function(req, res) {
   var reqHelper = request.create(req);
   var respHelper = response.create(res);
@@ -15,7 +17,8 @@ var process_get = function(req, res) {
   var ctx = req.context;
 
   var httpacl = req.context.httpacl;
-  var evp = req.context.evp;
+  //var evp = req.context.evp;
+  var jobcaller = req.context.jobcaller;
 
   var j = httpacl.findJob(appkey,'get');
 
@@ -45,7 +48,8 @@ var process_get = function(req, res) {
     var msg = job_execute_msg;
     msg.jobId = item.jobid;
 
-    evp.send(topic,msg);
+    jobcaller.send(msg);
+    //evp.send(topic,msg);
   });
 
   if(j.length > 0)

jobworker/worker.js

 var ctx = require('../context');
 var cfg = ctx.config;
 
+var QueueCaller = ctx.getLib('lib/amqp/queuecaller');
 var QueueReceiver = ctx.getLib('lib/amqp/queuereceiver');
 var ConnCtx = ctx.getLib('lib/conn/connection-context');
 var JobRegistry = ctx.getLib('lib/mems/job-registry');
@@ -25,32 +26,71 @@ var JW = function JobWorker (prm)
 
 JW.prototype.start = function ()
 {
-  this.amqp_job_start();
+  this.amqp_pmr_start();
+  this.amqp_snd_start();
 }
 
-JW.prototype.amqp_job_start = function ()
+JW.prototype.amqp_pmr_start = function ()
 {
   var self=this;
-  if(self.amqp_server){return;}
+  if(self.amqp_server_pmr){return;}
 
-  self.amqp_server = new QueueReceiver({
+  if(!self.QCaller){
+    self.QCaller = new QueueCaller({'url':self.conn.getAmqpUrl(),'name':'bs_jobs_queue'});
+  }
+
+  self.amqp_server_pmr = new QueueReceiver({
                 url : self.conn.getAmqpUrl(),
-                name : 'bs_jobs_queue'
+                name : 'bs_jobs_cmd'
               });
 
-  self.amqp_server.set_execute_function(function(data,callback){
-    var jt = new JobTransaction({'handle':self,'cmd':data});
-    jt.run(function(err){
-      if(err){
-        console.log(err);
-      }
-      callback();
-    });
+  self.amqp_server_pmr.set_execute_function(function(data,callback){
+    if(data.option && data.option.exe_level && data.option.exe_level=='secondary')
+    {
+      console.log('WORKER:Forword job[' + data.jobId + '] to SJW');
+      self.QCaller.send(data);
+    }else{
+      self._execute_job(data,function (err) {
+
+      });
+    }
+    callback();
+  });
+
+  self.amqp_server_pmr.start(function(err){
+    console.log('WORKER:Primary Start\t\t[OK]');
+  })
+
+}
+
+JW.prototype.amqp_snd_start = function ()
+{
+  var self=this;
+  if(self.amqp_server_snd){return;}
+
+  self.amqp_server_snd = new QueueReceiver({
+                url : self.conn.getAmqpUrl(),
+                name : 'bs_jobs_queue'
+              });
 
+  self.amqp_server_snd.set_execute_function(function(data,callback){
+    self._execute_job(data,callback);
   });
 
-  self.amqp_server.start(function(err){
-    console.log('worker start');
+  self.amqp_server_snd.start(function(err){
+    console.log('WORKER:Secondary Start\t\t[OK]');
   })
 
 }
+
+JW.prototype._execute_job = function (data,callback)
+{
+  var self=this;
+  var jt = new JobTransaction({'handle':self,'cmd':data});
+  jt.run(function(err){
+    if(err){
+      console.log(err);
+    }
+    callback();
+  });
+}

lib/mems/http-acl.js

-var Redis = require('ioredis');
+var Redis = require('redis');
 const PREFIX = 'bs:http:acl';
 
 module.exports.create = function(cfg)
@@ -22,7 +22,7 @@ function HttpACL(cfg)
   this.config = cfg;
 
   if(cfg.conn){
-    this.mem = new Redis(cfg.conn);
+    this.mem = Redis.createClient(cfg.conn);
   }else if(cfg.redis){
     this.mem = cfg.redis;
   }else{