diff --git a/backend/Makefile b/backend/Makefile index 5daf0c4..5ce42b2 100644 --- a/backend/Makefile +++ b/backend/Makefile @@ -5,9 +5,7 @@ CXX=g++ LD=g++ CXXFLAGS = -std=c++23 -pedantic-errors -Wall -Wextra -Wconversion -LDFLAGS = - -# libraries = "" +LDFLAGS = -lmosquitto # CXXFLAGS += $(shell pkg-config $(libraries) --cflags) # LDFLAGS += $(shell pkg-config $(libraries) --libs) diff --git a/backend/src/main.cpp b/backend/src/main.cpp index fa7565f..41ed88b 100644 --- a/backend/src/main.cpp +++ b/backend/src/main.cpp @@ -1,14 +1,38 @@ #include "event_loop.hpp" +#include "mqtt.hpp" #include "server.hpp" +#include +#include #include +#include +#include #include #include #include +#include #define PORT 8888 int main(void) { + auto client = mst::mqtt::Client("localhost", 1883, "test", "1234"); + + client.subscribe("/", [&](std::string_view text) { + std::cout << std::format("Received '{}'\n", text); + }); + + auto mqtt_thread = std::thread([&]() { + try { + client.run(); + } catch (mst::mqtt::Error& ex) { + std::cerr << std::format("MQTT Client failed: {}", ex.what()); + std::abort(); + } + }); + + std::this_thread::sleep_for(std::chrono::milliseconds(1000)); + client.publish("/", "published from c++"); + auto mgr = mst::event::Manager::create().value(); auto x = mst::Server::bind(mgr, "0.0.0.0", PORT); if (!x) { diff --git a/backend/src/mqtt.cpp b/backend/src/mqtt.cpp new file mode 100644 index 0000000..967b592 --- /dev/null +++ b/backend/src/mqtt.cpp @@ -0,0 +1,253 @@ +#include "mqtt.hpp" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace mst::mqtt { + +static auto static_mx = std::mutex(); +static auto use_count = 0; +static bool mosquitto_initialized = false; + +static void lib_init() +{ + auto lock = std::lock_guard(static_mx); + + if (mosquitto_initialized) + return; + + int major; + int minor; + int revision; + mosquitto_lib_version(&major, &minor, &revision); + std::cout << std::format( + "[MQTT] Initializing mosquitto {}.{}.{}\n", major, minor, revision); + + if (mosquitto_lib_init() != MOSQ_ERR_SUCCESS) + throw Error("failed to initialize"); + + mosquitto_initialized = true; + use_count += 1; +} + +static void lib_deinit() +{ + auto lock = std::lock_guard(static_mx); + + use_count -= 1; + + if (!mosquitto_initialized || use_count < 0) + return; + + std::cout << std::format("[MQTT] Deinitializing mosquitto\n"); + + mosquitto_lib_cleanup(); + mosquitto_initialized = false; +} + +namespace callbacks { + + static void connect(struct mosquitto* mosq, void* obj, int rc) + { + (void)mosq; + + auto client = static_cast(obj); + client->cb_connect(rc); + } + static void disconnect(struct mosquitto* mosq, void* obj, int rc) + { + (void)mosq; + (void)rc; + + auto client = static_cast(obj); + client->cb_disconnect(); + } + static void publish(struct mosquitto* mosq, void* obj, int mid) + { + (void)mosq; + (void)mid; + + auto client = static_cast(obj); + client->cb_publish(); + } + static void message(struct mosquitto* mosq, + void* obj, + const struct mosquitto_message* message) + { + (void)mosq; + + auto client = static_cast(obj); + client->cb_message( + message->topic, message->payload, (size_t)message->payloadlen); + } + static void subscribe(struct mosquitto* mosq, + void* obj, + int mid, + int qos_count, + const int* granted_qos) + { + (void)mosq; + (void)mid; + (void)qos_count; + (void)granted_qos; + + auto client = static_cast(obj); + client->cb_subscribe(); + } + static void unsubscribe(struct mosquitto* mosq, void* obj, int mid) + { + (void)mosq; + (void)mid; + + auto client = static_cast(obj); + client->cb_unsubscribe(); + } + +} + +Client::Client(const std::string& hostname, + uint16_t port, + const std::string& username, + const std::string& password) + : m_inst(nullptr) +{ + lib_init(); + m_inst = mosquitto_new(nullptr, true, this); + if (!m_inst) + throw Error( + std::format("failed to create instance({})", strerror(errno))); + + auto inst = static_cast(m_inst); + + mosquitto_threaded_set(inst, true); + + if (auto status + = mosquitto_username_pw_set(inst, username.c_str(), password.c_str()); + status != MOSQ_ERR_SUCCESS) { + + throw Error(std::format("could not set username and password ({})", + mosquitto_strerror(status))); + } + + mosquitto_connect_callback_set(inst, callbacks::connect); + mosquitto_disconnect_callback_set(inst, callbacks::disconnect); + mosquitto_publish_callback_set(inst, callbacks::publish); + mosquitto_message_callback_set(inst, callbacks::message); + mosquitto_subscribe_callback_set(inst, callbacks::subscribe); + mosquitto_unsubscribe_callback_set(inst, callbacks::unsubscribe); + + if (auto status = mosquitto_connect(inst, hostname.c_str(), port, 5); + status != MOSQ_ERR_SUCCESS) { + if (status == MOSQ_ERR_ERRNO) { + throw Error(std::format("could not connect ({})", strerror(errno))); + } else { + throw Error(std::format( + "could not connect ({})", mosquitto_strerror(status))); + } + } +} + +Client::~Client() +{ + auto inst = static_cast(m_inst); + mosquitto_disconnect(inst); + mosquitto_destroy(inst); + lib_deinit(); +} + +void Client::subscribe( + std::string topic, std::function func) +{ + auto inst = static_cast(m_inst); + + m_subscriptions.emplace_back(topic, std::move(func)); + + if (auto status = mosquitto_subscribe(inst, NULL, topic.c_str(), 0); + status != MOSQ_ERR_SUCCESS) { + + throw Error(std::format( + "could not subscribe ({})", mosquitto_strerror(status))); + } +} + +void Client::run() +{ + + auto inst = static_cast(m_inst); + + if (auto status = mosquitto_loop_forever(inst, -1, 1); + status != MOSQ_ERR_SUCCESS) { + + if (status == MOSQ_ERR_ERRNO) { + throw Error(std::format("lost connection ({})", strerror(errno))); + } else { + throw Error(std::format( + "lost connection ({})", mosquitto_strerror(status))); + } + } +} + +void Client::cb_connect(int rc) +{ + if (rc != 0) { + throw Error(std::format( + "client could not connect ({})", mosquitto_reason_string(rc))); + } + + std::cout << std::format("[MQTT] Client connected\n"); +} + +void Client::cb_disconnect() +{ + std::cout << std::format("[MQTT] Client disconnected\n"); +} + +void Client::cb_publish() +{ + std::cout << std::format("[MQTT] Message published\n"); +} + +void Client::cb_message(std::string_view topic, const void* data, size_t size) +{ + std::cout << std::format("[MQTT] Message received\n"); + + auto text = std::string_view(static_cast(data), size); + + for (auto sub : m_subscriptions) { + if (sub.topic == topic) { + sub.func(text); + } + } +} + +void Client::cb_subscribe() +{ + std::cout << std::format("[MQTT] Client subscribed\n"); +} + +void Client::cb_unsubscribe() +{ + std::cout << std::format("[MQTT] Client unsubscribed\n"); +} + +void Client::publish_raw( + const std::string& topic, const void* data, size_t size) +{ + auto inst = static_cast(m_inst); + + if (auto status = mosquitto_publish( + inst, nullptr, topic.c_str(), (int)size, data, 0, false); + status != MOSQ_ERR_SUCCESS) { + + throw Error( + std::format("could not publish ({})", mosquitto_strerror(status))); + } +} +} diff --git a/backend/src/mqtt.hpp b/backend/src/mqtt.hpp new file mode 100644 index 0000000..53a6f7f --- /dev/null +++ b/backend/src/mqtt.hpp @@ -0,0 +1,54 @@ +#pragma once + +#include +#include +#include +#include +#include +#include + +namespace mst::mqtt { + +struct Error : public std::runtime_error { + using std::runtime_error::runtime_error; +}; + +struct Subscription { + std::string topic; + std::function func; +}; + +class Client { +public: + explicit Client(const std::string& hostname, + uint16_t port, + const std::string& username, + const std::string& password); + + ~Client(); + + void publish(const std::string& topic, std::string_view text) + { + publish_raw(topic, text.data(), text.size()); + } + + void subscribe( + std::string topic, std::function func); + + void run(); + + void cb_connect(int rc); + void cb_disconnect(); + void cb_publish(); + void cb_message(std::string_view topic, const void* data, size_t size); + void cb_subscribe(); + void cb_unsubscribe(); + +private: + void publish_raw(const std::string& topic, const void* data, size_t size); + + void* m_inst; + std::vector m_subscriptions; +}; + +};