@@ -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'];
+ let message = value[k[0]][0]['values']['_event']['message']['en'];
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,5 +183,4 @@ exports.install = function(instance) {
instance.on('options', instance.reconfigure);
setTimeout(instance.reconfigure, 10000);
-
};
diff --git a/flow/thermometer.js b/flow/thermometer.js
index fa015c6..fc7074d 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,24 +42,25 @@ 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);
}
@@ -70,12 +71,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");
@@ -85,7 +86,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 b2a218d..0eba804 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,22 +96,19 @@ 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);
@@ -119,11 +116,9 @@ 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 = {
@@ -139,13 +134,12 @@ 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;
@@ -170,8 +164,7 @@ exports.install = function(instance) {
connectToTbServer();
}
- function connectToTbServer()
- {
+ function connectToTbServer() {
var url = "mqtt://" + opts.host + ":" + opts.port;
console.log("MQTT URL: ", url);
@@ -179,7 +172,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;
@@ -198,31 +191,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';
@@ -238,45 +231,39 @@ 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
- {
- //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 });
- if(saveTelemetryOnError)
- {
+ }
+ 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 });
+
+ if (saveTelemetryOnError) {
//create new file from tbdata.nosql, if file size exceeds given limit, and clear tbdata.nosql
makeBackupFromDbFile();
@@ -289,72 +276,63 @@ 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;
}
@@ -366,8 +344,7 @@ 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");
@@ -381,21 +358,20 @@ 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;
}
@@ -409,60 +385,55 @@ 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();