103 lines
4.0 KiB
JavaScript
103 lines
4.0 KiB
JavaScript
"use strict";
|
|
Object.defineProperty(exports, "__esModule", { value: true });
|
|
const Mqtt = require("mqtt");
|
|
const log = require("./log");
|
|
const callchain = require("./callchain");
|
|
const fs = require("fs");
|
|
const MQTT_BROKER_DEFAULT_URL = "mqtt://localhost";
|
|
function passThrough(message) {
|
|
return message;
|
|
}
|
|
exports.passThrough = passThrough;
|
|
class MqttDispatcher {
|
|
constructor(mqttBrokerUrl, mqttUser, mqttPass, mqttCAFile) {
|
|
this.mqttOptions = {};
|
|
this.mqttBrokerUrl = (mqttBrokerUrl) ? mqttBrokerUrl : MQTT_BROKER_DEFAULT_URL;
|
|
if (mqttUser && mqttPass) {
|
|
this.mqttOptions.username = mqttUser;
|
|
this.mqttOptions.password = mqttPass;
|
|
}
|
|
if (mqttCAFile) {
|
|
this.mqttOptions.ca = fs.readFileSync(mqttCAFile, 'ascii');
|
|
this.mqttOptions.rejectUnauthorized = true;
|
|
}
|
|
this.topicHandlers = [];
|
|
}
|
|
register(topic, label, newChainItemOrCallbackFunc) {
|
|
let newChainItem;
|
|
if (newChainItemOrCallbackFunc instanceof callchain.AChainItem) {
|
|
newChainItem = newChainItemOrCallbackFunc;
|
|
}
|
|
else {
|
|
let myNewChainItem = new callchain.ChainItem(label);
|
|
myNewChainItem.registerFunc(newChainItemOrCallbackFunc);
|
|
newChainItem = myNewChainItem;
|
|
}
|
|
let done = false;
|
|
for (let topicHandler of this.topicHandlers) {
|
|
if (topicHandler.topic === topic) {
|
|
topicHandler.last.registerNext(newChainItem);
|
|
topicHandler.last = newChainItem;
|
|
done = true;
|
|
log.info(`additional callback ${newChainItem.toString()} added for topic ${topic}`);
|
|
}
|
|
}
|
|
if (!done) {
|
|
this.topicHandlers.push({ topic: topic, root: newChainItem, last: newChainItem });
|
|
log.info(`first callback ${newChainItem.toString()} added for topic ${topic}`);
|
|
}
|
|
}
|
|
exec() {
|
|
for (let topicHandler of this.topicHandlers) {
|
|
topicHandler.root.begin();
|
|
}
|
|
log.info(`connecting to ${this.mqttBrokerUrl}`);
|
|
this.mqttClient = Mqtt.connect(this.mqttBrokerUrl, this.mqttOptions);
|
|
this.mqttClient.on('error', log.error);
|
|
this.mqttClient.on('connect', () => {
|
|
log.info("connected to mqtt broker");
|
|
for (let topicHandler of this.topicHandlers) {
|
|
this.mqttClient.subscribe(topicHandler.topic);
|
|
}
|
|
});
|
|
this.mqttClient.on('message', (topic, payload, packet) => {
|
|
log.info(`message received, topic ${topic}, payload ${payload}`);
|
|
if (packet.retain) {
|
|
log.info("IS RETAINED");
|
|
}
|
|
for (let topicHandler of this.topicHandlers) {
|
|
if (this.topicMatch(topicHandler.topic, topic)) {
|
|
log.info(`received topic ${topic} matches registered topic ${topicHandler.topic}`);
|
|
topicHandler.root.send(payload);
|
|
}
|
|
}
|
|
});
|
|
}
|
|
topicMatch(registeredTopic, receivedTopic) {
|
|
let registeredTopicFields = registeredTopic.split('/');
|
|
let receivedTopicFields = receivedTopic.split('/');
|
|
for (let field in registeredTopicFields) {
|
|
let regField = registeredTopicFields[field];
|
|
let recvField = receivedTopicFields[field];
|
|
log.debug(`topicMatch: recv: ${recvField}, reg: ${regField}`);
|
|
if (regField === "#") {
|
|
log.debug('topicMatch: true 1');
|
|
return true;
|
|
}
|
|
if (regField != recvField && regField != "+") {
|
|
log.debug('topicMatch: false 1');
|
|
return false;
|
|
}
|
|
}
|
|
if (registeredTopicFields.length == receivedTopicFields.length) {
|
|
log.debug('topicMatch: true 2');
|
|
return true;
|
|
}
|
|
else {
|
|
log.debug('topicMatch: false 2');
|
|
return false;
|
|
}
|
|
}
|
|
}
|
|
exports.MqttDispatcher = MqttDispatcher;
|
|
//# sourceMappingURL=mqttdispatcher.js.map
|