@@ -37,7 +37,7 @@ exports.install = function(instance) {
if (!running) return;
let value = response.data;
if (typeof value !== 'object') return;
-
+
let can = false
var k = Object.keys(value);
var interested = JSON.parse(instance.options.types);
@@ -57,11 +57,11 @@ exports.install = function(instance) {
let icon = ':totaljs:';
let type = value[k[0]][0]['values']['_event']['type'];
let source = value[k[0]][0]['values']['_event']['source']['func'];
- let message = value[k[0]][0]['values']['_event']['message']['en'];
+ let message = value[k[0]][0]['values']['_event']['message'];
let message_data = value[k[0]][0]['values']['_event']['message_data'];
let tag = '';
- switch (type) {
+ switch(type){
case 'debug':
icon = ':beetle:';
break;
@@ -89,15 +89,15 @@ exports.install = function(instance) {
}
// Check if this message includes one of the strings we are watching for
- for (const msg of msg_incl) {
- if (message.includes(msg)) {
+ for (const msg of msg_incl){
+ if (message.includes(msg)){
if (msg == 'is responding again') icon = ':large_green_circle:';
can = true;
break;
}
}
// Check if message is one of the types we are watching for
- if (interested.includes(type)) {
+ if (interested.includes(type)){
can = true;
}
@@ -105,10 +105,10 @@ exports.install = function(instance) {
// Check for each person tags based on what the message includes
- for (const person of tags) {
- for (const msg of person.includes) {
- if (message.includes(msg)) {
- tag += '<@' + person.user_id + '> ';
+ for (const person of tags){
+ for (const msg of person.includes){
+ if (message.includes(msg)){
+ tag += '<@'+person.user_id+'> ';
break; // Break out from this person checks as they are already tagged now
}
}
@@ -116,46 +116,46 @@ exports.install = function(instance) {
// Now that all people are tagged add new line symbol
if (tag != '') tag += '\n';
- let send_data = tag + instance.options.name + ' ' + type.toUpperCase() + '\n*Source*: ' + source + '\n*Message*: ' + message;
+ let send_data = tag+instance.options.name+' '+type.toUpperCase()+'\n*Source*: '+source+'\n*Message*: '+message;
if (message_data) {
- send_data += '\nData: ' + message_data;
+ send_data += '\nData: '+message_data;
}
let ignore_msg = false
- if (message.includes('Configuration of dimming profile to node no')) {
- for (let i = 0; i < FLOW["savedSlackMessages"].length; i++) {
- if (FLOW["savedSlackMessages"][i].message == message) {
+ if (message.includes('Configuration of dimming profile to node no')){
+ for (let i = 0; i < FLOW["savedSlackMessages"].length; i++){
+ if (FLOW["savedSlackMessages"][i].message == message){
ignore_msg = true;
break;
}
}
- if (!ignore_msg) {
- FLOW["savedSlackMessages"].push({ message, 'dateandtime': Date.now() });
- if (timer === null) {
- timer = setTimeout(checkSavedMessages, 60 * 60000);
+ if (!ignore_msg){
+ FLOW["savedSlackMessages"].push({message, 'dateandtime': Date.now()});
+ if (timer === null){
+ timer = setTimeout(checkSavedMessages, 60*60000);
}
}
}
- if (!ignore_msg) {
- instance.send2({ 'msg': send_data, 'bot_name': instance.options.name + ' ' + type.toUpperCase(), 'bot_icon': icon, 'channel': instance.options.slack_channel });
+ if (!ignore_msg){
+ instance.send2({'msg':send_data,'bot_name':instance.options.name+' '+type.toUpperCase(),'bot_icon':icon,'channel':instance.options.slack_channel});
}
});
- function checkSavedMessages() {
+ function checkSavedMessages(){
var d = Date.now();
d = d - 86400000; // older then 24hr
var a = [];
//Remove msgs older then 24hr
- for (let i = 0; i < FLOW["savedSlackMessages"].length; i++) {
- if (FLOW["savedSlackMessages"][i].dateandtime > d) {
+ for (let i = 0; i < FLOW["savedSlackMessages"].length; i++){
+ if (FLOW["savedSlackMessages"][i].dateandtime > d){
a.push(FLOW["savedSlackMessages"][i]);
}
}
FLOW["savedSlackMessages"] = a;
if (FLOW["savedSlackMessages"].length > 0) {
- timer = setTimeout(checkSavedMessages, 60 * 60000);
+ timer = setTimeout(checkSavedMessages, 60*60000);
} else {
timer = null;
}
@@ -163,7 +163,7 @@ exports.install = function(instance) {
instance.reconfigure = function() {
try {
- if (!FLOW["savedSlackMessages"]) {
+ if (!FLOW["savedSlackMessages"]){
FLOW["savedSlackMessages"] = [];
}
@@ -183,4 +183,5 @@ exports.install = function(instance) {
instance.on('options', instance.reconfigure);
setTimeout(instance.reconfigure, 10000);
+
};
diff --git a/flow/thermometer.js b/flow/thermometer.js
index fc7074d..fa015c6 100644
--- a/flow/thermometer.js
+++ b/flow/thermometer.js
@@ -33,7 +33,7 @@ exports.install = function(instance) {
logger.debug(exports.title, "installed");
- instance.on("close", function() {
+ instance.on("close", function(){
clearInterval(startRead);
})
@@ -42,25 +42,24 @@ exports.install = function(instance) {
try {
- if (temperatureAddress === "") throw "Thermometer: temperatureAddress is not defined";
+ if(temperatureAddress === "") throw "Thermometer: temperatureAddress is not defined";
exec(`owread -C ${temperatureAddress}/temperature`, (error, stdout, stderr) => {
- if (!error) {
+ if(!error)
+ {
parseData(stdout)
return;
}
counter++;
- if (counter == NUMBER_OF_FAILURES_TO_SEND_ERROR) {
- sendNotification("Thermometer_main", rvoTbName, "thermometer_is_not_responding", {}, { "Error": error }, SEND_TO.tb, instance, "thermometer");
- monitor.info("Thermometer is not responding", error);
- }
- instance.send(SEND_TO.dido_controller, { status: "NOK-thermometer" });
+ if(counter == NUMBER_OF_FAILURES_TO_SEND_ERROR) sendNotification("Thermometer_main", rvoTbName, "thermometer_is_not_responding", {}, {"Error": error}, SEND_TO.tb, instance, "thermometer");
+ monitor.info("Thermometer is not responding", error);
+ instance.send(SEND_TO.dido_controller, {status: "NOK-thermometer"});
});
}
- catch (err) {
+ catch(err) {
errLogger.error(exports.title, err);
clearInterval(startRead);
}
@@ -71,12 +70,12 @@ exports.install = function(instance) {
data = parseFloat(data);
//logger.debug("Thermometer", data);
- if (isNaN(data)) {
+ if(isNaN(data)) {
errLogger.error("Thermometer sends invalid data");
return;
}
- if (counter > NUMBER_OF_FAILURES_TO_SEND_ERROR) //1 hour
+ if(counter > NUMBER_OF_FAILURES_TO_SEND_ERROR) //1 hour
{
instance.send(SEND_TO.debug, "Thermometer - temperature data are comming again");
sendNotification("Thermometer_parseData", rvoTbName, "thermometer_is_responding_again", {}, "", SEND_TO.tb, instance, "thermometer");
@@ -86,7 +85,7 @@ exports.install = function(instance) {
"temperature": Number(data.toFixed(2)),
}
- instance.send(SEND_TO.dido_controller, { values: values });
+ instance.send(SEND_TO.dido_controller, {values: values});
counter = 0;
}
diff --git a/flow/wsmqttpublish.js b/flow/wsmqttpublish.js
index 0eba804..b2a218d 100644
--- a/flow/wsmqttpublish.js
+++ b/flow/wsmqttpublish.js
@@ -55,7 +55,7 @@ let saveTelemetryOnError = true;//backup_on_failure overrides this value
//------------------------
let rollers;
-if (createTelemetryBackup) rollers = require('streamroller');
+if(createTelemetryBackup) rollers = require('streamroller');
const noSqlFileSizeLimit = 4194304;//use 5MB - 4194304
let insertNoSqlCounter = 0;
@@ -70,14 +70,14 @@ let lastRestoreTime = 0;
// if there is an error in client connection, flow logs to monitor.txt. Not to log messages every second, we use sendClientError variable
let sendClientError = true;
-process.on('uncaughtException', function(err) {
+process.on('uncaughtException', function (err) {
errLogger.error('uncaughtException:', err.message)
errLogger.error(err.stack);
//TODO
//send to service
-
+
//process.exit(1);
})
@@ -96,19 +96,22 @@ exports.install = function(instance) {
let sendWsStatusVar = null;
let wsmqtt_status = 'disconnected';
- function getWsmqttName(host) {
- if (host == "tb-demo.worksys.io" || host == '192.168.252.4') return 'wsmqtt_demo';
- else if (host == "tb-qas01.worksys.io" || host == '192.168.252.5') return 'wsmqtt_qas01';
- else if (host == "tb-prod01.worksys.io" || host == '192.168.252.1') return 'wsmqtt_prod01';
+ function getWsmqttName(host)
+ {
+ if(host == "tb-demo.worksys.io" || host == '192.168.252.4') return 'wsmqtt_demo';
+ else if(host == "tb-qas01.worksys.io" || host == '192.168.252.5') return 'wsmqtt_qas01';
+ else if(host == "tb-prod01.worksys.io" || host == '192.168.252.1') return 'wsmqtt_prod01';
}
- function sendWsStatus() {
- instance.send(SEND_TO.services, { [wsmqttName]: wsmqtt_status });
+ function sendWsStatus()
+ {
+ instance.send(SEND_TO.services, {[wsmqttName]: wsmqtt_status});
}
- function main() {
- if (!FLOW.dbLoaded) return;
+ function main()
+ {
+ if(!FLOW.dbLoaded) return;
loadSettings();
clearInterval(sendWsStatus);
@@ -116,9 +119,11 @@ exports.install = function(instance) {
}
//set opts according to db settings
- function loadSettings() {
+ function loadSettings()
+ {
- if (instance.options.host !== "") {
+ if(instance.options.host !== "")
+ {
//override settings from database
var o = instance.options;
opts = {
@@ -134,12 +139,13 @@ exports.install = function(instance) {
console.log("wsmqttpublich -> loadSettings from instance.options", instance.options);
}
- else {
-
+ else
+ {
+
const SETTINGS = FLOW.GLOBALS.settings;
backup_on_failure = SETTINGS.backup_on_failure;
saveTelemetryOnError = backup_on_failure;
-
+
restore_from_backup = SETTINGS.restore_from_backup;
restore_backup_wait = SETTINGS.restore_backup_wait;
@@ -164,7 +170,8 @@ exports.install = function(instance) {
connectToTbServer();
}
- function connectToTbServer() {
+ function connectToTbServer()
+ {
var url = "mqtt://" + opts.host + ":" + opts.port;
console.log("MQTT URL: ", url);
@@ -172,7 +179,7 @@ exports.install = function(instance) {
client.on('connect', function() {
instance.status("Connected", "green");
- //monitor.info("MQTT client connected");
+ monitor.info("MQTT client connected");
sendClientError = true;
clientReady = true;
@@ -191,31 +198,31 @@ exports.install = function(instance) {
TRY(function() {
message = JSON.parse(message);
- if (message.hasOwnProperty("device") && message.hasOwnProperty("data") && message.data.hasOwnProperty("id")) {
- client.publish(topic, `{"device": ${message.device}, "id": ${message.data.id}, "data": {"success": true}}`, { qos: 1 });
- instance.send(SEND_TO.rpcCall, { "device": message.device, "id": message.data.id, "RPC response": { "success": true } });
+ if (message.hasOwnProperty("device") && message.hasOwnProperty("data") && message.data.hasOwnProperty("id")) {
+ client.publish(topic, `{"device": ${message.device}, "id": ${message.data.id}, "data": {"success": true}}`, {qos:1});
+ instance.send(SEND_TO.rpcCall, {"device": message.device, "id": message.data.id, "RPC response": {"success": true}});
}
}, () => instance.debug('MQTT: Error parsing data', message));
}
- instance.send(SEND_TO.rpcCall, { "topic": topic, "content": message });
+ instance.send(SEND_TO.rpcCall, {"topic":topic, "content":message });
});
client.on('close', function() {
clientReady = false;
wsmqtt_status = 'disconnected';
-
+
instance.status("Disconnected", "red");
- instance.send(SEND_TO.debug, { "message": "Client CLOSE signal received !" });
+ instance.send(SEND_TO.debug, {"message":"Client CLOSE signal received !"});
});
client.on('error', function(err) {
- instance.status("Err: " + err.code, "red");
- instance.send(SEND_TO.debug, { "message": "Client ERROR signal received !", "error": err, "opt": opts });
- if (sendClientError) {
- monitor.info('MQTT client error', err);
- sendClientError = false;
+ instance.status("Err: "+ err.code, "red");
+ instance.send(SEND_TO.debug, {"message":"Client ERROR signal received !", "error":err, "opt":opts });
+ if(sendClientError) {
+ monitor.info('MQTT client error', err);
+ sendClientError = false;
}
clientReady = false;
wsmqtt_status = 'disconnected';
@@ -231,39 +238,45 @@ exports.install = function(instance) {
instance.on('1', function(data) {
- if (clientReady) {
- //do we have some data in backup file? if any, process data from database
- if (saveTelemetryOnError) {
+ if(clientReady)
+ {
+ //do we have some data in backup file? if any, process data from database
+ if(saveTelemetryOnError)
+ {
//read telemetry data and send back to server
- if (!processingData) processDataFromDatabase();
+ if(!processingData) processDataFromDatabase();
}
-
+
let stringifiedJson = JSON.stringify(data.data);
- client.publish("v1/gateway/telemetry", stringifiedJson, { qos: 1 });
+ client.publish("v1/gateway/telemetry", stringifiedJson, {qos: 1});
//backup telemetry
- if (createTelemetryBackup) {
+ if(createTelemetryBackup)
+ {
data.data.id = UID();
nosqlBackup.insert(data.data);
insertBackupNoSqlCounter++;
- if (insertBackupNoSqlCounter > 150) {
- let options = { compress: true };
+ if(insertBackupNoSqlCounter > 150)
+ {
+ let options = {compress: true};
let path = __dirname + "/../databases/backup/tbdata.nosql";
var stream = new rollers.RollingFileStream(path, noSqlFileSizeLimit, 150, options);
stream.write("");
stream.end();
insertBackupNoSqlCounter = 0;
- }
+ }
}
-
+
}
- else {
+ else
+ {
//logger.debug("Client unavailable. Data not sent !", JSON.stringify(data.data));
- instance.send(SEND_TO.debug, { "message": "Client unavailable. Data not sent !", "data": data.data });
+ instance.send(SEND_TO.debug, {"message":"Client unavailable. Data not sent !", "data": data.data });
- if (saveTelemetryOnError) {
+ if(saveTelemetryOnError)
+ {
//create new file from tbdata.nosql, if file size exceeds given limit, and clear tbdata.nosql
makeBackupFromDbFile();
@@ -276,63 +289,72 @@ exports.install = function(instance) {
instance.close = function(done) {
- if (clientReady) {
+ if(clientReady){
client.end();
clearInterval(sendWsStatusVar);
}
};
+
-
- function getDbBackupFileCounter(type) {
+ function getDbBackupFileCounter(type)
+ {
var files = fs.readdirSync(__dirname + "/../databases");
let counter = 0;
- for (var i = 0; i < files.length; i++) {
+ for(var i = 0; i < files.length; i++)
+ {
- if (files[i] == "tbdata.nosql") continue;
+ if(files[i] == "tbdata.nosql") continue;
- if (files[i].endsWith(".nosql")) {
+ if(files[i].endsWith(".nosql"))
+ {
let pos = files[i].indexOf(".");
- if (pos > -1) {
+ if(pos > -1)
+ {
let fileCounter = counter;
let firstDigit = files[i].slice(0, pos);
fileCounter = parseInt(firstDigit);
- if (isNaN(fileCounter)) fileCounter = 0;
+ if(isNaN(fileCounter)) fileCounter = 0;
//console.log("getDbBackupFileCounter digit:", files[i], firstDigit, fileCounter, isNaN(fileCounter), type);
-
- if (type == "max") {
- if (fileCounter > counter) {
+
+ if(type == "max")
+ {
+ if(fileCounter > counter)
+ {
counter = fileCounter;
}
}
- else if (type == "min") {
- if (counter == 0) counter = fileCounter;
+ else if(type == "min")
+ {
+ if(counter == 0) counter = fileCounter;
- if (fileCounter < counter) {
+ if(fileCounter < counter)
+ {
counter = fileCounter;
}
}
}
}
-
+
}
-
- if (type == "max") counter++;
+
+ if(type == "max") counter++;
return counter;
}
const makeBackupFromDbFile = async () => {
- if (!saveTelemetryOnError) return;
+ if(!saveTelemetryOnError) return;
//to avoid large file: tbdata.nosql
//init value is 0!
- if (insertNoSqlCounter > 0) {
+ if(insertNoSqlCounter > 0)
+ {
--insertNoSqlCounter;
return;
}
@@ -344,7 +366,8 @@ exports.install = function(instance) {
var stats = fs.statSync(source);
var fileSizeInBytes = stats.size;
- if (fileSizeInBytes > noSqlFileSizeLimit) {
+ if(fileSizeInBytes > noSqlFileSizeLimit)
+ {
let counter = 1;
counter = getDbBackupFileCounter("max");
@@ -358,20 +381,21 @@ exports.install = function(instance) {
//clear tbdata.nosql
fs.writeFileSync(source, "");
fs.truncateSync(source, 0);
-
+
}
}
const processDataFromDatabase = async () => {
-
- if (restore_from_backup <= 0) return;
+
+ if(restore_from_backup <= 0) return;
//calculate diff
const now = new Date();
let currentTime = now.getTime();
let diff = currentTime - lastRestoreTime;
- if ((diff / 1000) < restore_backup_wait) {
+ if( (diff / 1000) < restore_backup_wait)
+ {
//console.log("*********restore_backup_wait", diff, restore_backup_wait);
return;
}
@@ -385,55 +409,60 @@ exports.install = function(instance) {
let dataBase = 'tbdata';
var nosql;
- if (counter == 0) dataBase = 'tbdata';
+ if(counter == 0) dataBase = 'tbdata';
else dataBase = counter + "." + 'tbdata';
nosql = NOSQL(dataBase);
//select all data - use limit restore_from_backup
let records = await promisifyBuilder(nosql.find().take(restore_from_backup));
-
- for (let i = 0; i < records.length; i++) {
- if (clientReady) {
+
+ for(let i = 0; i < records.length; i++)
+ {
+ if(clientReady) {
let item = records[i];
let id = item.id;
- if (id !== undefined) {
+ if(id !== undefined)
+ {
//console.log("------------processDataFromDatabase - remove", id, dataBase, i);
try {
let message = JSON.parse(JSON.stringify(item));
delete message.id;
-
- client.publish("v1/gateway/telemetry", JSON.stringify(message), { qos: 1 });
-
+
+ client.publish("v1/gateway/telemetry", JSON.stringify(message), {qos:1});
+
//remove from database
await promisifyBuilder(nosql.remove().where("id", id));
- } catch (error) {
+ } catch(error) {
//process error
console.log("processDataFromDatabase", error);
}
}
-
+
}
- else {
+ else
+ {
processingData = false;
return;
}
}
- if (records.length > 0) {
+ if(records.length > 0)
+ {
//clean backup file
- if (counter > 0) nosql.clean();
+ if(counter > 0) nosql.clean();
}
//no data in db, remove
- if (records.length == 0) {
- if (counter > 0) nosql.drop();
+ if(records.length == 0)
+ {
+ if(counter > 0) nosql.drop();
}
const d = new Date();