support for multiple callbacks per topic started

This commit is contained in:
Wolfgang Hottgenroth
2017-07-27 23:53:23 +02:00
parent 7e236974c2
commit 04873adad2
2 changed files with 29 additions and 7 deletions

17
dist/mqttclient.js vendored
View File

@ -13,8 +13,19 @@ class MqttClient {
this._topicHandlers = []; this._topicHandlers = [];
} }
register(topic, callback) { register(topic, callback) {
this._topicHandlers.push({ topic, callback }); let done = false;
log.info(`handler registered for topic ${topic}`); for (let topicHandler of this._topicHandlers) {
if (topicHandler.topic === topic) {
topicHandler.callbacks.push(callback);
done = true;
log.info(`additional callback added for topic ${topic}`);
}
}
if (!done) {
let cbs = [callback];
this._topicHandlers.push({ topic: topic, callbacks: cbs });
log.info(`first callback added for topic ${topic}`);
}
} }
exec() { exec() {
this._mqttClient = Mqtt.connect(this._mqttBrokerUrl); this._mqttClient = Mqtt.connect(this._mqttBrokerUrl);
@ -30,7 +41,7 @@ class MqttClient {
for (let topicHandler of this._topicHandlers) { for (let topicHandler of this._topicHandlers) {
if (this.topicMatch(topicHandler.topic, topic)) { if (this.topicMatch(topicHandler.topic, topic)) {
log.info(`received topic ${topic} matches registered topic ${topicHandler.topic}`); log.info(`received topic ${topic} matches registered topic ${topicHandler.topic}`);
topicHandler.callback(payload); // topicHandler.callback(payload)
} }
} }
}); });

View File

@ -7,7 +7,7 @@ type TopicHandlerCallback = (message: any) => any
interface TopicHandler { interface TopicHandler {
topic: string, topic: string,
callback: TopicHandlerCallback callbacks: TopicHandlerCallback[]
} }
export function passThrough(message: any) { export function passThrough(message: any) {
@ -25,8 +25,19 @@ export class MqttClient {
} }
register(topic: string, callback: TopicHandlerCallback) : void { register(topic: string, callback: TopicHandlerCallback) : void {
this._topicHandlers.push({topic, callback}) let done: boolean = false
log.info(`handler registered for topic ${topic}`) for (let topicHandler of this._topicHandlers) {
if (topicHandler.topic === topic) {
topicHandler.callbacks.push(callback)
done = true
log.info(`additional callback added for topic ${topic}`)
}
}
if (! done) {
let cbs : TopicHandlerCallback[] = [ callback ]
this._topicHandlers.push({topic:topic, callbacks:cbs})
log.info(`first callback added for topic ${topic}`)
}
} }
exec() : void { exec() : void {
@ -43,7 +54,7 @@ export class MqttClient {
for (let topicHandler of this._topicHandlers) { for (let topicHandler of this._topicHandlers) {
if (this.topicMatch(topicHandler.topic, topic)) { if (this.topicMatch(topicHandler.topic, topic)) {
log.info(`received topic ${topic} matches registered topic ${topicHandler.topic}`) log.info(`received topic ${topic} matches registered topic ${topicHandler.topic}`)
topicHandler.callback(payload) // topicHandler.callback(payload)
} }
} }
}) })