oms-is-citysys/citysys-flowserver
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

version 2025-02-01

Browse source
This commit is contained in:
rasta5man 2025-01-07 15:30:46 +01:00
parent ef7817bf22
commit c052887a1f
11 changed files with 368 additions and 700 deletions
Download patch file Download diff file Expand all files Collapse all files

67
flow/cloudmqttconnect.js
View file

@ -32,6 +32,7 @@ exports.html = `<div class="padding">
const { promisifyBuilder } = require('./helper/db_helper');
const fs = require('fs');
const mqtt = require('mqtt');
const nosql = NOSQL('tbdatacloud');
const SEND_TO = {
debug: 0,
@ -56,9 +57,6 @@ let lastRestoreTime = 0;
let sendClientError = true;
const nosql = NOSQL('tbdatacloud');
exports.install = function(instance) {
var client;
@ -112,11 +110,6 @@ exports.install = function(instance) {
});
client.on('reconnect', function() {
client.subscribe(`${o.topic}_backward`, (err) => {
if (!err) {
console.log("MQTT subscribed");
}
});
instance.status("Reconnecting", "yellow");
clientReady = false;
});
@ -139,18 +132,11 @@ exports.install = function(instance) {
instance.send(SEND_TO.rpcCall, {"topic":o.topic, "content":message });
});
client.on('close', function(err) {
client.on('close', function() {
clientReady = false;
if (err && err.toString().indexOf('Error')) {
instance.status("Err: "+err.code, "red");
instance.send(SEND_TO.debug, {"message":"Client CLOSE signal received !", "error":err, "opt":opts });
} else {
instance.status("Disconnected", "red");
instance.send(SEND_TO.debug, {"message":"Client CLOSE signal received !", "error":err, "opt":opts });
}
client.reconnect();
instance.status("Disconnected", "red");
instance.send(SEND_TO.debug, {"message":"Client CLOSE signal received !"});
});
client.on('error', function(err) {
@ -170,46 +156,30 @@ exports.install = function(instance) {
if(clientReady)
{
//do we have some data in backup file?
//if any, process data from database
//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(clientReady)
{
client.publish(`${o.topic}_forward`, data.data, {qos: 1});
//console.log("ondata..",data.data)
// Pokad ten error na 38 chápem tak tento parameter MUSÍ byt string...
// Tak to musim dekódovat na strane Cloudu zas
let stringifiedJson = JSON.stringify(data.data)
client.publish(`${o.topic}_forward`, stringifiedJson, {qos: 1});
}
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)
{
try {
let d = JSON.parse(data.data);
//create new file from tbdata.nosql, if file size exceeds given limit, and clear tbdata.nosql
makeBackupFromDbFile();
//create new file from tbdata.nosql, if file size exceeds given limit, and clear tbdata.nosql
makeBackupFromDbFile();
//write to tb
d.id = UID();
nosql.insert(d);
} catch(e) {
console.log("cloudconnect - unable to parse data from wsmqtt");
}
//write to tb
data.data.id = UID();
nosql.insert(data.data);
}
}
});
@ -358,12 +328,10 @@ exports.install = function(instance) {
try {
let o = JSON.parse(JSON.stringify(item));
delete o.id;
let message = JSON.stringify(o);
client.publish(`${o.topic}_forward`, message, {qos:1});
//console.log("db...", message)
let message = JSON.parse(JSON.stringify(item));
delete message.id;
client.publish(`${o.topic}_forward`, JSON.stringify(message), {qos:1});
//remove from database
await promisifyBuilder(nosql.remove().where("id", id));
@ -372,7 +340,6 @@ exports.install = function(instance) {
console.log("processDataFromDatabase", error);
}
}
}