file rename

2017-08-09 00:24:22 +02:00
parent dd7f313d0c
commit 09d7ea8ace
4 changed files with 3 additions and 3 deletions
--- a/dist/mqttdispatcher.js
+++ b/dist/mqttdispatcher.js
@ -0,0 +1,89 @@
+"use strict";
+Object.defineProperty(exports, "__esModule", { value: true });
+const Mqtt = require("mqtt");
+const log = require("./log");
+const callchain = require("./callchain");
+const MQTT_BROKER_DEFAULT_URL = "mqtt://localhost";
+function passThrough(message) {
+    return message;
+}
+exports.passThrough = passThrough;
+class MqttDispatcher {
+    constructor(mqttBrokerUrl) {
+        this._mqttBrokerUrl = (mqttBrokerUrl) ? mqttBrokerUrl : MQTT_BROKER_DEFAULT_URL;
+        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();
+        }
+        this._mqttClient = Mqtt.connect(this._mqttBrokerUrl);
+        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) => {
+            log.info(`message received, topic ${topic}, payload ${payload}`);
+            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.info(`recv: ${recvField}, reg: ${regField}`);
+            if (regField === "#") {
+                log.info('true');
+                return true;
+            }
+            if (regField != recvField && regField != "+") {
+                log.info('false');
+                return false;
+            }
+        }
+        if (registeredTopicFields.length == receivedTopicFields.length) {
+            log.info('true');
+            return true;
+        }
+        else {
+            log.info('false');
+            return false;
+        }
+    }
+}
+exports.MqttDispatcher = MqttDispatcher;
+//# sourceMappingURL=mqttdispatcher.js.map