added MQTT monitor type

server/database.js

@@ -52,6 +52,7 @@ class Database {
         "patch-http-monitor-method-body-and-headers.sql": true,
         "patch-2fa-invalidate-used-token.sql": true,
         "patch-notification_sent_history.sql": true,
+        "patch-added-mqtt-monitor.sql": true,
     }
 
     /**

server/model/monitor.js

@@ -1,5 +1,6 @@
 const https = require("https");
 const dayjs = require("dayjs");
+const mqtt = require("mqtt");
 const utc = require("dayjs/plugin/utc");
 let timezone = require("dayjs/plugin/timezone");
 dayjs.extend(utc);
@@ -7,7 +8,7 @@ dayjs.extend(timezone);
 const axios = require("axios");
 const { Prometheus } = require("../prometheus");
 const { debug, UP, DOWN, PENDING, flipStatus, TimeLogger } = require("../../src/util");
-const { tcping, ping, dnsResolve, checkCertificate, checkStatusCode, getTotalClientInRoom, setting, errorLog } = require("../util-server");
+const { tcping, ping, dnsResolve, checkCertificate, checkStatusCode, getTotalClientInRoom, setting, errorLog, mqttAsync } = require("../util-server");
 const { R } = require("redbean-node");
 const { BeanModel } = require("redbean-node/dist/bean-model");
 const { Notification } = require("../notification");
@@ -112,7 +113,7 @@ class Monitor extends BeanModel {
             // undefined if not https
             let tlsInfo = undefined;
 
-            if (! previousBeat) {
+            if (!previousBeat) {
                 previousBeat = await R.findOne("heartbeat", " monitor_id = ? ORDER BY time DESC", [
                     this.id,
                 ]);
@@ -130,7 +131,7 @@ class Monitor extends BeanModel {
             }
 
             // Duration
-            if (! isFirstBeat) {
+            if (!isFirstBeat) {
                 bean.duration = dayjs(bean.time).diff(dayjs(previousBeat.time), "second");
             } else {
                 bean.duration = 0;
@@ -153,7 +154,7 @@ class Monitor extends BeanModel {
                         },
                         httpsAgent: new https.Agent({
                             maxCachedSessions: 0,      // Use Custom agent to disable session reuse (https://github.com/nodejs/node/issues/3940)
-                            rejectUnauthorized: ! this.getIgnoreTls(),
+                            rejectUnauthorized: !this.getIgnoreTls(),
                         }),
                         maxRedirects: this.maxredirects,
                         validateStatus: (status) => {
@@ -296,7 +297,7 @@ class Monitor extends BeanModel {
                         },
                         httpsAgent: new https.Agent({
                             maxCachedSessions: 0,      // Use Custom agent to disable session reuse (https://github.com/nodejs/node/issues/3940)
-                            rejectUnauthorized: ! this.getIgnoreTls(),
+                            rejectUnauthorized: !this.getIgnoreTls(),
                         }),
                         maxRedirects: this.maxredirects,
                         validateStatus: (status) => {
@@ -319,6 +320,14 @@ class Monitor extends BeanModel {
                         throw new Error("Server not found on Steam");
                     }
 
+                } else if (this.type === "mqtt") {
+                    try {
+                        bean.msg = await mqttAsync(this.url, this.topic, this.successMessage);
+                        bean.status = UP;
+                    } catch (error) {
+                        bean.status = DOWN;
+                        bean.msg = error.message;
+                    }
                 } else {
                     bean.msg = "Unknown Monitor Type";
                     bean.status = PENDING;
@@ -385,7 +394,7 @@ class Monitor extends BeanModel {
 
             previousBeat = bean;
 
-            if (! this.isStop) {
+            if (!this.isStop) {
 
                 if (demoMode) {
                     if (beatInterval < 20) {
@@ -407,7 +416,7 @@ class Monitor extends BeanModel {
                 errorLog(e, false);
                 console.error("Please report to https://github.com/louislam/uptime-kuma/issues");
 
-                if (! this.isStop) {
+                if (!this.isStop) {
                     console.log("Try to restart the monitor");
                     this.heartbeatInterval = setTimeout(safeBeat, this.interval * 1000);
                 }
@@ -590,7 +599,7 @@ class Monitor extends BeanModel {
 
         } else {
             // Handle new monitor with only one beat, because the beat's duration = 0
-            let status = parseInt(await R.getCell("SELECT `status` FROM heartbeat WHERE monitor_id = ?", [ monitorID ]));
+            let status = parseInt(await R.getCell("SELECT `status` FROM heartbeat WHERE monitor_id = ?", [monitorID]));
 
             if (status === UP) {
                 uptime = 1;

server/util-server.js

@@ -10,6 +10,8 @@ const iconv = require("iconv-lite");
 const chardet = require("chardet");
 const fs = require("fs");
 const nodeJsUtil = require("util");
+const mqtt = require("mqtt");
+
 
 // From ping-lite
 exports.WIN = /^win/.test(process.platform);
@@ -26,7 +28,7 @@ exports.initJWTSecret = async () => {
         "jwtSecret",
     ]);
 
-    if (! jwtSecretBean) {
+    if (!jwtSecretBean) {
         jwtSecretBean = R.dispense("setting");
         jwtSecretBean.key = "jwtSecret";
     }
@@ -88,6 +90,30 @@ exports.pingAsync = function (hostname, ipv6 = false) {
     });
 };
 
+exports.mqttAsync = function (hostname, topic, okMessage) {
+    return new Promise((resolve, reject) => {
+        try {
+            let client = mqtt.connect(hostname);
+            client.on("connect", () => {
+                client.subscribe(topic);
+            }
+            );
+            client.on("message", (messageTopic, message) => {
+                console.log(messageTopic);
+                if (messageTopic == topic && message.toString() !== okMessage) {
+                    client.end();
+                    reject(new Error(`Error; Topic: ${messageTopic}; Message: ${message.toString()}`));
+                } else {
+                    client.end();
+                    resolve(`Topic: ${messageTopic}; Message: ${message.toString()}`);
+                }
+            });
+        } catch (error) {
+            reject(new Error(error));
+        }
+    });
+};
+
 exports.dnsResolve = function (hostname, resolver_server, rrtype) {
     const resolver = new Resolver();
     resolver.setServers([resolver_server]);
@@ -269,13 +295,13 @@ exports.getTotalClientInRoom = (io, roomName) => {
 
     const sockets = io.sockets;
 
-    if (! sockets) {
+    if (!sockets) {
         return 0;
     }
 
     const adapter = sockets.adapter;
 
-    if (! adapter) {
+    if (!adapter) {
         return 0;
     }
 
@@ -300,7 +326,7 @@ exports.allowAllOrigin = (res) => {
 };
 
 exports.checkLogin = (socket) => {
-    if (! socket.userID) {
+    if (!socket.userID) {
         throw new Error("You are not logged in.");
     }
 };