mysql plugins

Changelog

@@ -7,9 +7,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

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/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;

version.json

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