wn/MqttDispatcher
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

debug improved

Browse Source
This commit is contained in:
Wolfgang Hottgenroth
2017-08-28 12:39:53 +02:00
parent 9356a81fd1
commit 5b909f5a7a
8 changed files with 55 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
dist/callchain.js vendored
View File

@ -23,6 +23,7 @@ class AChainItem extends events.EventEmitter {
this.next = next;
}
send(message) {
log.info(`Calling ${this.label} with ${message}`);
this.emit('yourturn', message);
}
}
@ -36,7 +37,7 @@ class AAsyncBaseChainItem extends AChainItem {
this.next.begin();
}
this.addListener('yourturn', (message) => {
log.info(`Calling ${this.toString()}`);
log.info(`Executing ${this.toString()}`);
this.func(message, this.next.send);
});
}
@ -51,7 +52,7 @@ class ABaseChainItem extends AChainItem {
this.next.begin();
}
this.addListener('yourturn', (message) => {
log.info(`Calling ${this.toString()}`);
log.info(`Executing ${this.toString()}`);
let result = this.func(message);
this.next.send(result);
});

26
dist/log.js vendored
View File

@ -2,6 +2,8 @@
Object.defineProperty(exports, "__esModule", { value: true });
const chalk = require("chalk");
const moment = require("moment");
const config = require("./config");
const nodemailer = require("nodemailer");
var Level;
(function (Level) {
Level[Level["All"] = 0] = "All";
@ -28,6 +30,30 @@ function setLevel(value) {
}
}
exports.setLevel = setLevel;
function sendAlarmMail(message) {
let transport = nodemailer.createTransport({
host: config.dict.smtpHost,
port: config.dict.smtpPort,
secure: false,
tls: {
rejectUnauthorized: false
}
});
let mail = {
from: config.dict.smtpSender,
to: config.dict.smtpReceiver,
subject: "Alarm from Dispatcher",
text: message
};
transport.sendMail(mail)
.then((v) => {
info(`Alarm mail sent, ${message}, ${v.response}`);
})
.catch((reason) => {
error(`Failure when sending alarm mail: ${message}, ${reason}`);
});
}
exports.sendAlarmMail = sendAlarmMail;
function info(message) {
if (level < Level.NoDebugNoInfo) {
console.log(`${timestamp()} ${chalk.bold.cyan('[ II ]')} ${message}`);

2
dist/main.js vendored
View File

@ -13,8 +13,10 @@ const ESP_THERM_TOPIC = 'IoT/espThermometer2/#';
dispatcher.register(ESP_THERM_TOPIC, 'toJson', EspThermToJson.espThermToJson);
let missingeventdetector = new MissingEventDetector.MissingEventDetector();
dispatcher.register(ESP_THERM_TOPIC, 'MissingEventDetector', missingeventdetector);
if (!config.dict.disableDatabaseAccess) {
let mongo = new MongoSave.MongoSave(config.dict.mongoDbUrl);
dispatcher.register(ESP_THERM_TOPIC, 'MongoSave', mongo);
}
dispatcher.exec();
log.info("Dispatcher running");
//# sourceMappingURL=main.js.map

32
dist/missingeventdetector.js vendored
View File

@ -1,10 +1,8 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const nodemailer = require("nodemailer");
const log = require("./log");
const CallChain = require("./callchain");
const Processor = require("./processor");
const config = require("./config");
const CHECK_PERIOD = 60; // seconds
class ClientEntry {
}
@ -12,40 +10,20 @@ class MissingEventProcessor extends Processor.AProcessor {
constructor() {
super("MissingEventProcessor");
this.clientMap = new Map();
this.smtp = nodemailer.createTransport({
host: config.dict.smtpHost,
port: config.dict.smtpPort,
secure: false,
tls: {
rejectUnauthorized: false
}
});
this.timer = setInterval(() => {
this.clientMap.forEach((value, key) => {
let currentTime = new Date().getTime();
let elapsedTime = currentTime - value.lastEvent;
log.info(`Checking ${key}, elapsed: ${elapsedTime / 1000}, avg. delay: ${value.avgDelay / 1000}`);
log.info(`MissingEventProcessor: Checking ${key}, elapsed: ${elapsedTime / 1000}, avg. delay: ${value.avgDelay / 1000}`);
if ((value.avgDelay != 0) && (elapsedTime > (value.avgDelay * 3))) {
let mail = {
from: config.dict.smtpSender,
to: config.dict.smtpReceiver,
subject: `Missing Event Detected for ${key}`,
text: `Missing Event Detected: ${key}, elapsed: ${elapsedTime / 1000}, avg. delay: ${value.avgDelay / 1000}`
};
this.smtp.sendMail(mail)
.then((v) => {
log.info(`Missing event info mail sent, ${v.response}`);
})
.catch((reason) => {
log.error(`Failure when sending missing event info: ${reason}`);
});
log.sendAlarmMail(`Missing Event Detected: ${key}, elapsed: ${elapsedTime / 1000}, avg. delay: ${value.avgDelay / 1000}`);
}
});
}, CHECK_PERIOD * 1000);
}
process(message) {
let client = message.metadata.client;
log.info(`Event for client ${client}`);
log.info(`MissingEventProcessor: Event for client ${client}`);
let currentTime = new Date().getTime();
if (this.clientMap.has(client)) {
let clientEntry = this.clientMap.get(client);
@ -58,7 +36,7 @@ class MissingEventProcessor extends Processor.AProcessor {
clientEntry.avgDelay = clientEntry.delaySum / clientEntry.count;
}
this.clientMap.set(client, clientEntry);
log.info(`Entry for ${client} updated`);
log.info(`MissingEventProcessor: Entry for ${client} updated`);
}
else {
let clientEntry = new ClientEntry();
@ -69,7 +47,7 @@ class MissingEventProcessor extends Processor.AProcessor {
clientEntry.delaySum = 0;
clientEntry.avgDelay = 0;
this.clientMap.set(client, clientEntry);
log.info(`Entry for ${client} inserted`);
log.info(`MissingEventProcessor: Entry for ${client} inserted`);
}
}
}

10
dist/mqttdispatcher.js vendored
View File

@ -79,22 +79,22 @@ class MqttDispatcher {
for (let field in registeredTopicFields) {
let regField = registeredTopicFields[field];
let recvField = receivedTopicFields[field];
log.info(`recv: ${recvField}, reg: ${regField}`);
log.debug(`topicMatch: recv: ${recvField}, reg: ${regField}`);
if (regField === "#") {
log.info('true');
log.debug('topicMatch: true 1');
return true;
}
if (regField != recvField && regField != "+") {
log.info('false');
log.debug('topicMatch: false 1');
return false;
}
}
if (registeredTopicFields.length == receivedTopicFields.length) {
log.info('true');
log.debug('topicMatch: true 2');
return true;
}
else {
log.info('false');
log.debug('topicMatch: false 2');
return false;
}
}

5
src/callchain.ts
View File

@ -39,6 +39,7 @@ export abstract class AChainItem extends events.EventEmitter implements Receivab
}
public send(message : any) : void {
log.info(`Calling ${this.label} with ${message}`)
this.emit('yourturn', message)
}
@ -57,7 +58,7 @@ export abstract class AAsyncBaseChainItem extends AChainItem {
this.next.begin()
}
this.addListener('yourturn', (message : any) : void => {
log.info(`Calling ${this.toString()}`)
log.info(`Executing ${this.toString()}`)
this.func(message, this.next.send)
})
}
@ -75,7 +76,7 @@ export abstract class ABaseChainItem extends AChainItem {
this.next.begin()
}
this.addListener('yourturn', (message : any) : void => {
log.info(`Calling ${this.toString()}`)
log.info(`Executing ${this.toString()}`)
let result : any = this.func(message)
this.next.send(result)
})

8
src/missingeventdetector.ts
View File

@ -29,7 +29,7 @@ class MissingEventProcessor extends Processor.AProcessor {
this.clientMap.forEach((value : ClientEntry, key : string) : void => {
let currentTime : number = new Date().getTime()
let elapsedTime : number = currentTime - value.lastEvent
log.info(`Checking ${key}, elapsed: ${elapsedTime / 1000}, avg. delay: ${value.avgDelay / 1000}`)
log.info(`MissingEventProcessor: Checking ${key}, elapsed: ${elapsedTime / 1000}, avg. delay: ${value.avgDelay / 1000}`)
if ((value.avgDelay != 0) && (elapsedTime > (value.avgDelay * 3))) {
log.sendAlarmMail(`Missing Event Detected: ${key}, elapsed: ${elapsedTime / 1000}, avg. delay: ${value.avgDelay / 1000}`)
@ -40,7 +40,7 @@ class MissingEventProcessor extends Processor.AProcessor {
protected process(message : any) : void {
let client = message.metadata.client
log.info(`Event for client ${client}`)
log.info(`MissingEventProcessor: Event for client ${client}`)
let currentTime : number = new Date().getTime()
if (this.clientMap.has(client)) {
let clientEntry = <ClientEntry>this.clientMap.get(client)
@ -53,7 +53,7 @@ class MissingEventProcessor extends Processor.AProcessor {
clientEntry.avgDelay = clientEntry.delaySum / clientEntry.count
}
this.clientMap.set(client, clientEntry)
log.info(`Entry for ${client} updated`)
log.info(`MissingEventProcessor: Entry for ${client} updated`)
} else {
let clientEntry : ClientEntry = new ClientEntry()
clientEntry.client = client
@ -63,7 +63,7 @@ class MissingEventProcessor extends Processor.AProcessor {
clientEntry.delaySum = 0
clientEntry.avgDelay = 0
this.clientMap.set(client, clientEntry)
log.info(`Entry for ${client} inserted`)
log.info(`MissingEventProcessor: Entry for ${client} inserted`)
}
}
}

10
src/mqttdispatcher.ts
View File

@ -104,21 +104,21 @@ export class MqttDispatcher implements IDispatcher {
for (let field in registeredTopicFields) {
let regField = registeredTopicFields[field]
let recvField = receivedTopicFields[field]
log.info(`recv: ${recvField}, reg: ${regField}`)
log.debug(`topicMatch: recv: ${recvField}, reg: ${regField}`)
if (regField === "#") {
log.info('true')
log.debug('topicMatch: true 1')
return true
}
if (regField != recvField && regField != "+") {
log.info('false')
log.debug('topicMatch: false 1')
return false
}
}
if (registeredTopicFields.length == receivedTopicFields.length) {
log.info('true')
log.debug('topicMatch: true 2')
return true
} else {
log.info('false')
log.debug('topicMatch: false 2')
return false
}