Repository: louislam/uptime-kuma
File: server/util-server.js (JavaScript)

Code before:
        });

        client.on("message", (messageTopic, message) => {
            if (messageTopic === topic) {
                client.end();
                clearTimeout(timeoutID);
                if (okMessage != null && okMessage !== "" && message.toString() !== okMessage) {
                    reject(new Error(`Message Mismatch - Topic: ${messageTopic}; Message: ${message.toString()}`));
                } else {
                    resolve(`Topic: ${messageTopic}; Message: ${message.toString()}`);
                }
            }
        });

    });
};

/**
 * Monitor Kafka using Producer
 * @param {string} topic Topic name to produce into
 * @param {string} message Message to produce
 * @param {Object} [options={interval = 20, allowAutoTopicCreation = false, ssl = false, clientId = "Uptime-Kuma"}]
 * Kafka client options. Contains ssl, clientId, allowAutoTopicCreation and
 * interval (interval defaults to 20, allowAutoTopicCreation defaults to false, clientId defaults to "Uptime-Kuma"
 * and ssl defaults to false)
 * @param {string[]} brokers List of kafka brokers to connect, host and port joined by ':'
 * @param {SASLOptions} [saslOptions={mechanism: "plain"}] Options for kafka client Authentication (SASL) (defaults to
 * {mechanism: "plain"})
 * @returns {Promise<string>}
 */
exports.kafkaProducerAsync = function (brokers, topic, message, options = {}, saslOptions = {
    mechanism: "plain",
}) {
    return new Promise((resolve, reject) => {
        const { interval = 20, allowAutoTopicCreation = false, ssl = false, clientId = "Uptime-Kuma" } = options;

        // Used to track manual disconnect calls
        let success = false;

        const timeoutID = setTimeout(() => {
            log.debug("kafkaProducer", "KafkaProducer timeout triggered");
            success = true;
            producer.disconnect();
            reject(new Error("Timeout"));
        }, interval * 1000 * 0.8);

        let client = new Kafka({
            brokers: brokers,
            clientId: clientId,
            sasl: saslOptions,
            ssl: ssl,

Review comment (suggestion):
Is it correct that this is a `string[]` the databse saves the brokers only as a string or am I missing something?
(the reason why I was looking at this in particular is if this supports IPv6, given this mentiones a `:` 😉)

Code after:
        });

        client.on("message", (messageTopic, message) => {
            if (messageTopic === topic) {
                client.end();
                clearTimeout(timeoutID);
                if (okMessage != null && okMessage !== "" && message.toString() !== okMessage) {
                    reject(new Error(`Message Mismatch - Topic: ${messageTopic}; Message: ${message.toString()}`));
                } else {
                    resolve(`Topic: ${messageTopic}; Message: ${message.toString()}`);
                }
            }
        });

    });
};

/**
 * Monitor Kafka using Producer
 * @param {string} topic Topic name to produce into
 * @param {string} message Message to produce
 * @param {Object} [options={interval = 20, allowAutoTopicCreation = false, ssl = false, clientId = "Uptime-Kuma"}]
 * Kafka client options. Contains ssl, clientId, allowAutoTopicCreation and
 * interval (interval defaults to 20, allowAutoTopicCreation defaults to false, clientId defaults to "Uptime-Kuma"
 * and ssl defaults to false)
 * @param {string[]} brokers List of kafka brokers to connect, host and port joined by ':'
 * @param {SASLOptions} [saslOptions={}] Options for kafka client Authentication (SASL) (defaults to
 * {})
 * @returns {Promise<string>}
 */
exports.kafkaProducerAsync = function (brokers, topic, message, options = {}, saslOptions = {}) {
    return new Promise((resolve, reject) => {
        const { interval = 20, allowAutoTopicCreation = false, ssl = false, clientId = "Uptime-Kuma" } = options;

        let connectedToKafka = false;

        const timeoutID = setTimeout(() => {
            log.debug("kafkaProducer", "KafkaProducer timeout triggered");
            connectedToKafka = true;
            reject(new Error("Timeout"));
        }, interval * 1000 * 0.8);

        if (saslOptions.mechanism === "None") {
            saslOptions = undefined;
        }

        let client = new Kafka({
            brokers: brokers,
            clientId: clientId,
            sasl: saslOptions,
            retry: {