From c3a62f74613760b11330f25f2d570c5e0b443d18 Mon Sep 17 00:00:00 2001 From: Eric Duminil Date: Sun, 26 Oct 2025 20:36:47 +0100 Subject: [PATCH] Allow MQTT topic to have wildcards (# or +) (#5398) --- server/monitor-types/mqtt.js | 12 +++--- test/backend-test/test-mqtt.js | 67 +++++++++++++++++++++++++++++++--- 2 files changed, 67 insertions(+), 12 deletions(-) diff --git a/server/monitor-types/mqtt.js b/server/monitor-types/mqtt.js index 1865bbb42..18595b3a4 100644 --- a/server/monitor-types/mqtt.js +++ b/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, @@ -25,7 +25,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}`); @@ -110,11 +110,9 @@ class MqttMonitorType extends MonitorType { }); client.on("message", (messageTopic, message) => { - if (messageTopic === topic) { - client.end(); - clearTimeout(timeoutID); - resolve(message.toString("utf8")); - } + client.end(); + clearTimeout(timeoutID); + resolve([ messageTopic, message.toString("utf8") ]); }); }); diff --git a/test/backend-test/test-mqtt.js b/test/backend-test/test-mqtt.js index d616b12ed..921df48fc 100644 --- a/test/backend-test/test-mqtt.js +++ b/test/backend-test/test-mqtt.js @@ -10,16 +10,18 @@ const { UP, PENDING } = require("../../src/util"); * @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 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} 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, @@ -36,9 +38,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); } }); }); @@ -53,7 +55,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 () => { @@ -62,11 +64,63 @@ 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 nested topic (with special chars)", async () => { + const heartbeat = await testMqtt("KEYWORD", null, "-> KEYWORD <-", "a/'/$/./*/%", "a/'/$/./*/%"); + assert.strictEqual(heartbeat.status, UP); + assert.strictEqual(heartbeat.msg, "Topic: a/'/$/./*/%; 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("valid wildcard topic (with + and #)", async () => { + const heartbeat = await testMqtt("KEYWORD", null, "-> KEYWORD <-", "a/+/c/#", "a/b/c/d/e"); + assert.strictEqual(heartbeat.status, UP); + assert.strictEqual(heartbeat.msg, "Topic: a/b/c/d/e; 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 <-"), @@ -80,12 +134,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( @@ -93,6 +149,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(