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

typescriptifying

Browse Source
This commit is contained in:
Wolfgang Hottgenroth
2018-01-08 21:54:59 +01:00
parent c1fa639ae4
commit 9e39d74084
12 changed files with 607 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
dispatcher_ng.conf Normal file
View File

@ -0,0 +1,10 @@
{
"brokerUrl": "mqtt://172.16.2.15:1883",
"brokerUser": "",
"brokerPass": "",
"brokerCa": "",
"smtpHost": "localhost",
"smtpPort": 25,
"smtpSender": "dispatcher@hottis.de",
"smtpReceiver": "woho@hottis.de"
}

87
dist/MqttDispatcher.js vendored Normal file
View File

@ -0,0 +1,87 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const logger = require("./log");
const Mqtt = require("mqtt");
const fs = require("fs");
const config = require("./config");
class MqttHandler {
constructor() {
this.mqttOptions = {};
this.mqttBrokerUrl = config.dict.brokerUrl;
if (config.dict.brokerUser && config.dict.brokerPass) {
this.mqttOptions.username = config.dict.brokerUser;
this.mqttOptions.password = config.dict.brokerPass;
}
if (config.dict.brokerCa) {
this.mqttOptions.ca = fs.readFileSync(config.dict.brokerCa, 'ascii');
this.mqttOptions.rejectUnauthorized = true;
}
this.topicHandlers = [];
}
register(topics, cb) {
topics.forEach((topic) => {
this.topicHandlers.push({ topic: topic, callback: cb });
logger.info(`additional callback registered for ${topic}`);
});
}
exec() {
logger.info(`connecting to ${this.mqttBrokerUrl}`);
this.mqttClient = Mqtt.connect(this.mqttBrokerUrl, this.mqttOptions);
this.mqttClient.on('error', (err) => {
logger.error(`Error in mqttHandler: ${err}`);
});
this.mqttClient.on('connect', () => {
this.mqttClient.publish('dispatcher_ng/status', 'dispatcher_ng running');
this.mqttClient.subscribe('dispatcher_ng/cmd');
this.topicHandlers.forEach((topicHandler) => {
this.mqttClient.subscribe(topicHandler.topic);
logger.info(`${topicHandler.topic} subscribed`);
});
logger.info('mqtt connection established');
});
this.mqttClient.on('message', (topic, payload, packet) => {
if (!packet.retain) {
let payloadStr = payload.toString('UTF-8');
logger.info(`message received on topic ${topic}: ${payload}`);
this.processMessage(topic, payloadStr);
}
});
}
processMessage(topic, payload) {
let found = false;
this.topicHandlers.forEach((topicHandler) => {
// logger.warn(`Test: ${subscribedTopic}, ${topic}`);
// console.log(`Test: ${subscribedTopic}, ${topic}`);
if (topicHandler.topic == topic) {
// logger.warn('1');
topicHandler.callback(topic, payload);
found = true;
}
else if (topicHandler.topic.endsWith('#') &&
(topicHandler.topic.substring(0, topicHandler.topic.length - 1) ==
topic.substring(0, topicHandler.topic.length - 1))) {
// logger.warn('2');
// console.log('2');
topicHandler.callback(topic, payload);
found = true;
}
});
return found;
}
send(topic, payload, internalFirst = false) {
let sent = false;
if (internalFirst) {
logger.info(`Try internal sending: ${topic}`);
sent = this.processMessage(topic, payload);
}
if (!sent) {
logger.info(`External sending required: ${topic}`);
this.mqttClient.publish(topic, payload);
}
else {
logger.info(`Internally delivered: ${topic}`);
}
}
}
exports.mqttHandler = new MqttHandler();
//# sourceMappingURL=MqttDispatcher.js.map

16
dist/config.js vendored Normal file
View File

@ -0,0 +1,16 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const fs = require("fs");
const cmdargs = require("command-line-args");
const logger = require("./log");
const OPTION_DEFINITIONS = [
{ name: 'verbose', alias: 'v', type: Boolean },
{ name: 'config', alias: 'c', type: String, defaultValue: '~/dispatcher_ng.conf' }
];
function readConfig() {
let options = cmdargs(OPTION_DEFINITIONS);
exports.dict = JSON.parse(fs.readFileSync(options.config, "utf8"));
logger.info(JSON.stringify(exports.dict));
}
exports.readConfig = readConfig;
//# sourceMappingURL=config.js.map

82
dist/log.js vendored Normal file
View File

@ -0,0 +1,82 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const moment = require("moment");
const config = require("./config");
const nodemailer = require("nodemailer");
var Level;
(function (Level) {
Level[Level["All"] = 0] = "All";
Level[Level["NoDebug"] = 1] = "NoDebug";
Level[Level["NoDebugNoInfo"] = 2] = "NoDebugNoInfo";
Level[Level["NoDebugNoInfoNoWarning"] = 3] = "NoDebugNoInfoNoWarning";
})(Level || (Level = {}));
var level = Level.NoDebug;
function timestamp() {
return moment().format('HH:mm:ss.SSS');
}
function setLevel(value) {
switch (value) {
case 'info':
level = Level.NoDebug;
break;
case 'warn':
level = Level.NoDebugNoInfo;
break;
case 'error':
level = Level.NoDebugNoInfoNoWarning;
break;
default: level = Level.All;
}
}
exports.setLevel = setLevel;
function sendAlarmMail(subject, 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: subject,
text: message
};
transport.sendMail(mail)
.then((v) => {
info(`Mail sent, ${subject}, ${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()} [ II ] ${message}`);
}
}
exports.info = info;
function warn(message) {
if (level < Level.NoDebugNoInfoNoWarning) {
console.log(`${timestamp()} [ WW ] ${message}`);
}
}
exports.warn = warn;
function error(message) {
console.log(`${timestamp()} [ EE ] ${message}`);
}
exports.error = error;
function success(message) {
console.log(`${timestamp()} [ OK ] ${message}`);
}
exports.success = success;
function debug(message) {
if (level < Level.NoDebug) {
console.log(`${timestamp()} [ DB ] ${message}`);
}
}
exports.debug = debug;
//# sourceMappingURL=log.js.map

17
dist/main.js vendored
View File

@ -1,11 +1,18 @@
class Test {
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const logger = require("./log");
const config = require("./config");
// import { mqttHandler } from './MqttDispatcher'
config.readConfig();
class Dispatcher {
constructor() {
console.log("Test constructed");
logger.info("Dispatcher starting");
}
exec() {
console.log("Hello world");
logger.info("Hello world");
// mqttHandler.exec()
}
}
const test = new Test();
test.exec();
const dispatcher = new Dispatcher();
dispatcher.exec();
//# sourceMappingURL=main.js.map

172
package-lock.json generated
View File

@ -4,6 +4,70 @@
"lockfileVersion": 1,
"requires": true,
"dependencies": {
"@types/chalk": {
"version": "2.2.0",
"resolved": "https://registry.npmjs.org/@types/chalk/-/chalk-2.2.0.tgz",
"integrity": "sha512-1zzPV9FDe1I/WHhRkf9SNgqtRJWZqrBWgu7JGveuHmmyR9CnAPCie2N/x+iHrgnpYBIcCJWHBoMRv2TRWktsvw==",
"dev": true,
"requires": {
"chalk": "2.3.0"
}
},
"@types/command-line-args": {
"version": "4.0.2",
"resolved": "https://registry.npmjs.org/@types/command-line-args/-/command-line-args-4.0.2.tgz",
"integrity": "sha512-lLN/Rtud+SONdclROGc8CnZeZ/ujQ69U4otFMtoCfCdLfs2x2R2lJFSy583uI9dnnsd6I9Pdy/Mw3efqxO/2bg==",
"dev": true
},
"@types/moment": {
"version": "2.13.0",
"resolved": "https://registry.npmjs.org/@types/moment/-/moment-2.13.0.tgz",
"integrity": "sha1-YE69GJvDvDShVIaJQE5hoqSqyJY=",
"dev": true,
"requires": {
"moment": "2.20.1"
}
},
"@types/mqtt": {
"version": "2.5.0",
"resolved": "https://registry.npmjs.org/@types/mqtt/-/mqtt-2.5.0.tgz",
"integrity": "sha512-n+0/ErBin30j+UbhcHGK/STjHjh65k85WNR6NlUjRG0g9yctpF12pS+SOkwz0wmp+7momAo9Cyi4Wmvy8UsCQg==",
"dev": true,
"requires": {
"mqtt": "2.15.0"
}
},
"@types/node": {
"version": "8.5.8",
"resolved": "https://registry.npmjs.org/@types/node/-/node-8.5.8.tgz",
"integrity": "sha512-8KmlRxwbKZfjUHFIt3q8TF5S2B+/E5BaAoo/3mgc5h6FJzqxXkCK/VMetO+IRDtwtU6HUvovHMBn+XRj7SV9Qg==",
"dev": true
},
"@types/nodemailer": {
"version": "4.3.1",
"resolved": "https://registry.npmjs.org/@types/nodemailer/-/nodemailer-4.3.1.tgz",
"integrity": "sha512-6Y+W44okdSco6w3zhahnwJXbONrCrfgi0LnF/aQAdHFnrel4VYwfF0u81NNKHTbweKacu2TVDv/2BIZhgutxIA==",
"dev": true,
"requires": {
"@types/node": "8.5.8"
}
},
"ansi-styles": {
"version": "3.2.0",
"resolved": "https://registry.npmjs.org/ansi-styles/-/ansi-styles-3.2.0.tgz",
"integrity": "sha512-NnSOmMEYtVR2JVMIGTzynRkkaxtiq1xnFBcdQD/DnNCYPoEPsVJhM98BDyaoNOQIi7p4okdi3E27eN7GQbsUug==",
"requires": {
"color-convert": "1.9.1"
}
},
"array-back": {
"version": "2.0.0",
"resolved": "https://registry.npmjs.org/array-back/-/array-back-2.0.0.tgz",
"integrity": "sha512-eJv4pLLufP3g5kcZry0j6WXpIbzYw9GUB4mVJZno9wfwiBxbizTnHCw3VJb07cBihbFX48Y7oSrW9y+gt4glyw==",
"requires": {
"typical": "2.6.1"
}
},
"async-limiter": {
"version": "1.0.0",
"resolved": "https://registry.npmjs.org/async-limiter/-/async-limiter-1.0.0.tgz",
@ -40,6 +104,39 @@
"readable-stream": "2.3.3"
}
},
"chalk": {
"version": "2.3.0",
"resolved": "https://registry.npmjs.org/chalk/-/chalk-2.3.0.tgz",
"integrity": "sha512-Az5zJR2CBujap2rqXGaJKaPHyJ0IrUimvYNX+ncCy8PJP4ltOGTrHUIo097ZaL2zMeKYpiCdqDvS6zdrTFok3Q==",
"requires": {
"ansi-styles": "3.2.0",
"escape-string-regexp": "1.0.5",
"supports-color": "4.5.0"
}
},
"color-convert": {
"version": "1.9.1",
"resolved": "https://registry.npmjs.org/color-convert/-/color-convert-1.9.1.tgz",
"integrity": "sha512-mjGanIiwQJskCC18rPR6OmrZ6fm2Lc7PeGFYwCmy5J34wC6F1PzdGL6xeMfmgicfYcNLGuVFA3WzXtIDCQSZxQ==",
"requires": {
"color-name": "1.1.3"
}
},
"color-name": {
"version": "1.1.3",
"resolved": "https://registry.npmjs.org/color-name/-/color-name-1.1.3.tgz",
"integrity": "sha1-p9BVi9icQveV3UIyj3QIMcpTvCU="
},
"command-line-args": {
"version": "4.0.7",
"resolved": "https://registry.npmjs.org/command-line-args/-/command-line-args-4.0.7.tgz",
"integrity": "sha512-aUdPvQRAyBvQd2n7jXcsMDz68ckBJELXNzBybCHOibUWEg0mWTnaYCSRU8h9R+aNRSvDihJtssSRCiDRpLaezA==",
"requires": {
"array-back": "2.0.0",
"find-replace": "1.0.3",
"typical": "2.6.1"
}
},
"commist": {
"version": "1.0.0",
"resolved": "https://registry.npmjs.org/commist/-/commist-1.0.0.tgz",
@ -88,11 +185,35 @@
"once": "1.4.0"
}
},
"escape-string-regexp": {
"version": "1.0.5",
"resolved": "https://registry.npmjs.org/escape-string-regexp/-/escape-string-regexp-1.0.5.tgz",
"integrity": "sha1-G2HAViGQqN/2rjuyzwIAyhMLhtQ="
},
"extend": {
"version": "3.0.1",
"resolved": "https://registry.npmjs.org/extend/-/extend-3.0.1.tgz",
"integrity": "sha1-p1Xqe8Gt/MWjHOfnYtuq3F5jZEQ="
},
"find-replace": {
"version": "1.0.3",
"resolved": "https://registry.npmjs.org/find-replace/-/find-replace-1.0.3.tgz",
"integrity": "sha1-uI5zZNLZyVlVnziMZmcNYTBEH6A=",
"requires": {
"array-back": "1.0.4",
"test-value": "2.1.0"
},
"dependencies": {
"array-back": {
"version": "1.0.4",
"resolved": "https://registry.npmjs.org/array-back/-/array-back-1.0.4.tgz",
"integrity": "sha1-ZEun8JX3/898Q7Xw3DnTwfA8Bjs=",
"requires": {
"typical": "2.6.1"
}
}
}
},
"fs.realpath": {
"version": "1.0.0",
"resolved": "https://registry.npmjs.org/fs.realpath/-/fs.realpath-1.0.0.tgz",
@ -137,6 +258,11 @@
"unique-stream": "2.2.1"
}
},
"has-flag": {
"version": "2.0.0",
"resolved": "https://registry.npmjs.org/has-flag/-/has-flag-2.0.0.tgz",
"integrity": "sha1-6CB68cx7MNRGzHC3NLXovhj4jVE="
},
"help-me": {
"version": "1.1.0",
"resolved": "https://registry.npmjs.org/help-me/-/help-me-1.1.0.tgz",
@ -287,10 +413,10 @@
"safe-buffer": "5.1.1"
}
},
"node-cron": {
"version": "1.2.1",
"resolved": "https://registry.npmjs.org/node-cron/-/node-cron-1.2.1.tgz",
"integrity": "sha1-jJC8XccjpWKJsHhmVatKHEy2A2g="
"nodemailer": {
"version": "4.4.1",
"resolved": "https://registry.npmjs.org/nodemailer/-/nodemailer-4.4.1.tgz",
"integrity": "sha512-1bnszJJXatcHJhLpxQ1XMkLDjCjPKvGKMtRQ73FOsoNln3UQjddEQmz6fAwM3aj0GtQ3dQX9qtMHPelz63GU7A=="
},
"once": {
"version": "1.4.0",
@ -412,6 +538,33 @@
"safe-buffer": "5.1.1"
}
},
"supports-color": {
"version": "4.5.0",
"resolved": "https://registry.npmjs.org/supports-color/-/supports-color-4.5.0.tgz",
"integrity": "sha1-vnoN5ITexcXN34s9WRJQRJEvY1s=",
"requires": {
"has-flag": "2.0.0"
}
},
"test-value": {
"version": "2.1.0",
"resolved": "https://registry.npmjs.org/test-value/-/test-value-2.1.0.tgz",
"integrity": "sha1-Edpv9nDzRxpztiXKTz/c97t0gpE=",
"requires": {
"array-back": "1.0.4",
"typical": "2.6.1"
},
"dependencies": {
"array-back": {
"version": "1.0.4",
"resolved": "https://registry.npmjs.org/array-back/-/array-back-1.0.4.tgz",
"integrity": "sha1-ZEun8JX3/898Q7Xw3DnTwfA8Bjs=",
"requires": {
"typical": "2.6.1"
}
}
}
},
"through2": {
"version": "2.0.3",
"resolved": "https://registry.npmjs.org/through2/-/through2-2.0.3.tgz",
@ -444,6 +597,17 @@
"resolved": "https://registry.npmjs.org/typedarray/-/typedarray-0.0.6.tgz",
"integrity": "sha1-hnrHTjhkGHsdPUfZlqeOxciDB3c="
},
"typescript": {
"version": "2.6.2",
"resolved": "https://registry.npmjs.org/typescript/-/typescript-2.6.2.tgz",
"integrity": "sha1-PFtv1/beCRQmkCfwPAlGdY92c6Q=",
"dev": true
},
"typical": {
"version": "2.6.1",
"resolved": "https://registry.npmjs.org/typical/-/typical-2.6.1.tgz",
"integrity": "sha1-XAgOXWYcu+OCWdLnCjxyU+hziB0="
},
"ultron": {
"version": "1.1.1",
"resolved": "https://registry.npmjs.org/ultron/-/ultron-1.1.1.tgz",

8
package.json
View File

@ -19,12 +19,20 @@
},
"homepage": "https://gitlab.com/wolutator/dispatcher_ng#README",
"dependencies": {
"chalk": "^2.3.0",
"command-line-args": "^4.0.7",
"moment": "^2.20.1",
"mqtt": "^2.15.0",
"nodemailer": "^4.4.1",
"simple-node-logger": "^0.93.33"
},
"devDependencies": {
"@types/chalk": "^2.2.0",
"@types/command-line-args": "^4.0.2",
"@types/moment": "^2.13.0",
"@types/mqtt": "^2.5.0",
"@types/node": "^8.5.8",
"@types/nodemailer": "^4.3.1",
"typescript": "^2.6.2"
}
}

107
src/MqttDispatcher.ts Normal file
View File

@ -0,0 +1,107 @@
import * as logger from './log'
import * as Mqtt from 'mqtt'
import * as fs from 'fs'
import * as config from './config'
export type TopicCallbackFunc = (topic: string, payload: string) => void
export interface TopicHandler {
topic: string,
callback: TopicCallbackFunc
}
class MqttHandler {
private mqttClient: Mqtt.Client
private mqttOptions: Mqtt.IClientOptions = {}
private mqttBrokerUrl: string
private topicHandlers: TopicHandler[]
constructor() {
this.mqttBrokerUrl = config.dict.brokerUrl
if (config.dict.brokerUser && config.dict.brokerPass) {
this.mqttOptions.username = config.dict.brokerUser
this.mqttOptions.password = config.dict.brokerPass
}
if (config.dict.brokerCa) {
this.mqttOptions.ca = fs.readFileSync(config.dict.brokerCa, 'ascii')
this.mqttOptions.rejectUnauthorized = true
}
this.topicHandlers = []
}
register(topics: string[], cb: TopicCallbackFunc) : void {
topics.forEach((topic) => {
this.topicHandlers.push({topic: topic, callback: cb})
logger.info(`additional callback registered for ${topic}`)
})
}
exec() : void {
logger.info(`connecting to ${this.mqttBrokerUrl}`)
this.mqttClient = Mqtt.connect(this.mqttBrokerUrl, this.mqttOptions)
this.mqttClient.on('error', (err) => {
logger.error(`Error in mqttHandler: ${err}`)
})
this.mqttClient.on('connect', () : void => {
this.mqttClient.publish('dispatcher_ng/status', 'dispatcher_ng running')
this.mqttClient.subscribe('dispatcher_ng/cmd')
this.topicHandlers.forEach((topicHandler) => {
this.mqttClient.subscribe(topicHandler.topic)
logger.info(`${topicHandler.topic} subscribed`)
})
logger.info('mqtt connection established')
})
this.mqttClient.on('message', (topic: string, payload: Buffer, packet : Mqtt.IPublishPacket): void => {
if (! packet.retain) {
let payloadStr : string = payload.toString('UTF-8')
logger.info(`message received on topic ${topic}: ${payload}`)
this.processMessage(topic, payloadStr)
}
})
}
processMessage(topic: string, payload: string) : boolean {
let found = false;
this.topicHandlers.forEach((topicHandler) => {
// logger.warn(`Test: ${subscribedTopic}, ${topic}`);
// console.log(`Test: ${subscribedTopic}, ${topic}`);
if (topicHandler.topic == topic) {
// logger.warn('1');
topicHandler.callback(topic, payload)
found = true
} else if (topicHandler.topic.endsWith('#') &&
(topicHandler.topic.substring(0, topicHandler.topic.length-1) ==
topic.substring(0, topicHandler.topic.length-1))) {
// logger.warn('2');
// console.log('2');
topicHandler.callback(topic, payload)
found = true
}
})
return found
}
send(topic: string, payload: string, internalFirst: boolean = false) : void {
let sent = false
if (internalFirst) {
logger.info(`Try internal sending: ${topic}`)
sent = this.processMessage(topic, payload)
}
if (! sent) {
logger.info(`External sending required: ${topic}`)
this.mqttClient.publish(topic, payload)
} else {
logger.info(`Internally delivered: ${topic}`)
}
}
}
export let mqttHandler = new MqttHandler()

19
src/config.ts Normal file
View File

@ -0,0 +1,19 @@
import * as fs from 'fs'
import * as cmdargs from 'command-line-args'
import * as logger from './log'
const OPTION_DEFINITIONS = [
{ name: 'verbose', alias: 'v', type: Boolean },
{ name: 'config', alias: 'c', type: String, defaultValue: '~/dispatcher_ng.conf' }
];
export let dict : any
export function readConfig() {
let options = cmdargs(OPTION_DEFINITIONS)
dict = JSON.parse(fs.readFileSync(options.config, "utf8"))
logger.info(JSON.stringify(dict))
}

79
src/log.ts Normal file
View File

@ -0,0 +1,79 @@
import * as moment from 'moment'
import * as config from './config'
import * as nodemailer from 'nodemailer'
enum Level {
All,
NoDebug,
NoDebugNoInfo,
NoDebugNoInfoNoWarning
}
var level = Level.NoDebug
function timestamp(): string {
return moment().format('HH:mm:ss.SSS')
}
export function setLevel(value: string): void {
switch (value) {
case 'info': level = Level.NoDebug; break
case 'warn': level = Level.NoDebugNoInfo; break
case 'error': level = Level.NoDebugNoInfoNoWarning; break
default: level = Level.All
}
}
export function sendAlarmMail(subject : string, message : string): void {
let transport = nodemailer.createTransport({
host: config.dict.smtpHost,
port: config.dict.smtpPort,
secure: false,
tls: {
rejectUnauthorized: false
}
});
let mail : nodemailer.SendMailOptions = {
from: config.dict.smtpSender,
to: config.dict.smtpReceiver,
subject: subject,
text: message
};
transport.sendMail(mail)
.then((v : nodemailer.SentMessageInfo) => {
info(`Mail sent, ${subject}, ${message}, ${v.response}`)
})
.catch((reason : any) => {
error(`Failure when sending alarm mail: ${message}, ${reason}`)
})
}
export function info(message: string): void {
if (level < Level.NoDebugNoInfo) {
console.log(`${timestamp()} [ II ] ${message}`)
}
}
export function warn(message: string): void {
if (level < Level.NoDebugNoInfoNoWarning) {
console.log(`${timestamp()} [ WW ] ${message}`)
}
}
export function error(message: string): void {
console.log(`${timestamp()} [ EE ] ${message}`)
}
export function success(message: string): void {
console.log(`${timestamp()} [ OK ] ${message}`)
}
export function debug(message: string): void {
if (level < Level.NoDebug) {
console.log(`${timestamp()} [ DB ] ${message}`)
}
}

24
src/main.ts
View File

@ -1,14 +1,28 @@
class Test {
import * as logger from './log'
import * as config from './config'
// import { mqttHandler } from './MqttDispatcher'
config.readConfig()
class Dispatcher {
constructor() {
console.log("Test constructed")
logger.info("Dispatcher starting")
}
exec() : void {
console.log("Hello world")
logger.info("Hello world")
// mqttHandler.exec()
}
}
const test = new Test()
test.exec()
const dispatcher = new Dispatcher()
dispatcher.exec()

91
src/mqttHandler.js
View File

@ -1,91 +0,0 @@
let logger = require('./log')
logger.info('mqttHandler executed')
var mqtt = require('mqtt');
var client = undefined;
var topicCallbacks = {};
function start() {
client = mqtt.connect('mqtt://172.16.2.16');
client.on('error', (err) => {
logger.error(`Error in mqttHandler: ${err}`)
});
client.on('connect', () => {
client.publish('dispatcher_ng/status', 'dispatcher_ng running');
client.subscribe('dispatcher_ng/cmd');
Object.keys(topicCallbacks).forEach((topic) => {
client.subscribe(topic);
logger.info(`${topic} subscribed`);
});
logger.info('mqtt connection established');
});
client.on('message', (topic, payload, packet) => {
if (! packet.retain) {
payload = payload.toString('UTF-8');
logger.info(`message received on topic ${topic}: ${payload}`);
processMessage(topic, payload);
}
});
}
function processMessage(topic, payload) {
let found = false;
Object.keys(topicCallbacks).forEach((subscribedTopic) => {
// logger.warn(`Test: ${subscribedTopic}, ${topic}`);
// console.log(`Test: ${subscribedTopic}, ${topic}`);
if (subscribedTopic == topic) {
// logger.warn('1');
topicCallbacks[topic].forEach((cb) => { cb(topic, payload) });
found = true;
} else if (subscribedTopic.endsWith('#') &&
(subscribedTopic.substring(0, subscribedTopic.length-1) ==
topic.substring(0, subscribedTopic.length-1))) {
// logger.warn('2');
// console.log('2');
topicCallbacks[subscribedTopic].forEach((cb) => { cb(topic, payload) });
found = false;
}
});
return found;
}
function send(topic, payload, internalFirst = false) {
let sent = false;
if (internalFirst) {
logger.info(`Try internal sending: ${topic}`);
sent = processMessage(topic, payload);
}
if (! sent) {
logger.info(`External sending required: ${topic}`);
client.publish(topic, payload);
} else {
logger.info(`Internally delivered: ${topic}`);
}
}
function register(topics, cb) {
if (! (topics instanceof Array)) {
topics = [ topics ];
}
topics.forEach((topic) => {
if (topic in topicCallbacks) {
topicCallbacks[topic].push(cb);
logger.info(`additional callback registered for ${topic}`);
} else {
topicCallbacks[topic] = [ cb ];
logger.info(`first callback registered for ${topic}`);
}
})
}
module.exports = {
start,
send,
register
};