Merge branch 'dev' into 'master'

1.2.3 stable

See merge request !33

.env

-BIGSTREAM_TAG=dev
+BIGSTREAM_TAG=test
 BS_SECRET=bigstream-server
 REDIS_TAG=4
 PREFIX_NO=19
-VOLUME=../volume
+VOLUME=~/bsvolume
\ No newline at end of file

Changelog

 #Changelog
-## [1.2.3-UR]
+## [1.2.3] - 2021-05-05
+### Update
+- AMQPLib Support Heartbeat Monitoring
+- RedisLib
 ### Fixed
 - BS :: AMQP RPC Singleton Connection ,auto ack
 - STORAGE :: 50x speedup
@@ -7,9 +10,11 @@
 - BS :: Configuration Context with ENV
 - BS :: keystore env
 - BS :: httplistener headers
+- PLUGIN :: dt-mysql
 - PLUGIN :: dt-pgpcrypt
 - PLUGIN :: dt-transform fn extension
 - PLUGIN :: dt-transform register
+- PLUGIN :: do-mysql
 - PLUGIN :: do-http
 - PLUGIN :: do-bsspeak
 - PLUGIN :: do-mqtt

Dockerfile

-FROM node:lts-alpine
+FROM node:12-alpine
 
 RUN apk add --no-cache make gcc g++ python linux-headers udev
 
@@ -7,8 +7,10 @@ COPY . /app/node-bigstream
 WORKDIR /app/node-bigstream
 
 RUN npm install
+RUN node script/install_plugins.js
 
-FROM node:lts-alpine
+FROM node:12-alpine
+RUN apk add --no-cache python
 
 COPY --from=0 /app/node-bigstream /app/node-bigstream
 
@@ -20,4 +22,4 @@ EXPOSE 19980 19080 19180
 
 # start server
 WORKDIR /app/node-bigstream
-CMD ["pm2-runtime", "start", "pm2-default.json"]
+CMD pm2-runtime pm2.config.js
\ No newline at end of file

conf/env/default.json

-[
-    {"env":"BSCONFIG_AMQP_TYPE","conf":"amqp.type"},
-    {"env":"BSCONFIG_AMQP_URL","conf":"amqp.url"},
-    {"env":"BSCONFIG_MQTT_URL","conf":"mqtt.url"},
-    {"env":"BSCONFIG_MEMSTORE_TYPE","conf":"memstore.type"},
-    {"env":"BSCONFIG_MEMSTORE_URL","conf":"memstore.url"},
-    {"env":"BSCONFIG_STORAGE_REPOSITORY","conf":"storage.repository"},
-    {"env":"BSCONFIG_STORAGE_APIHOSTNAME","conf":"storage.api_hostname"},
-    {"env":"BSCONFIG_SECRET_TEXT","conf":"auth.secret.value"}
-]
\ No newline at end of file

conf/env/index.js

-module.exports = require('./default.json');
\ No newline at end of file

docker-build.sh

+#!/bin/bash
+
+docker build --no-cache -t bigstream:test .

docker-compose.yml

@@ -14,8 +14,9 @@ services:
      - "19150:19150/udp"
     environment:
      - "BS_SECRET=${BS_SECRET}"
+     - "BSCONFIG_HTTPLISTENER_MAXBODY=128mb"
     volumes:
-     - bsdata:/var/bigstream/data
+     - ${VOLUME:-./volume}/bigstream/data:/var/bigstream/data
   redis:
     image: "redis:${REDIS_TAG}"
     command: redis-server --appendonly yes
@@ -28,7 +29,7 @@ services:
     ports:
      - "6379:6379"
     volumes:
-     - bsredis:/data
+     - ${VOLUME:-./volume}/redis/data:/data
   rabbitmq:
     image: "igridproject/rabbitmq"
     command: rabbitmq-server --hostname rabbitmq-server
@@ -40,9 +41,6 @@ services:
              - rabbitmq-server
     ports:
      - "5672:5672"
-volumes:
-  bsdata:
-  bsredis:
 networks:
   bsnet:
      driver: bridge
\ No newline at end of file

env/default.json

@@ -2,10 +2,12 @@
     {"env":"BSCONFIG_AMQP_TYPE","conf":"amqp.type"},
     {"env":"BSCONFIG_AMQP_URL","conf":"amqp.url"},
     {"env":"BSCONFIG_MQTT_URL","conf":"mqtt.url"},
+    {"env":"BSCONFIG_HTTPLISTENER_MAXBODY","conf":"httplistener.max_body"},
     {"env":"BSCONFIG_MEMSTORE_TYPE","conf":"memstore.type"},
     {"env":"BSCONFIG_MEMSTORE_URL","conf":"memstore.url"},
     {"env":"BSCONFIG_STORAGE_REPOSITORY","conf":"storage.repository"},
     {"env":"BSCONFIG_STORAGE_APIHOSTNAME","conf":"storage.api_hostname"},
+    {"env":"BSCONFIG_STORAGE_APIMAXBODY","conf":"storage.api_max_body"},
     {"env":"BSCONFIG_SECRET_TEXT","conf":"auth.secret.value"},
     {"env":"BSCONFIG_KEYSTORE_DIR","conf":"keystore.dir"}
 ]
\ No newline at end of file

http-listener/main.js

@@ -48,13 +48,15 @@ HTTPListener.prototype._http_start = function()
     }
   });
 
-  app.use(bodyParser.json({limit: '128mb',type:"*/json"}));
+  var MAX_BODY = ctx.getConfig('httplistener.max_body','128mb')
+
+  app.use(bodyParser.json({limit: MAX_BODY,type:"*/json"}));
   app.use(bodyParser.urlencoded({
       extended: true,
-      limit: '128mb'
+      limit: MAX_BODY
   }));
-  app.use(bodyParser.text({limit: '128mb',type:"text/*"}));
-  app.use(bodyParser.raw({limit: '128mb',type:"*/*"}));
+  app.use(bodyParser.text({limit: MAX_BODY,type:"text/*"}));
+  app.use(bodyParser.raw({limit: MAX_BODY,type:"*/*"}));
 
   var context = require('./lib/http-context');
   app.use(context.middleware({

package.json

@@ -10,7 +10,7 @@
   "keywords": [],
   "license": "Apache-2.0",
   "dependencies": {
-    "amqplib": "^0.4.2",
+    "amqplib": "^0.7.1",
     "async": "^2.0.1",
     "axon": "^2.0.3",
     "body-parser": "^1.15.2",
@@ -34,13 +34,14 @@
     "query-string": "^4.2.3",
     "quickq": "^0.8.1",
     "random-access-file": "^1.3.0",
-    "redis": "^2.6.5",
+    "redis": "^3.1.2",
     "request": "^2.79.0",
     "thunky": "^1.0.2",
     "tiny-worker": "^2.1.1"
-  },  "scripts": {
+  },
+  "scripts": {
     "test": "echo \"Error: no test specified\" && exit 1",
-    "install": "node ./script/install_plugins.js",
+    "install_plugins": "node ./script/install_plugins.js",
     "start": "pm2 start pm2-default.json",
     "stop": "pm2 stop pm2-default.json"
   }

plugins/do/do-mysql/index.js

+var util = require('util');
+var DOPlugin = require('../do-plugin');
+
+function DOTask(context,request){
+  DOPlugin.call(this,context,request);
+  this.name = "mysql";
+}
+util.inherits(DOTask,DOPlugin);
+
+DOTask.prototype.perform  =  require('./perform');
+
+module.exports = DOTask;

plugins/do/do-mysql/package.json

+{
+  "name": "do-mysql",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "mysql": "^2.18.1"
+  }
+}

plugins/do/do-mysql/perform.js

+var mysql = require('mysql');
+
+var ctx = require('../../../context');
+var Utils = ctx.getLib('lib/util/plugin-utils');
+
+function perform_function(context,request,response){
+  var job_id = context.jobconfig.job_id;
+  var transaction_id = context.transaction.id;
+  var param = context.jobconfig.data_out.param;
+  var memstore = context.task.memstore;
+
+  var in_type = request.type;
+  var data = (Array.isArray(request.data))?request.data:[request.data];
+  var meta = request.meta || {};
+
+  var req_host = param.host || "localhost";
+  var req_user = param.user || "";
+  var req_pass = param.password || "";
+  var req_db = param.database || "";
+  var req_sql = param.sql || "";
+
+  if(typeof data == 'string' && req_sql == ""){
+    req_sql = "${data}"
+  }
+
+  var rsql='';
+  data.forEach((dat)=>{
+    var ev =  {
+      'type' : in_type,
+      'meta' : meta,
+      'data' : dat
+    }
+    rsql+=Utils.vm_execute_text(ev,req_sql) + ';';
+  });
+
+  var conf = {
+    "host" : req_host,
+    "user" : req_user,
+    "password" : req_pass,
+    "database" : req_db
+  }
+
+  response.meta = meta;
+  myexcute(conf,rsql,function(err,result){
+    if(!err){
+      response.success();
+    }else{
+      response.error("mysql error");
+    }
+  });
+
+  //response.success();
+  //response.reject();
+  //response.error("error message")
+
+}
+
+
+function myexcute(conf,sql,cb){
+  var conn = mysql.createConnection(conf);
+  
+  conn.connect(function(err) {
+    
+    if(err) {
+      cb(err);
+      return console.error('could not connect to mysql', err);
+    }
+    
+    conn.query(sql, function (err, result, fields) {
+      var res = {
+        "result":result,
+        "fields":fields
+      }
+      cb(err,res);
+      conn.destroy();
+    });
+
+  });
+
+}
+
+module.exports = perform_function;

plugins/do/do-pgsql/package.json

@@ -9,6 +9,6 @@
   "author": "",
   "license": "ISC",
   "dependencies": {
-    "pg": "^7.0.2"
+    "pg": "^8.5.1"
   }
 }

plugins/dt/dt-mysql/index.js

+var util = require('util');
+var DTPlugin = require('../dt-plugin');
+
+function DTTask(context,request){
+  DTPlugin.call(this,context,request);
+  this.name = "mysql";
+  this.output_type = "";
+}
+util.inherits(DTTask,DTPlugin);
+
+DTTask.prototype.perform  =  require('./perform');
+
+module.exports = DTTask;
\ No newline at end of file

plugins/dt/dt-mysql/package.json

+{
+  "name": "dt-mysql",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "mysql": "^2.18.1"
+  }
+}

plugins/dt/dt-mysql/perform.js

+var mysql = require('mysql');
+
+var ctx = require('../../../context');
+var Utils = ctx.getLib('lib/util/plugin-utils');
+
+function perform_function(context,request,response){
+  var job_id = context.jobconfig.job_id;
+  var transaction_id = context.transaction.id;
+  var param = context.task.config.param || {};
+  var memstore = context.task.memstore
+
+  var output_type = request.input_type;
+  var data = request.data;
+  var meta = request.meta || {};
+
+  var req_host = param.host || "localhost";
+  var req_user = param.user || "";
+  var req_pass = param.password || "";
+  var req_db = param.database || "";
+  var req_sql = param.sql || "";
+
+  var env = {
+    'type' : output_type,
+    'data' : data,
+    'meta' : meta
+  }
+ 
+  if(typeof data == 'string' && req_sql == ""){
+    req_sql = "${data}"
+  }
+  req_sql = Utils.vm_execute_text(env,req_sql);
+
+  //parsing param from meta
+  if(typeof meta._param == 'object')
+  {
+    var _prm = meta._param;
+    req_host = (_prm.host)?_prm.host:req_host;
+    req_user = (_prm.user)?_prm.user:req_user;
+    req_pass = (_prm.password)?_prm.password:req_pass;
+    req_db = (_prm.database)?_prm.database:req_db;
+    req_sql = (_prm.sql)?_prm.sql:req_sql;
+  }
+
+  var conf = {
+    "host" : req_host,
+    "user" : req_user,
+    "password" : req_pass,
+    "database" : req_db
+  }
+
+  response.meta = meta;
+  myexcute(conf,req_sql,function(err,result){
+    if(!err){
+      response.success(result,output_type);
+    }else{
+      response.error("mysql error");
+    }
+  });
+  //response.success();
+  //response.reject();
+  //response.error("error message")
+}
+
+function myexcute(conf,sql,cb){
+  var conn = mysql.createConnection(conf);
+  
+  conn.connect(function(err) {
+    
+    if(err) {
+      cb(err);
+      return console.error('could not connect to mysql', err);
+    }
+    
+    conn.query(sql, function (err, result, fields) {
+      var res = {
+        "result":result,
+        "fields":fields
+      }
+      cb(err,res);
+      conn.destroy();
+    });
+
+  });
+
+}
+
+module.exports = perform_function;

pm2.config.js

+module.exports = {
+  "apps" : [{
+    "name"        : "bs.storage.write",
+    "script"      : "./serv-storage.js",
+	  "args"		  : "--process-write"
+  },
+  {
+    "name"        : "bs.storage.read",
+    "script"      : "./serv-storage.js",
+	  "args"		  : "--process-read",
+	  "exec_mode"   : "cluster",
+    "instances"   : process.env['BS_NUM_READER']||2
+  },
+  {
+    "name"        : "bs.worker",
+    "script"      : "./work-jobworker.js",
+    "exec_mode"   : "cluster",
+    "instances"   : process.env['BS_NUM_WORKER']||0
+  },
+  {
+    "name"        : "bs.trigger.core",
+    "script"      : "./serv-coretrigger.js"
+  },
+  {
+    "name"        : "bs.trigger.httplistener",
+    "script"      : "./serv-httplistener.js",
+    "exec_mode"   : "cluster",
+    "instances"   : process.env['BS_NUM_HTTPLISTENER']||2
+  },
+  {
+    "name"        : "bs.api.service",
+    "script"      : "./serv-api.js"
+  },
+  {
+    "name"        : "bs.trigger.nbudp",
+    "script"      : "./serv-nbudptrigger.js"
+  }]
+}
\ No newline at end of file

storage-service/lib/bss_engine.js

@@ -121,7 +121,7 @@ BSSEngine.prototype.cmd_write = function(prm,cb)
         }
   
         //dataevent.newdata({'resourceId':obj_id.toString(),'storageId':self.name});
-        if(self.context){
+        if(self.context && self.context.evp){
             newdata_event(self.context,{'resourceId':obj_id.toString(),'storageId':self.name});
         }
   

storage-service/main.js

@@ -140,7 +140,7 @@ SS.prototype.http_start = function()
 
   var API_PORT = (this.config.storage.api_port)?this.config.storage.api_port:19080;
 
-  app.use(bodyParser.json({limit: '64mb'}));
+  app.use(bodyParser.json({limit: ctx.getConfig('storage.api_max_body','128mb')}));
   app.use(bodyParser.urlencoded({
       extended: true
   }));

version.json

 {
     "version":"1.2.3",
-    "build":"202007171500"
+    "build":"202105050000"
 }
\ No newline at end of file