GlobalDB

Author: David Gómez Matarrodona

examples/config/test047_global.json

@@ -38,7 +38,7 @@
 		"console" : {
 			"type" : "console",
 			"config" : {
-				"format" : "Message from flow1 : ${global.test}"
+				"format" : "Message from flow1 : ${_.test} / ${_.itop.asamblea.clientId}"
 			}
 		},
 		"global" : {

lib/cli/basic.js

@@ -66,7 +66,7 @@ else {
 		  command('global').
 			description('Shows global memory').
 		  action(function(args,callback){
-				vorpal.log(global);
+				vorpal.log(global._);
 				callback();
 			});
 

lib/config/index.js

@@ -175,11 +175,11 @@ async function read(file,callback,options) {
 
 		// Set defaults for TLS
 		TLS.defaults({},json.$path);
-		sanitizeModules(json);												// Extend modules config
+		sanitizeModules(json);								// Extend modules config
 		let sflowErrors = sanitizeFlows(json);				// Sanitize flow configs
 
 		let inputErrors = await processInputs(json);	// Get Inputs
-		let filterErrors = processFilters(json);			// Get Filters
+		let filterErrors = processFilters(json);		// Get Filters
 		let flowErrors  = await processFlows(json);		// Get Flows
 
 		let errors = flatten([].concat(

lib/nsyslog/globaldb.js

@@ -0,0 +1,34 @@
+const 
+    fs = require('fs-extra'),
+    path = require('path'),
+    Shm = require('../shm'),
+    logger = require('../logger'),
+    {timer} = require('../util'),
+    chokidar = require('chokidar');
+
+class GlobalDB {
+    constructor(datadir) {
+        this.file = path.resolve(datadir,'global.json');
+        this.watcher = chokidar.watch(path.resolve(datadir,'global.json'));
+        this.watcher.on('all',(event, path)=>{
+            logger.warn('GlobalDB file ',this.file, 'has changed');
+            this.reloadDatabase();
+        });
+    }
+
+    async reloadDatabase() {
+        try {
+            await timer(500);
+            let raw = await fs.readFile(this.file, 'utf-8');
+            let json = JSON.parse(raw);
+
+            Object.keys(json).forEach(k=>{
+                Shm.hset('global',k,json[k]);
+            });
+        }catch(err) {
+            logger.error(err);
+        }
+    }
+}
+
+module.exports = GlobalDB;

lib/nsyslog/index.js

@@ -3,6 +3,7 @@ const
 	Logger = require("../logger"),
 	mingo = require('../mingo'),
 	ClusterAware = require("./clusteraware"),
+	GlobalDB = require('./globaldb'),
 	FileQueue = require("fileq"),
 	Reemiter = require("../reemiter"),
 	Config = require("../config"),
@@ -52,6 +53,7 @@ class NSyslog extends ClusterAware {
 			this.allconfigs = this.config.split();
 			this.config = this.allconfigs[Config.MASTER_PROC];
 			this.modules = this.config.modules;
+			this.globaldb = new GlobalDB(this.datadir);
 		}
 
 		this.mingo = mingo;

lib/processor/parser/sm.js

@@ -58,6 +58,8 @@ class StateMachine {
 		// A la hora de parsear, __ almacena el fragmento de linea a parsear por la regla
 		let obj = {_:line,__:line};
 		let results = [];
+		
+		this.state = this.multi? this.state : START;
 
 		while(true) {
 			let state = this.rules[this.state] || this.rules[START];

lib/shm.js

@@ -153,6 +153,7 @@ else {
 				switch(msg.cmd) {
 					case 'subscribe' :
 						Object.assign(SHM,msg.res);
+						this.shm.emit('init',msg.res);
 						break;
 					default :
 						this[msg.cmd].apply(this,msg.args.concat(true));

lib/transporter/global.js

@@ -10,15 +10,28 @@ const DEF_CONF = {
 	"output" : "nsyslog"
 };
 
-const OUTCACHE = {};
-Shm.shm.on('hset',(parent,key,value)=>{
-	if(parent!='global') return;
-	if(!OUTCACHE[key]) {
-		OUTCACHE[key] = jsexpr.assign(key);
-	}
-	OUTCACHE[key](global,value);
-	logger.silly("HSET",{parent,key,value});
-});
+function initialize() {
+	const OUTCACHE = {};
+	global._ = global._ || {};	
+
+	Shm.shm.on('hset',(parent,key,value)=>{
+		if(parent!='global') return;
+		if(!OUTCACHE[key]) {
+			OUTCACHE[key] = jsexpr.assign(key);
+		}
+		OUTCACHE[key](global._, value);
+		logger.silly("SHM HSET",{parent,key,value});
+	});
+	
+	Shm.shm.on('init',data=>{
+		if(data['global']) {
+			global._ = data.global;
+			logger.silly('SHM INIT',global._);
+		}
+	});
+	
+	
+}
 
 class GlobalTransporter extends Transporter {
 	constructor(id,type) {
@@ -38,10 +51,11 @@ class GlobalTransporter extends Transporter {
 
 	transport(entry, callback) {
 		var msg = this.input(entry);
-		this.output(global, msg);
+		this.output(global._, msg);
 		Shm.hset('global',this.outkey,msg);
 		callback(null,entry);
 	}
 }
 
+initialize();
 module.exports = GlobalTransporter;