mqttbroker.cpp File Reference

#include "SocketContextFactory.h"
#include "config.h"
#include "lib/Mqtt.h"
#include "lib/MqttModel.h"
#include "lib/inja.hpp"

Include dependency graph for mqttbroker.cpp:

Go to the source code of this file.

Functions
static void upgrade	APPLICATION (req, res)
static std::string	href (const std::string &text, const std::string &link)
static std::string	href (const std::string &text, const std::string &url, const std::string &windowId, uint16_t width, uint16_t height)
static std::string	getOverviewPage (inja::Environment environment, std::shared_ptr< iot::mqtt::server::broker::Broker > broker, mqtt::mqttbroker::lib::MqttModel &mqttModel)
static std::string	getDetailedPage (inja::Environment environment, const mqtt::mqttbroker::lib::Mqtt *mqtt)
static std::string	getRedirectSpinnerPage (inja::Environment environment, const std::string &location)
static std::string	urlDecode (const std::string &encoded)
static express::Router	getRouter (const inja::Environment &environment, std::shared_ptr< iot::mqtt::server::broker::Broker > broker)
static void	reportState (const std::string &instanceName, const core::socket::SocketAddress &socketAddress, const core::socket::State &state)
int	main (int argc, char *argv[])

Function Documentation

APPLICATION()

void upgrade APPLICATION ( req , res  )

static

Definition at line 135 of file mqttbroker.cpp.

                                          {
    const std::string connectionName = res->getSocketContext()->getSocketConnection()->getConnectionName();
 
    VLOG(2) << connectionName << " HTTP: Upgrade request:\n"
            << httputils::toString(req->method,
                                   req->url,
                                   "HTTP/" + std::to_string(req->httpMajor) + "." + std::to_string(req->httpMinor),
                                   req->queries,
                                   req->headers,
                                   req->trailer,
                                   req->cookies,
                                   std::vector<char>());
 
    if (req->get("sec-websocket-protocol").find("mqtt") != std::string::npos) {
        res->upgrade(req, [req, res, connectionName](const std::string& name) {
            if (!name.empty()) {
                VLOG(1) << connectionName << ": Successful upgrade:";
                VLOG(1) << connectionName << ":    Selected: " << name;
                VLOG(1) << connectionName << ":   Requested: " << req->get("sec-websocket-protocol");
 
                res->end();
            } else {
                VLOG(1) << connectionName << ": Can not upgrade to any of '" << req->get("upgrade") << "'";
 
                res->sendStatus(404);
            }
        });
    } else {
        VLOG(1) << connectionName << ": Unsupported subprotocol(s):";
        VLOG(1) << "    Expected: mqtt";
        VLOG(1) << "   Requested: " << req->get("sec-websocket-protocol");
 
        res->sendStatus(404);
    }
}

getDetailedPage()

std::string getDetailedPage ( inja::Environment environment, const mqtt::mqttbroker::lib::Mqtt * mqtt )

static

Definition at line 255 of file mqttbroker.cpp.

                                                                                                 {
    return environment.render_file("DetailPage.html",
                                   {{"table-id", "detail"},
                                    {"client_id", mqtt->getClientId()},
                                    {"title", mqtt->getClientId()},
                                    {"header_row", {"Attribute", "Value"}},
                                    {"data_rows",
                                     inja::json::array({{"Client ID", mqtt->getClientId()},
                                                        {"Connection", mqtt->getConnectionName()},
                                                        {"Clean Session", mqtt->getCleanSession() ? "true" : "false"},
                                                        {"Connect Flags", std::to_string(mqtt->getConnectFlags())},
                                                        {"Username", mqtt->getUsername()},
                                                        {"Username Flag", mqtt->getUsernameFlag() ? "true" : "false"},
                                                        {"Password", mqtt->getPassword()},
                                                        {"Password Flag", mqtt->getPasswordFlag() ? "true" : "false"},
                                                        {"Keep Alive", std::to_string(mqtt->getKeepAlive())},
                                                        {"Protocol", mqtt->getProtocol()},
                                                        {"Protocol Level", std::to_string(mqtt->getLevel())},
                                                        {"Loop Prevention", !mqtt->getReflect() ? "true" : "false"},
                                                        {"Will Message", mqtt->getWillMessage()},
                                                        {"Will Topic", mqtt->getWillTopic()},
                                                        {"Will QoS", std::to_string(mqtt->getWillQoS())},
                                                        {"Will Flag", mqtt->getWillFlag() ? "true" : "false"},
                                                        {"Will Retain", mqtt->getWillRetain() ? "true" : "false"}})},
                                    {"client_id", mqtt->getClientId()},
                                    {"topics", mqtt->getSubscriptions()}});
}

Referenced by getRouter().

Here is the caller graph for this function:

getOverviewPage()

std::string getOverviewPage ( inja::Environment environment, std::shared_ptr< iot::mqtt::server::broker::Broker > broker, mqtt::mqttbroker::lib::MqttModel & mqttModel )

static

Definition at line 195 of file mqttbroker.cpp.

                                                                          {
    inja::json json;
 
    json["table-id"] = "overview";
    json["title"] = "MQTTBroker | Active Clients";
    json["voc"] = href("Volker Christian", "https://github.com/VolkerChristian/");
    json["broker"] = href("MQTTBroker", "https://github.com/SNodeC/mqttsuite/tree/master/mqttbroker");
    json["suite"] = href("MQTTSuite", "https://github.com/SNodeC/mqttsuite");
    json["snodec"] = "Powered by " + href("SNode.C", "https://github.com/SNodeC/snode.c");
    json["since"] = mqttModel.onlineSince();
    json["duration"] = mqttModel.onlineDuration();
    json["header_row"] = {"Client ID", "Online Since", "Duration", "Connection", "Local Address", "Remote Address", "Action"};
    json["data_rows"] = inja::json::array();
    json["subscribed_topics"] = inja::json::array();
    json["retained_topics"] = inja::json::array();
 
    const std::map<std::string, std::list<std::pair<std::string, uint8_t>>>& subscribedTopics = broker->getSubscriptionTree();
 
    inja::json& subscribedTopicsJson = json["subscribed_topics"];
    for (const auto& [topic, clients] : subscribedTopics) {
        inja::json topicJson;
 
        topicJson["key"] = topic;
        for (const auto& client : clients) {
            std::ostringstream windowId("window");
            for (char ch : client.first) {
                if (std::isalnum(static_cast<unsigned char>(ch))) {
                    windowId << ch;
                } else {
                    windowId << std::hex << std::uppercase << std::setw(2) << std::setfill('0')
                             << static_cast<int>(static_cast<unsigned char>(ch));
                }
            }
 
            topicJson["values"].push_back({{"client_id", client.first},
                                           {"link", href(client.first, "/client?" + client.first, windowId.str(), 450, 900)},
                                           {"topic", topic},
                                           {"qos", std::to_string(static_cast<int>(client.second))}});
        }
 
        subscribedTopicsJson.push_back(topicJson);
    }
 
    std::list<std::pair<std::string, std::string>> retainTree = broker->getRetainTree();
 
    inja::json& retainedTopicsJson = json["retained_topics"];
    for (const auto& [topic, message] : retainTree) {
        inja::json topicJson;
 
        topicJson["key"] = topic;
        topicJson["values"].push_back(message);
 
        retainedTopicsJson.push_back(topicJson);
    }
 
    return environment.render_file("OverviewPage.html", json);
}

References mqtt::mqttbroker::lib::MqttModel::onlineDuration(), and mqtt::mqttbroker::lib::MqttModel::onlineSince().

Referenced by getRouter().

Here is the call graph for this function:

Here is the caller graph for this function:

getRedirectSpinnerPage()

std::string getRedirectSpinnerPage ( inja::Environment environment, const std::string & location )

static

Definition at line 283 of file mqttbroker.cpp.

                                                                                                {
    return environment.render_file("Spinner.html", {{"location", location}});
}

getRouter()

express::Router getRouter ( const inja::Environment & environment, std::shared_ptr< iot::mqtt::server::broker::Broker > broker )

static

Definition at line 320 of file mqttbroker.cpp.

                                                                                                                    {
    const express::Router& jsonRouter = express::middleware::JsonMiddleware();
 
    jsonRouter.post("/disconnect", [] APPLICATION(req, res) { // cppcheck-suppress unknownMacro
        VLOG(1) << "POST /disconnect";
 
        req->getAttribute<nlohmann::json>(
            [&res](nlohmann::json& json) {
                std::string jsonString = json.dump(4);
 
                VLOG(1) << jsonString;
 
                std::string clientId = json["client_id"].get<std::string>();
                const mqtt::mqttbroker::lib::Mqtt* mqtt = mqtt::mqttbroker::lib::MqttModel::instance().getMqtt(clientId);
 
                if (mqtt != nullptr) {
                    mqtt->getMqttContext()->getSocketConnection()->close();
                    res->send(jsonString);
                } else {
                    res->status(404).send("MQTT client has never existed or already gone away: '" + clientId + "'");
                }
            },
            [&res](const std::string& key) {
                VLOG(1) << "Attribute type not found: " << key;
 
                res->status(400).send("Attribute type not found: " + key);
            });
    });
 
    jsonRouter.post("/unsubscribe", [] APPLICATION(req, res) {
        VLOG(1) << "POST /unsubscribe";
 
        req->getAttribute<nlohmann::json>(
            [&res](nlohmann::json& json) {
                std::string jsonString = json.dump(4);
 
                VLOG(1) << jsonString;
 
                std::string clientId = json["client_id"].get<std::string>();
                std::string topic = json["topic"].get<std::string>();
 
                const mqtt::mqttbroker::lib::Mqtt* mqtt = mqtt::mqttbroker::lib::MqttModel::instance().getMqtt(clientId);
 
                if (mqtt != nullptr) {
                    mqtt->unsubscribe(topic);
                    res->send(jsonString);
                } else {
                    res->status(404).send("MQTT client has never existed or already gone away: '" + clientId + "'");
                }
            },
            [&res](const std::string& key) {
                VLOG(1) << "Attribute type not found: " << key;
 
                res->status(400).send("Attribute type not found: " + key);
            });
    });
 
    jsonRouter.post("/release", [broker] APPLICATION(req, res) {
        VLOG(1) << "POST /release";
 
        req->getAttribute<nlohmann::json>(
            [&res, broker](nlohmann::json& json) {
                std::string jsonString = json.dump(4);
 
                VLOG(1) << jsonString;
 
                std::string topic = json["topic"].get<std::string>();
 
                broker->release(topic);
 
                res->send(jsonString);
            },
            [&res](const std::string& key) {
                VLOG(1) << "Attribute type not found: " << key;
 
                res->status(400).send("Attribute type not found: " + key);
            });
    });
 
    jsonRouter.post("/subscribe", [] APPLICATION(req, res) {
        VLOG(1) << "POST /subscribe";
 
        req->getAttribute<nlohmann::json>(
            [&res](nlohmann::json& json) {
                std::string jsonString = json.dump(4);
 
                VLOG(1) << jsonString;
 
                std::string clientId = json["client_id"].get<std::string>();
                std::string topic = json["topic"].get<std::string>();
                uint8_t qoS = json["qos"].get<uint8_t>();
 
                const mqtt::mqttbroker::lib::Mqtt* mqtt = mqtt::mqttbroker::lib::MqttModel::instance().getMqtt(clientId);
 
                if (mqtt != nullptr) {
                    mqtt->subscribe(topic, qoS);
                    res->send(jsonString);
                } else {
                    res->status(404).send("MQTT client has never existed or already gone away: '" + clientId + "'");
                }
            },
            [&res](const std::string& key) {
                VLOG(1) << "Attribute type not found: " << key;
 
                res->status(400).send("Attribute type not found: " + key);
            });
    });
    const express::Router router;
 
    router.use(jsonRouter);
 
    router.get("/clients", [environment, broker] APPLICATION(req, res) {
        std::string responseString;
        int responseStatus = 200;
 
        try {
            responseString = getOverviewPage(environment, broker, mqtt::mqttbroker::lib::MqttModel::instance());
        } catch (const inja::InjaError& error) {
            responseStatus = 500;
            responseString = "Internal Server Error\n";
            responseString += std::string(error.what()) + " " + error.type + " " + error.message + " " +
                              std::to_string(error.location.line) + ":" + std::to_string(error.location.column);
        }
 
        res->status(responseStatus).send(responseString);
    });
 
    router.get("/client", [environment] APPLICATION(req, res) {
        std::string responseString;
        int responseStatus = 200;
 
        if (req->queries.size() == 1) {
            const mqtt::mqttbroker::lib::Mqtt* mqtt =
                mqtt::mqttbroker::lib::MqttModel::instance().getMqtt(urlDecode(req->queries.begin()->first));
 
            if (mqtt != nullptr) {
                VLOG(1) << "Subscriptions for client " << mqtt->getClientId();
                for (const std::string& subscription : mqtt->getSubscriptions()) {
                    VLOG(1) << "  " << subscription;
                }
 
                try {
                    responseString = getDetailedPage(environment, mqtt);
                } catch (const inja::InjaError& error) {
                    responseStatus = 500;
                    responseString = "Internal Server Error\n";
                    responseString += std::string(error.what()) + " " + error.type + " " + error.message + " " +
                                      std::to_string(error.location.line) + ":" + std::to_string(error.location.column);
                }
            } else {
                responseStatus = 404;
                responseString = "Not Found: " + urlDecode(req->queries.begin()->first);
            }
        } else {
            responseStatus = 400;
            responseString = "Bad Request: No Client requested";
        }
 
        res->status(responseStatus).send(responseString);
    });
 
    router.get("/spinner", [environment] APPLICATION(req, res) {
        std::string responseString;
        int responseStatus = 200;
 
        if (req->queries.size() == 1) {
            responseString = getRedirectSpinnerPage(environment, req->queries.begin()->first);
        } else {
            responseStatus = 400;
            responseString = "Bad Request: No Client requested";
        }
        res->status(responseStatus).send(responseString);
    });
 
    router.get("/ws", [] APPLICATION(req, res) {
        if (req->headers.contains("upgrade")) {
            upgrade(req, res);
        } else {
            res->redirect("/spinner?/clients");
        }
    });
 
    router.get("/mqtt", [] APPLICATION(req, res) {
        if (req->headers.contains("upgrade")) {
            upgrade(req, res);
        } else {
            res->redirect("/spinner?/clients");
        }
    });
 
    router.get("/", [] APPLICATION(req, res) {
        if (req->headers.contains("upgrade")) {
            upgrade(req, res);
        } else {
            res->redirect("/spinner?/clients");
        }
    });
 
    router.get("/sse", [] APPLICATION(req, res) {
        if (web::http::ciContains(req->get("Accept"), "text/event-stream")) {
            res->set("Content-Type", "text/event-stream") //
                .set("Cache-Control", "no-cache")
                .set("Connection", "keep-alive");
            res->sendHeader();
 
            mqtt::mqttbroker::lib::MqttModel::instance().addEventReceiver(res, req->get("Last-Event-ID"));
        } else {
            res->redirect("/spinner?/clients");
        }
    });
 
    return router;
}

References mqtt::mqttbroker::lib::MqttModel::addEventReceiver(), inja::SourceLocation::column, getDetailedPage(), mqtt::mqttbroker::lib::MqttModel::getMqtt(), getOverviewPage(), mqtt::mqttbroker::lib::MqttModel::instance(), inja::SourceLocation::line, inja::InjaError::location, inja::InjaError::message, and inja::InjaError::type.

Referenced by main().

Here is the call graph for this function:

Here is the caller graph for this function:

href() [1/2]

std::string href ( const std::string & text, const std::string & link )

static

Definition at line 171 of file mqttbroker.cpp.

                                                                    {
    return "<a href=\"" + link + "\" style=\"color:inherit;\">" + text + "</a>";
}

href() [2/2]

std::string href ( const std::string & text, const std::string & url, const std::string & windowId, uint16_t width, uint16_t height )

static

Definition at line 175 of file mqttbroker.cpp.

                                                                                                                               {
    return "<a href=\"#\" onClick=\""
           "let key = '" +
           windowId +
           "'; "
           "if (!localStorage.getItem(key)) "
           "  localStorage.setItem(key, key + '-' + Math.random().toString(36).substr(2, 6)); "
           "let uniqueId = localStorage.getItem(key); "
           "if (!window._openWindows) window._openWindows = {}; "
           "if (!window._openWindows[uniqueId] || window._openWindows[uniqueId].closed) { "
           "  window._openWindows[uniqueId] = window.open('" +
           url + "', uniqueId, 'width=" + std::to_string(width) + ", height=" + std::to_string(height) +
           ",location=no, menubar=no, status=no, toolbar=no'); "
           "} else { "
           "  window._openWindows[uniqueId].focus(); "
           "} return false;\" "
           "style=\"color:inherit;\">" +
           text + "</a>";
}

main()

int main	(	int	argc,
		char *	argv[] )

Definition at line 552 of file mqttbroker.cpp.

                                 {
    utils::Config::addStringOption("--mqtt-mapping-file", "MQTT mapping file (json format) for integration", "[path]", "");
    utils::Config::addStringOption("--mqtt-session-store", "Path to file for the persistent session store", "[path]", "");
    utils::Config::addStringOption(
        "--html-dir", "Path to html source directory", "[path]", std::string(CMAKE_INSTALL_PREFIX) + "/var/www/mqttsuite/mqttbroker");
 
    core::SNodeC::init(argc, argv);
 
    std::shared_ptr<iot::mqtt::server::broker::Broker> broker =
        iot::mqtt::server::broker::Broker::instance(SUBSCRIPTION_MAX_QOS, utils::Config::getStringOptionValue("--mqtt-session-store"));
 
#ifdef CONFIG_MQTTSUITE_BROKER_TCP_IPV4
    net::in::stream::legacy::Server<mqtt::mqttbroker::SocketContextFactory>(
        "in-mqtt",
        [](auto& config) {
            config.setPort(1883);
            config.setRetry();
            config.setDisableNagleAlgorithm();
        },
        broker)
        .listen([](const auto& socketAddress, core::socket::State state) {
            reportState("in-mqtt", socketAddress, state);
        });
 
#ifdef CONFIG_MQTTSUITE_BROKER_TLS_IPV4
    net::in::stream::tls::Server<mqtt::mqttbroker::SocketContextFactory>(
        "in-mqtts",
        [](auto& config) {
            config.setPort(8883);
            config.setRetry();
            config.setDisableNagleAlgorithm();
        },
        broker)
        .listen([](const auto& socketAddress, core::socket::State state) {
            reportState("in-mqtts", socketAddress, state);
        });
#endif
#endif
 
#ifdef CONFIG_MQTTSUITE_BROKER_TCP_IPV6
    net::in6::stream::legacy::Server<mqtt::mqttbroker::SocketContextFactory>(
        "in6-mqtt",
        [](auto& config) {
            config.setPort(1883);
            config.setRetry();
            config.setDisableNagleAlgorithm();
 
            config.setIPv6Only();
        },
        broker)
        .listen([](const auto& socketAddress, core::socket::State state) {
            reportState("in6-mqtt", socketAddress, state);
        });
 
#ifdef CONFIG_MQTTSUITE_BROKER_TLS_IPV6
    net::in6::stream::tls::Server<mqtt::mqttbroker::SocketContextFactory>(
        "in6-mqtts",
        [](auto& config) {
            config.setPort(8883);
            config.setRetry();
            config.setDisableNagleAlgorithm();
 
            config.setIPv6Only();
        },
        broker)
        .listen([](const auto& socketAddress, core::socket::State state) {
            reportState("in6-mqtts", socketAddress, state);
        });
#endif
#endif
 
#ifdef CONFIG_MQTTSUITE_BROKER_UNIX
    net::un::stream::legacy::Server<mqtt::mqttbroker::SocketContextFactory>(
        "un-mqtt",
        [](auto& config) {
            config.setSunPath("/tmp/" + utils::Config::getApplicationName() + "-" + config.getInstanceName());
            config.setRetry();
        },
        broker)
        .listen([](const auto& socketAddress, core::socket::State state) {
            reportState("un-mqtt", socketAddress, state);
        });
 
#ifdef CONFIG_MQTTSUITE_BROKER_UNIX_TLS
    net::un::stream::tls::Server<mqtt::mqttbroker::SocketContextFactory>(
        "un-mqtts",
        [](auto& config) {
            config.setSunPath("/tmp/" + utils::Config::getApplicationName() + "-" + config.getInstanceName());
            config.setRetry();
        },
        broker)
        .listen([](const auto& socketAddress, core::socket::State state) {
            reportState("un-mqtts", socketAddress, state);
        });
#endif
#endif
 
    inja::Environment environment{utils::Config::getStringOptionValue("--html-dir") + "/"};
    express::Router router = getRouter(environment, broker);
 
#ifdef CONFIG_MQTTSUITE_BROKER_TCP_IPV4
    express::legacy::in::Server("in-http", router, reportState, [](auto& config) {
        config.setPort(8080);
        config.setRetry();
        config.setDisableNagleAlgorithm();
    });
 
#ifdef CONFIG_MQTTSUITE_BROKER_TLS_IPV4
    express::tls::in::Server("in-https", router, reportState, [](auto& config) {
        config.setPort(8088);
        config.setRetry();
        config.setDisableNagleAlgorithm();
    });
#endif
#endif
 
#ifdef CONFIG_MQTTSUITE_BROKER_TCP_IPV6
    express::legacy::in6::Server("in6-http", router, reportState, [](auto& config) {
        config.setPort(8080);
        config.setRetry();
        config.setDisableNagleAlgorithm();
 
        config.setIPv6Only();
    });
 
#ifdef CONFIG_MQTTSUITE_BROKER_TLS_IPV6
    express::tls::in6::Server("in6-https", router, reportState, [](auto& config) {
        config.setPort(8088);
        config.setRetry();
        config.setDisableNagleAlgorithm();
 
        config.setIPv6Only();
    });
#endif
#endif
 
#ifdef CONFIG_MQTTSUITE_BROKER_UNIX
    express::legacy::un::Server("un-http", router, reportState, [](auto& config) {
        config.setSunPath("/tmp/" + utils::Config::getApplicationName() + "-" + config.getInstanceName());
    });
 
#ifdef CONFIG_MQTTSUITE_BROKER_UNIX_TLS
    express::tls::un::Server("un-https", router, reportState, [](auto& config) {
        config.setSunPath("/tmp/" + utils::Config::getApplicationName() + "-" + config.getInstanceName());
    });
#endif
#endif
 
    return core::SNodeC::start();
}

References getRouter().

Here is the call graph for this function:

reportState()

void reportState ( const std::string & instanceName, const core::socket::SocketAddress & socketAddress, const core::socket::State & state )

static

Definition at line 535 of file mqttbroker.cpp.

                                                                                                                       {
    switch (state) {
        case core::socket::State::OK:
            VLOG(1) << instanceName << ": listening on '" << socketAddress.toString() << "'";
            break;
        case core::socket::State::DISABLED:
            VLOG(1) << instanceName << ": disabled";
            break;
        case core::socket::State::ERROR:
            VLOG(1) << instanceName << ": " << socketAddress.toString() << ": " << state.what();
            break;
        case core::socket::State::FATAL:
            VLOG(1) << instanceName << ": " << socketAddress.toString() << ": " << state.what();
            break;
    }
}

urlDecode()

std::string urlDecode ( const std::string & encoded )

static

Definition at line 287 of file mqttbroker.cpp.

                                                     {
    std::string decoded;
    size_t i = 0;
 
    while (i < encoded.length()) {
        char ch = encoded[i];
        if (ch == '%') {
            // Make sure there are at least two characters after '%'
            if (i + 2 < encoded.length() && std::isxdigit(encoded[i + 1]) && std::isxdigit(encoded[i + 2])) {
                // Convert the two hex digits to a character
                std::string hexValue = encoded.substr(i + 1, 2);
                char decodedChar = static_cast<char>(std::stoi(hexValue, nullptr, 16));
                decoded.push_back(decodedChar);
                i += 3; // Skip over the % and the two hex digits
            } else {
                // Malformed encoding, just add the '%' as is.
                decoded.push_back(ch);
                ++i;
            }
        } else if (ch == '+') {
            // Convert '+' to space (common in URL encoding)
            decoded.push_back(' ');
            ++i;
        } else {
            // Regular character, just append it.
            decoded.push_back(ch);
            ++i;
        }
    }
 
    return decoded;
}