self-hosted/uptime-kuma
1
0
Fork 0
mirror of https://github.com/louislam/uptime-kuma.git synced 2024-11-24 08:32:23 +02:00
Code Issues Actions 2 Releases Activity

[MQTT] Try to improve error handling

Browse Source
This commit is contained in:
Louis Lam 2022-01-13 12:42:34 +08:00
parent 6272514820
commit 227bbdea2f
2 changed files with 40 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
server/model/monitor.js
View File

@ -351,7 +351,7 @@ class Monitor extends BeanModel {
} }
} else if (this.type === "mqtt") { } else if (this.type === "mqtt") {
try { try {
bean.msg = await mqttAsync(this.url, this.mqttTopic, this.mqttSuccessMessage, { bean.msg = await mqttAsync(this.hostname, this.mqttTopic, this.mqttSuccessMessage, {
mqttPort: this.port, mqttPort: this.port,
mqttUsername: this.mqttUsername, mqttUsername: this.mqttUsername,
mqttPassword: this.mqttPassword, mqttPassword: this.mqttPassword,

69
server/util-server.js
View File

@ -12,7 +12,6 @@ const fs = require("fs");
const nodeJsUtil = require("util"); const nodeJsUtil = require("util");
const mqtt = require("mqtt"); const mqtt = require("mqtt");
// From ping-lite // From ping-lite
exports.WIN = /^win/.test(process.platform); exports.WIN = /^win/.test(process.platform);
exports.LIN = /^linux/.test(process.platform); exports.LIN = /^linux/.test(process.platform);
@ -94,36 +93,46 @@ exports.pingAsync = function (hostname, ipv6 = false) {
exports.mqttAsync = function (hostname, topic, okMessage, options = {}) { exports.mqttAsync = function (hostname, topic, okMessage, options = {}) {
return new Promise((resolve, reject) => { return new Promise((resolve, reject) => {
const { port, username, password, interval = 20 } = options; const { port, username, password, interval = 20 } = options;
try {
// Adds MQTT protocol to the hostname if not already present // Adds MQTT protocol to the hostname if not already present
if (!/^(?:http|mqtt)s?:\/\//.test(hostname)) { if (!/^(?:http|mqtt)s?:\/\//.test(hostname)) {
hostname = "mqtt://" + hostname; hostname = "mqtt://" + hostname;
}
let client = mqtt.connect(hostname, {
port,
username,
password
});
client.on("connect", () => {
client.subscribe(topic);
});
client.on("message", (messageTopic, message) => {
if (messageTopic == topic) {
if (message.toString() === okMessage) {
client.end();
resolve(`Topic: ${messageTopic}; Message: ${message.toString()}`);
} else {
client.end();
reject(new Error(`Error; Topic: ${messageTopic}; Message: ${message.toString()}`));
}
}
});
setTimeout(() => {
client.end();
}, interval * 1000);
} catch (error) {
reject(new Error(error));
} }
debug("MQTT connecting");
let client = mqtt.connect(hostname, {
port,
username,
password
});
client.on("connect", () => {
debug("MQTT subscribe topic");
client.subscribe(topic);
});
client.on("error", (error) => {
client.end();
reject(error);
});
client.on("message", (messageTopic, message) => {
if (messageTopic == topic) {
if (message.toString() === okMessage) {
client.end();
resolve(`Topic: ${messageTopic}; Message: ${message.toString()}`);
} else {
client.end();
reject(new Error(`Error; Topic: ${messageTopic}; Message: ${message.toString()}`));
}
}
});
setTimeout(() => {
client.end();
}, interval * 1000);
}); });
}; };