Merge 275ab89e62 into 7dc6191b0a

server/monitor-types/mqtt.js

@@ -10,7 +10,7 @@ class MqttMonitorType extends MonitorType {
      * @inheritdoc
      */
     async check(monitor, heartbeat, server) {
-        const receivedMessage = await this.mqttAsync(monitor.hostname, monitor.mqttTopic, {
+        const [ messageTopic, receivedMessage ] = await this.mqttAsync(monitor.hostname, monitor.mqttTopic, {
             port: monitor.port,
             username: monitor.mqttUsername,
             password: monitor.mqttPassword,
@@ -24,7 +24,7 @@ class MqttMonitorType extends MonitorType {
 
         if (monitor.mqttCheckType === "keyword") {
             if (receivedMessage != null && receivedMessage.includes(monitor.mqttSuccessMessage)) {
-                heartbeat.msg = `Topic: ${monitor.mqttTopic}; Message: ${receivedMessage}`;
+                heartbeat.msg = `Topic: ${messageTopic}; Message: ${receivedMessage}`;
                 heartbeat.status = UP;
             } else {
                 throw Error(`Message Mismatch - Topic: ${monitor.mqttTopic}; Message: ${receivedMessage}`);
@@ -101,11 +101,9 @@ class MqttMonitorType extends MonitorType {
             });
 
             client.on("message", (messageTopic, message) => {
-                if (messageTopic === topic) {
+                client.end();
-                    client.end();
+                clearTimeout(timeoutID);
-                    clearTimeout(timeoutID);
+                resolve([ messageTopic, message.toString("utf8") ]);
-                    resolve(message.toString("utf8"));
-                }
             });
 
         });

test/backend-test/test-mqtt.js

@@ -9,17 +9,19 @@ const { UP, PENDING } = require("../../src/util");
  * Runs an MQTT test with the
  * @param  {string} mqttSuccessMessage the message that the monitor expects
  * @param {null|"keyword"|"json-query"} mqttCheckType the type of check we perform
- * @param {string} receivedMessage what message is recieved from the mqtt channel
+ * @param {string} receivedMessage what message is received from the mqtt channel
+ * @param {string} monitorTopic which MQTT topic is monitored (wildcards are allowed)
+ * @param {string} publishTopic to which MQTT topic the message is sent
  * @returns {Promise<Heartbeat>} the heartbeat produced by the check
  */
-async function testMqtt(mqttSuccessMessage, mqttCheckType, receivedMessage) {
+async function testMqtt(mqttSuccessMessage, mqttCheckType, receivedMessage, monitorTopic = "test", publishTopic = "test") {
     const hiveMQContainer = await new HiveMQContainer().start();
     const connectionString = hiveMQContainer.getConnectionString();
     const mqttMonitorType = new MqttMonitorType();
     const monitor = {
         jsonPath: "firstProp", // always return firstProp for the json-query monitor
         hostname: connectionString.split(":", 2).join(":"),
-        mqttTopic: "test",
+        mqttTopic: monitorTopic,
         port: connectionString.split(":")[2],
         mqttUsername: null,
         mqttPassword: null,
@@ -35,9 +37,9 @@ async function testMqtt(mqttSuccessMessage, mqttCheckType, receivedMessage) {
 
     const testMqttClient = mqtt.connect(hiveMQContainer.getConnectionString());
     testMqttClient.on("connect", () => {
-        testMqttClient.subscribe("test", (error) => {
+        testMqttClient.subscribe(monitorTopic, (error) => {
             if (!error) {
-                testMqttClient.publish("test", receivedMessage);
+                testMqttClient.publish(publishTopic, receivedMessage);
             }
         });
     });
@@ -52,7 +54,7 @@ async function testMqtt(mqttSuccessMessage, mqttCheckType, receivedMessage) {
 }
 
 describe("MqttMonitorType", {
-    concurrency: true,
+    concurrency: 4,
     skip: !!process.env.CI && (process.platform !== "linux" || process.arch !== "x64")
 }, () => {
     test("valid keywords (type=default)", async () => {
@@ -61,11 +63,51 @@ describe("MqttMonitorType", {
         assert.strictEqual(heartbeat.msg, "Topic: test; Message: -> KEYWORD <-");
     });
 
+    test("valid nested topic", async () => {
+        const heartbeat = await testMqtt("KEYWORD", null, "-> KEYWORD <-", "a/b/c", "a/b/c");
+        assert.strictEqual(heartbeat.status, UP);
+        assert.strictEqual(heartbeat.msg, "Topic: a/b/c; Message: -> KEYWORD <-");
+    });
+
+    test("valid wildcard topic (with #)", async () => {
+        const heartbeat = await testMqtt("KEYWORD", null, "-> KEYWORD <-", "a/#", "a/b/c");
+        assert.strictEqual(heartbeat.status, UP);
+        assert.strictEqual(heartbeat.msg, "Topic: a/b/c; Message: -> KEYWORD <-");
+    });
+
+    test("valid wildcard topic (with +)", async () => {
+        const heartbeat = await testMqtt("KEYWORD", null, "-> KEYWORD <-", "a/+/c", "a/b/c");
+        assert.strictEqual(heartbeat.status, UP);
+        assert.strictEqual(heartbeat.msg, "Topic: a/b/c; Message: -> KEYWORD <-");
+    });
+
+    test("invalid topic", async () => {
+        await assert.rejects(
+            testMqtt("keyword will not be checked anyway", null, "message", "x/y/z", "a/b/c"),
+            new Error("Timeout, Message not received"),
+        );
+    });
+
+    test("invalid wildcard topic (with #)", async () => {
+        await assert.rejects(
+            testMqtt("", null, "# should be last character", "#/c", "a/b/c"),
+            new Error("Timeout, Message not received"),
+        );
+    });
+
+    test("invalid wildcard topic (with +)", async () => {
+        await assert.rejects(
+            testMqtt("", null, "message", "x/+/z", "a/b/c"),
+            new Error("Timeout, Message not received"),
+        );
+    });
+
     test("valid keywords (type=keyword)", async () => {
         const heartbeat = await testMqtt("KEYWORD", "keyword", "-> KEYWORD <-");
         assert.strictEqual(heartbeat.status, UP);
         assert.strictEqual(heartbeat.msg, "Topic: test; Message: -> KEYWORD <-");
     });
+
     test("invalid keywords (type=default)", async () => {
         await assert.rejects(
             testMqtt("NOT_PRESENT", null, "-> KEYWORD <-"),
@@ -79,12 +121,14 @@ describe("MqttMonitorType", {
             new Error("Message Mismatch - Topic: test; Message: -> KEYWORD <-"),
         );
     });
+
     test("valid json-query", async () => {
         // works because the monitors' jsonPath is hard-coded to "firstProp"
         const heartbeat = await testMqtt("present", "json-query", "{\"firstProp\":\"present\"}");
         assert.strictEqual(heartbeat.status, UP);
         assert.strictEqual(heartbeat.msg, "Message received, expected value is found");
     });
+
     test("invalid (because query fails) json-query", async () => {
         // works because the monitors' jsonPath is hard-coded to "firstProp"
         await assert.rejects(
@@ -92,6 +136,7 @@ describe("MqttMonitorType", {
             new Error("Message received but value is not equal to expected value, value was: [undefined]"),
         );
     });
+
     test("invalid (because successMessage fails) json-query", async () => {
         // works because the monitors' jsonPath is hard-coded to "firstProp"
         await assert.rejects(