diff --git a/docker/MQTTManager/CMakeLists.txt b/docker/MQTTManager/CMakeLists.txt index 6a78036b..7c68ee92 100644 --- a/docker/MQTTManager/CMakeLists.txt +++ b/docker/MQTTManager/CMakeLists.txt @@ -89,6 +89,11 @@ set_target_properties(MQTTManager_OpenhabManager PROPERTIES PUBLIC_HEADER ${CMAK target_include_directories(MQTTManager_OpenhabManager PUBLIC ${CMAKE_INCLUDE_SRC_DIRECTORY}) target_link_libraries(MQTTManager_OpenhabManager Boost::stacktrace_backtrace spdlog::spdlog nlohmann_json::nlohmann_json ixwebsocket::ixwebsocket Boost::boost MQTTManager_Config MQTTManager_WebsocketServer gtest::gtest) +add_library(MQTTManager_ManagerClock STATIC ${CMAKE_INCLUDE_SRC_DIRECTORY}/manager_clock/manager_clock.cpp ${CMAKE_INCLUDE_SRC_DIRECTORY}/manager_clock/manager_clock.hpp) +set_target_properties(MQTTManager_ManagerClock PROPERTIES PUBLIC_HEADER ${CMAKE_INCLUDE_SRC_DIRECTORY}/manager_clock/manager_clock.hpp) +target_include_directories(MQTTManager_ManagerClock PUBLIC ${CMAKE_INCLUDE_SRC_DIRECTORY}) +target_link_libraries(MQTTManager_ManagerClock MQTT_Manager MQTTManager_Config spdlog::spdlog gtest::gtest) + add_library(MQTTManager_Weather STATIC ${CMAKE_INCLUDE_SRC_DIRECTORY}/weather/weather.cpp ${CMAKE_INCLUDE_SRC_DIRECTORY}/weather/weather.hpp ${CMAKE_INCLUDE_SRC_DIRECTORY}/weather/weather_icons.hpp) set_target_properties(MQTTManager_Weather PROPERTIES PUBLIC_HEADER ${CMAKE_INCLUDE_SRC_DIRECTORY}/weather/weather.hpp) target_include_directories(MQTTManager_Weather PUBLIC ${CMAKE_INCLUDE_SRC_DIRECTORY}) @@ -122,7 +127,7 @@ target_link_libraries(MQTTManager_Switch MQTTManager_HomeAssistantManager spdlog add_library(MQTTManager_NSPanel STATIC ${CMAKE_INCLUDE_SRC_DIRECTORY}/nspanel/nspanel.cpp ${CMAKE_INCLUDE_SRC_DIRECTORY}/nspanel/nspanel.hpp) set_target_properties(MQTTManager_NSPanel PROPERTIES PUBLIC_HEADER ${CMAKE_INCLUDE_SRC_DIRECTORY}/nspanel/nspanel.hpp) target_include_directories(MQTTManager_NSPanel PUBLIC ${CMAKE_INCLUDE_SRC_DIRECTORY}) -target_link_libraries(MQTTManager_NSPanel spdlog::spdlog nlohmann_json::nlohmann_json MQTT_Manager MQTTManager_Room MQTTManager_WebsocketServer tz::tz Boost::boost MQTTManager_Config Protobuf_MQTTManager MQTTManager_DatabaseManager gtest::gtest) +target_link_libraries(MQTTManager_NSPanel spdlog::spdlog nlohmann_json::nlohmann_json MQTT_Manager MQTTManager_ManagerClock MQTTManager_Room MQTTManager_WebsocketServer tz::tz Boost::boost MQTTManager_Config Protobuf_MQTTManager MQTTManager_DatabaseManager gtest::gtest) add_library(MQTTManager_Scene STATIC ${CMAKE_INCLUDE_SRC_DIRECTORY}/scenes/scene.hpp ${CMAKE_INCLUDE_SRC_DIRECTORY}/scenes/scene.cpp ${CMAKE_INCLUDE_SRC_DIRECTORY}/scenes/nspm_scene.hpp ${CMAKE_INCLUDE_SRC_DIRECTORY}/scenes/nspm_scene.cpp ${CMAKE_INCLUDE_SRC_DIRECTORY}/scenes/home_assistant_scene.hpp ${CMAKE_INCLUDE_SRC_DIRECTORY}/scenes/home_assistant_scene.cpp ${CMAKE_INCLUDE_SRC_DIRECTORY}/scenes/openhab_scene.hpp ${CMAKE_INCLUDE_SRC_DIRECTORY}/scenes/openhab_scene.cpp) # set_target_properties(MQTTManager_Scene PROPERTIES PUBLIC_HEADER ${CMAKE_INCLUDE_SRC_DIRECTORY}/scene/scene.hpp ${CMAKE_INCLUDE_SRC_DIRECTORY}/scene/nspm_scene.hpp) @@ -155,7 +160,7 @@ target_link_libraries(MQTTManager_NextionImageServer ixwebsocket::ixwebsocket sp add_executable(${PROJECT_NAME} src/main.cpp) target_include_directories(${PROJECT_NAME} PUBLIC ${CMAKE_INCLUDE_SRC_DIRECTORY}) -target_link_libraries(${PROJECT_NAME} MQTT_Manager MQTTManager_Config MQTTManager_HomeAssistantManager MQTTManager_OpenhabManager MQTTManager_EntityManager MQTTManager_NextionImageServer MQTTManager_Scene MQTTManager_Room MQTTManager_WebsocketServer MQTTManager_CommandManager gtest::gtest) +target_link_libraries(${PROJECT_NAME} MQTT_Manager MQTTManager_ManagerClock MQTTManager_Config MQTTManager_HomeAssistantManager MQTTManager_OpenhabManager MQTTManager_EntityManager MQTTManager_NextionImageServer MQTTManager_Scene MQTTManager_Room MQTTManager_WebsocketServer MQTTManager_CommandManager gtest::gtest) if(NSPM_BUILD_TESTS) enable_testing() diff --git a/docker/MQTTManager/include/manager_clock/manager_clock.cpp b/docker/MQTTManager/include/manager_clock/manager_clock.cpp new file mode 100644 index 00000000..7827c776 --- /dev/null +++ b/docker/MQTTManager/include/manager_clock/manager_clock.cpp @@ -0,0 +1,96 @@ +#include "manager_clock.hpp" +#include "mqtt_manager/mqtt_manager.hpp" +#include "mqtt_manager_config/mqtt_manager_config.hpp" +#include +#include +#include +#include + +void ManagerClock::start() { + ManagerClock::init(); + if (!ManagerClock::_instance.joinable()) { + ManagerClock::_instance = std::thread(ManagerClock::_run_clock_thread); + } +} + +void ManagerClock::init() { + MQTT_Manager::attach_connected_callback(&ManagerClock::publish_epoch); +} + +void ManagerClock::_run_clock_thread() { + // Set timezone + setenv("TZ", MqttManagerConfig::timezone.c_str(), 1); + tzset(); + + // Timezone loaded, proceed to update MQTT + for (;;) { + char time_buffer[20]; + char ampm_buffer[20]; + std::string time_str; + char date_buffer[100]; + std::string date_str; + + std::time_t time = std::time({}); + std::strftime(date_buffer, 100, MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::DATE_FORMAT).c_str(), std::localtime(&time)); + date_str = date_buffer; + + if (!MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::CLOCK_US_STYLE)) [[likely]] { + std::strftime(time_buffer, 20, "%H:%M", std::localtime(&time)); + time_str = time_buffer; + } else { + std::strftime(time_buffer, 20, "%I:%M", std::localtime(&time)); + std::strftime(ampm_buffer, 20, "%p", std::localtime(&time)); + time_str = time_buffer; + } + + MQTT_Manager::publish(fmt::format("nspanel/mqttmanager_{}/status/time", MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::MANAGER_ADDRESS)), time_buffer, true); + if (!MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::CLOCK_US_STYLE)) [[likely]] { + MQTT_Manager::publish(fmt::format("nspanel/mqttmanager_{}/status/ampm", MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::MANAGER_ADDRESS)), "", true); + } else { + MQTT_Manager::publish(fmt::format("nspanel/mqttmanager_{}/status/ampm", MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::MANAGER_ADDRESS)), ampm_buffer, true); + } + + MQTT_Manager::publish(fmt::format("nspanel/mqttmanager_{}/status/date", MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::MANAGER_ADDRESS)), date_buffer, true); + ManagerClock::publish_epoch(); + + // Sleep until next minute + auto t_now = std::chrono::system_clock::now(); + auto t_now_c = std::chrono::system_clock::to_time_t(t_now); + std::tm next_minute_tm = *std::localtime(&t_now_c); + next_minute_tm.tm_sec = 1; // Always wait to minute change over and not the exact same second + + next_minute_tm.tm_min++; + if (next_minute_tm.tm_min >= 60) { // We went over to next hour + next_minute_tm.tm_min = 0; + next_minute_tm.tm_hour++; + if (next_minute_tm.tm_hour >= 24) { // We went over to next day + next_minute_tm.tm_hour = 0; + next_minute_tm.tm_mday++; + // mktime (called below) normalises month/year overflow and handles + // varying month lengths and leap years, so no manual rollover needed. + } + } + + auto t_next_minute = std::chrono::system_clock::from_time_t(std::mktime(&next_minute_tm)); + std::this_thread::sleep_until(t_next_minute); + } +} + +void ManagerClock::publish_epoch() { + std::string topic = fmt::format("nspanel/mqttmanager_{}/status/epoch", MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::MANAGER_ADDRESS)); + MQTT_Manager::publish_if_connected(topic, [] { + return ManagerClock::epoch_payload(std::chrono::system_clock::now()); + }); +} + +std::optional ManagerClock::epoch_payload(std::chrono::system_clock::time_point now) { + if (now < ManagerClock::_earliest_plausible_time) { + if (!ManagerClock::_warned_about_unset_clock.exchange(true)) { + SPDLOG_WARN("System clock reads {} ms since the Unix epoch, which is too early to be right. Will not send the time to NSPanels until the clock is set.", std::chrono::duration_cast(now.time_since_epoch()).count()); + } + return std::nullopt; + } + + ManagerClock::_warned_about_unset_clock = false; + return std::to_string(std::chrono::duration_cast(now.time_since_epoch()).count()); +} diff --git a/docker/MQTTManager/include/manager_clock/manager_clock.hpp b/docker/MQTTManager/include/manager_clock/manager_clock.hpp new file mode 100644 index 00000000..41cfa200 --- /dev/null +++ b/docker/MQTTManager/include/manager_clock/manager_clock.hpp @@ -0,0 +1,58 @@ +#ifndef MQTTMANAGER_MANAGER_CLOCK +#define MQTTMANAGER_MANAGER_CLOCK + +#include +#include +#include +#include +#include + +/** + * Tells NSPanels the current time. + * + * Every minute the time and date to display are sent, retained, on nspanel/mqttmanager_
/status/time, + * status/ampm and status/date, formatted as configured. + * + * The time is also sent as Unix time in milliseconds on nspanel/mqttmanager_
/status/epoch so that + * panels can set their own clock. It is never retained, so a panel only ever receives a time that was current + * when it was sent. It is sent every minute, whenever the manager connects to MQTT, and whenever a panel comes + * online. + */ +class ManagerClock { +public: + /** + * Start the thread that sends the time every minute, and send the time each time the manager connects + * to MQTT. Call before MQTT_Manager::init(). + */ + static void start(); + + /** + * Send the time each time the manager connects to MQTT. Called by start(). + */ + static void init(); + + /** + * Send the current time now, if connected to MQTT. + */ + static void publish_epoch(); + + /** + * The epoch payload for the given time: Unix time in milliseconds. Empty if the time is too early + * to be right, as when the host clock has not been set yet. + */ + static std::optional epoch_payload(std::chrono::system_clock::time_point now); + +private: + static inline std::thread _instance; + + // Send the time and date at the start of every minute. Never returns. + static void _run_clock_thread(); + + // 2025-01-01T00:00:00Z. Anything earlier means the host clock has not been set. + static constexpr std::chrono::system_clock::time_point _earliest_plausible_time{std::chrono::seconds(1735689600)}; + + // Warn once when the host clock is not set, not every minute. + static inline std::atomic _warned_about_unset_clock = false; +}; + +#endif // !MQTTMANAGER_MANAGER_CLOCK diff --git a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp b/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp index 5d697f90..23b830a8 100644 --- a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp +++ b/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp @@ -295,6 +295,8 @@ void MQTT_Manager::_reconnect_mqtt_client() { } catch (std::exception &e) { SPDLOG_ERROR("Caught exception trying to subscribe to topics: {}", boost::diagnostic_information(e, true)); } + + MQTT_Manager::_connected_callbacks(); } void MQTT_Manager::_process_mqtt_messages() { @@ -356,18 +358,7 @@ void MQTT_Manager::publish(const std::string &topic, const std::string &payload, } } - // Replicate messages into the websocket STOMP topics. - // Convert payload to base64 as to be able to send it over the websocket. - std::size_t encoded_size = (payload.size() + 2) / 3 * 4; - std::vector encoded_buffer(encoded_size); - // Encode the string - std::size_t written = boost::beast::detail::base64::encode( - encoded_buffer.data(), - payload.data(), - payload.size()); - encoded_buffer.resize(written); - WebsocketServer::set_stomp_topic_retained(fmt::format("mqtt/{}", topic), retain); - WebsocketServer::update_stomp_topic_value(fmt::format("mqtt/{}", topic), std::string(encoded_buffer.begin(), encoded_buffer.end())); + MQTT_Manager::_mirror_to_stomp(topic, payload, retain); } else { MQTT_Manager::_mqtt_messages_buffer.push_back(msg); } @@ -376,6 +367,64 @@ void MQTT_Manager::publish(const std::string &topic, const std::string &payload, } } +bool MQTT_Manager::publish_if_connected(const std::string &topic, std::function()> make_payload) { + if (topic.empty()) { + SPDLOG_ERROR("Tried to publish to empty topic."); + return false; + } + + std::lock_guard mutex_guard(MQTT_Manager::_mqtt_client_mutex); + // Create the payload only now that the lock is held, so it is current when it is sent. + std::optional payload = make_payload(); + if (!payload.has_value()) { + return false; + } + + // STOMP mirrors every message, whether or not MQTT is connected. + MQTT_Manager::_mirror_to_stomp(topic, payload.value(), false); + + // Also dropped while still sending what was buffered after reconnecting: the connected callbacks + // run once that is done. + if (!MQTT_Manager::is_connected()) { + SPDLOG_DEBUG("Not connected to MQTT, will not send message to '{}'.", topic); + return false; + } + + try { + SPDLOG_TRACE("Publising '{}' -> '{}'", topic, payload.value()); + MQTT_Manager::_send(mqtt::make_message(topic.c_str(), payload->c_str(), payload->size(), 2, false)); + } catch (std::exception &ex) { + SPDLOG_WARN("Caught exception while trying to publish '{}' -> '{}'. Will not retry.", topic, payload.value()); + return false; + } + return true; +} + +void MQTT_Manager::attach_connected_callback(std::function callback) { + MQTT_Manager::_connected_callbacks.connect(callback); +} + +void MQTT_Manager::_mirror_to_stomp(const std::string &topic, const std::string &payload, bool retain) { +#if defined(TEST_MODE) && TEST_MODE == 1 + { + std::lock_guard lock_guard(MQTT_Manager::_test_published_messages_mutex); + MQTT_Manager::_test_mirrored_messages.push_back({topic, payload, retain}); + } +#endif + + // Convert payload to base64 as to be able to send it over the websocket. + std::size_t encoded_size = (payload.size() + 2) / 3 * 4; + std::vector encoded_buffer(encoded_size); + // Encode the string + std::size_t written = boost::beast::detail::base64::encode( + encoded_buffer.data(), + payload.data(), + payload.size()); + encoded_buffer.resize(written); + WebsocketServer::set_stomp_topic_retained(fmt::format("mqtt/{}", topic), retain); + WebsocketServer::update_stomp_topic_value(fmt::format("mqtt/{}", topic), std::string(encoded_buffer.begin(), encoded_buffer.end())); +} + void MQTT_Manager::publish_protobuf(const std::string &topic, google::protobuf::Message &payload, bool retain) { std::string payload_string; if (payload.SerializeToString(&payload_string)) { diff --git a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp b/docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp index 19b84b0b..a8084958 100644 --- a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp +++ b/docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp @@ -12,6 +12,7 @@ #include #include #include +#include #include #include #include @@ -110,6 +111,26 @@ class MQTT_Manager { */ static void publish_protobuf(const std::string &topic, google::protobuf::Message &payload, bool retain); + /** + * Will send a non-retained MQTT message with QoS 2, but only if connected right now. The message is + * never buffered while disconnected or retried on failure, so it suits payloads that are only valid at + * the moment they are sent, such as the current time. (The MQTT client itself may still resend it + * after reconnecting if the connection dropped before the server acknowledged it.) + * The message is mirrored to STOMP whether or not MQTT is connected. + * @param topic: The MQTT topic to send payload to. + * @param make_payload: Called with the MQTT client locked, immediately before sending. Must not call + * back into MQTT_Manager. Return std::nullopt to send nothing. + * @return true if the message was sent over MQTT. + */ + static bool publish_if_connected(const std::string &topic, std::function()> make_payload); + + /** + * Attach a callback that is called each time the connection to the MQTT server has been + * (re)established and every topic has been subscribed to again. + * @param callback: Function to call. + */ + static void attach_connected_callback(std::function callback); + /** * Will clear retain on a topic (not recursive) * @param topic: The MQTT topic to send payload to. @@ -129,6 +150,17 @@ class MQTT_Manager { return std::exchange(MQTT_Manager::_test_published_messages, {}); } + // Number of messages waiting to be sent once connected. + static size_t test_buffered_message_count() { + std::lock_guard mutex_guard(MQTT_Manager::_mqtt_client_mutex); + return MQTT_Manager::_mqtt_messages_buffer.size(); + } + + // Calls the connected callbacks as if a connection to the MQTT server had just been established. + static void test_notify_connected() { + MQTT_Manager::_connected_callbacks(); + } + // Number of callbacks attached to a topic. static size_t test_callback_count(const std::string &topic) { std::lock_guard mutex_guard(MQTT_Manager::_mqtt_client_mutex); @@ -155,6 +187,13 @@ class MQTT_Manager { std::lock_guard lock_guard(MQTT_Manager::_test_published_messages_mutex); return std::exchange(MQTT_Manager::_test_sent_messages, {}); } + + // Everything mirrored to STOMP since the last call, in order, with the MQTT topic (without the + // "mqtt/" prefix) and the payload before base64 encoding. + static std::vector test_take_mirrored_messages() { + std::lock_guard lock_guard(MQTT_Manager::_test_published_messages_mutex); + return std::exchange(MQTT_Manager::_test_mirrored_messages, {}); + } #endif private: @@ -162,6 +201,7 @@ class MQTT_Manager { static inline std::mutex _test_published_messages_mutex; static inline std::vector _test_published_messages; static inline std::vector _test_sent_messages; + static inline std::vector _test_mirrored_messages; static inline std::atomic _test_broker_connected = false; #endif static inline boost::lockfree::spsc_queue> _mqtt_message_queue; @@ -193,6 +233,12 @@ class MQTT_Manager { static inline boost::ptr_map> _mqtt_callbacks; static inline std::unordered_map _subscribed_topics; + static inline boost::signals2::signal _connected_callbacks; + + /** + * Replicate a sent MQTT message into the websocket STOMP topic "mqtt/". + */ + static void _mirror_to_stomp(const std::string &topic, const std::string &payload, bool retain); static void _process_mqtt_messages(); static void _process_mqtt_command(nlohmann::json &data); diff --git a/docker/MQTTManager/include/nspanel/nspanel.cpp b/docker/MQTTManager/include/nspanel/nspanel.cpp index 22f95cb3..3c1cdfcd 100644 --- a/docker/MQTTManager/include/nspanel/nspanel.cpp +++ b/docker/MQTTManager/include/nspanel/nspanel.cpp @@ -1,6 +1,7 @@ #include "nspanel.hpp" #include "database_manager/database_manager.hpp" #include "entity_manager/entity_manager.hpp" +#include "manager_clock/manager_clock.hpp" #include "mqtt_manager/mqtt_manager.hpp" #include "mqtt_manager_config/mqtt_manager_config.hpp" #include "protobuf_nspanel.pb.h" @@ -649,6 +650,8 @@ void NSPanel::mqtt_callback(std::string topic, std::string payload) { if (state.compare("online") == 0) { this->_state = MQTT_MANAGER_NSPANEL_STATE::ONLINE; SPDLOG_DEBUG("NSPanel {}::{} became ONLINE.", this->_id, this->_name); + // A panel that has just started or reconnected has no clock until it is told the time. + ManagerClock::publish_epoch(); } else if (state.compare("offline") == 0) { this->_state = MQTT_MANAGER_NSPANEL_STATE::OFFLINE; SPDLOG_DEBUG("NSPanel {}::{} became OFFLINE.", this->_id, this->_name); diff --git a/docker/MQTTManager/src/main.cpp b/docker/MQTTManager/src/main.cpp index d022c4da..b163bce9 100644 --- a/docker/MQTTManager/src/main.cpp +++ b/docker/MQTTManager/src/main.cpp @@ -1,6 +1,7 @@ #include "command_manager/command_manager.hpp" #include "database_manager/database_manager.hpp" #include "entity/entity.hpp" +#include "manager_clock/manager_clock.hpp" #include "openhab_manager/openhab_manager.hpp" #include "spdlog/sinks/ansicolor_sink.h" #include "spdlog/sinks/rotating_file_sink.h" @@ -52,64 +53,6 @@ void sigusr1_handler(int signal) { } } -void publish_time_and_date() { - // Set timezone - setenv("TZ", MqttManagerConfig::timezone.c_str(), 1); - tzset(); - - // Timezone loaded, proceed to update MQTT - for (;;) { - char time_buffer[20]; - char ampm_buffer[20]; - std::string time_str; - char date_buffer[100]; - std::string date_str; - - std::time_t time = std::time({}); - std::strftime(date_buffer, 100, MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::DATE_FORMAT).c_str(), std::localtime(&time)); - date_str = date_buffer; - - if (!MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::CLOCK_US_STYLE)) [[likely]] { - std::strftime(time_buffer, 20, "%H:%M", std::localtime(&time)); - time_str = time_buffer; - } else { - std::strftime(time_buffer, 20, "%I:%M", std::localtime(&time)); - std::strftime(ampm_buffer, 20, "%p", std::localtime(&time)); - time_str = time_buffer; - } - - MQTT_Manager::publish(fmt::format("nspanel/mqttmanager_{}/status/time", MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::MANAGER_ADDRESS)), time_buffer, true); - if (!MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::CLOCK_US_STYLE)) [[likely]] { - MQTT_Manager::publish(fmt::format("nspanel/mqttmanager_{}/status/ampm", MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::MANAGER_ADDRESS)), "", true); - } else { - MQTT_Manager::publish(fmt::format("nspanel/mqttmanager_{}/status/ampm", MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::MANAGER_ADDRESS)), ampm_buffer, true); - } - - MQTT_Manager::publish(fmt::format("nspanel/mqttmanager_{}/status/date", MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::MANAGER_ADDRESS)), date_buffer, true); - - // Sleep until next minute - auto t_now = std::chrono::system_clock::now(); - auto t_now_c = std::chrono::system_clock::to_time_t(t_now); - std::tm next_minute_tm = *std::localtime(&t_now_c); - next_minute_tm.tm_sec = 1; // Always wait to minute change over and not the exact same second - - next_minute_tm.tm_min++; - if (next_minute_tm.tm_min >= 60) { // We went over to next hour - next_minute_tm.tm_min = 0; - next_minute_tm.tm_hour++; - if (next_minute_tm.tm_hour >= 24) { // We went over to next day - next_minute_tm.tm_hour = 0; - next_minute_tm.tm_mday++; - // mktime (called below) normalises month/year overflow and handles - // varying month lengths and leap years, so no manual rollover needed. - } - } - - auto t_next_minute = std::chrono::system_clock::from_time_t(std::mktime(&next_minute_tm)); - std::this_thread::sleep_until(t_next_minute); - } -} - int main(int argc, char *argv[]) { // If we are in test mode, don't start the MQTTManager. Simply run all test then exit. #if defined(TEST_MODE) && TEST_MODE == 1 @@ -197,7 +140,6 @@ int main(int argc, char *argv[]) { std::thread openhab_manager_thread; std::thread websocket_server_thread; std::thread nextion_image_server_thread; - std::thread time_and_date_thread; SPDLOG_INFO("Starting Websocket Server on port 8002."); websocket_server_thread = std::thread(WebsocketServer::start); @@ -207,7 +149,7 @@ int main(int argc, char *argv[]) { SPDLOG_INFO("Config loaded. Starting components."); - time_and_date_thread = std::thread(publish_time_and_date); + ManagerClock::start(); // Send the time and date to NSPanels every minute and each time MQTT connects home_assistant_manager_thread = std::thread(HomeAssistantManager::connect); openhab_manager_thread = std::thread(OpenhabManager::connect); diff --git a/docker/MQTTManager/tests/CMakeLists.txt b/docker/MQTTManager/tests/CMakeLists.txt index b649bf8b..d23b8ab7 100644 --- a/docker/MQTTManager/tests/CMakeLists.txt +++ b/docker/MQTTManager/tests/CMakeLists.txt @@ -30,7 +30,7 @@ target_link_options(nspm_mqttmanager_tests PRIVATE -rdynamic) target_link_libraries(nspm_mqttmanager_tests PRIVATE ${NSPM_LIBRARIES_WITH_INLINE_TESTS} MQTTManager_DatabaseManager MQTTManager_Entity - MQTT_Manager MQTTManager_HomeAssistantManager MQTTManager_OpenhabManager MQTTManager_EntityManager + MQTT_Manager MQTTManager_ManagerClock MQTTManager_HomeAssistantManager MQTTManager_OpenhabManager MQTTManager_EntityManager MQTTManager_Scene MQTTManager_Room MQTTManager_WebsocketServer MQTTManager_CommandManager MQTTManager_Switch MQTTManager_Button MQTTManager_Thermostat MQTTManager_MediaPlayer MQTTManager_NSPanel ixwebsocket::ixwebsocket gtest::gtest diff --git a/docker/MQTTManager/tests/manager_clock_test.cpp b/docker/MQTTManager/tests/manager_clock_test.cpp new file mode 100644 index 00000000..49f5ca4d --- /dev/null +++ b/docker/MQTTManager/tests/manager_clock_test.cpp @@ -0,0 +1,93 @@ +#include "test_helpers.hpp" +#include +#include +#include +#include +#include +#include + +using nspm_test::ScopedErrorLog; +using nspm_test::ScopedBrokerConnected; + +namespace { + +std::chrono::system_clock::time_point unix_ms(int64_t milliseconds) { + return std::chrono::system_clock::time_point(std::chrono::milliseconds(milliseconds)); +} + +int64_t now_unix_ms() { + return std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count(); +} + +} // namespace + +TEST(ManagerClockTest, epoch_payload_is_unix_time_in_milliseconds) { + EXPECT_EQ(ManagerClock::epoch_payload(unix_ms(1760724061234)), "1760724061234"); +} + +TEST(ManagerClockTest, epoch_payload_drops_parts_of_a_millisecond) { + auto time = unix_ms(1760724061234) + std::chrono::microseconds(999); + EXPECT_EQ(ManagerClock::epoch_payload(time), "1760724061234"); +} + +TEST(ManagerClockTest, epoch_payload_is_empty_before_2025) { + EXPECT_FALSE(ManagerClock::epoch_payload(unix_ms(0)).has_value()); + EXPECT_FALSE(ManagerClock::epoch_payload(unix_ms(1735689599999)).has_value()); // 2024-12-31T23:59:59.999Z + EXPECT_EQ(ManagerClock::epoch_payload(unix_ms(1735689600000)), "1735689600000"); // 2025-01-01T00:00:00Z +} + +TEST(ManagerClockTest, unset_clock_is_warned_about_once_until_it_is_set) { + ManagerClock::epoch_payload(unix_ms(1760724061234)); // Start from a set clock. + ScopedErrorLog log(spdlog::level::warn); + + ManagerClock::epoch_payload(unix_ms(0)); + ManagerClock::epoch_payload(unix_ms(60000)); + EXPECT_EQ(log.errors().size(), 1); + + ManagerClock::epoch_payload(unix_ms(1760724061234)); + ManagerClock::epoch_payload(unix_ms(0)); + EXPECT_EQ(log.errors().size(), 2); +} + +TEST(ManagerClockTest, publish_epoch_sends_the_current_time_unretained) { + ScopedBrokerConnected connected; + + int64_t before = now_unix_ms(); + ManagerClock::publish_epoch(); + int64_t after = now_unix_ms(); + + auto messages = nspm_test::mqtt_sent_to(nspm_test::epoch_topic()); + ASSERT_EQ(messages.size(), 1); + EXPECT_FALSE(messages[0].retain); + int64_t sent = std::stoll(messages[0].payload); + EXPECT_GE(sent, before); + EXPECT_LE(sent, after); +} + +TEST(ManagerClockTest, publish_epoch_does_not_buffer_the_time_while_disconnected) { + size_t buffered = MQTT_Manager::test_buffered_message_count(); + + ManagerClock::publish_epoch(); + + EXPECT_EQ(MQTT_Manager::test_buffered_message_count(), buffered); +} + +TEST(ManagerClockTest, publish_epoch_mirrors_the_time_to_stomp_while_disconnected) { + nspm_test::stomp_mirrored_to(nspm_test::epoch_topic()); + + ManagerClock::publish_epoch(); + + auto mirrored = nspm_test::stomp_mirrored_to(nspm_test::epoch_topic()); + ASSERT_EQ(mirrored.size(), 1); + EXPECT_FALSE(mirrored[0].retain); +} + +// ManagerClock::init() attaches for the rest of the test run, so this is the only test calling it. +TEST(ManagerClockTest, time_is_sent_when_mqtt_connects) { + ManagerClock::init(); + ScopedBrokerConnected connected; + + MQTT_Manager::test_notify_connected(); + + EXPECT_EQ(nspm_test::mqtt_sent_to(nspm_test::epoch_topic()).size(), 1); +} diff --git a/docker/MQTTManager/tests/mqtt_manager_test.cpp b/docker/MQTTManager/tests/mqtt_manager_test.cpp new file mode 100644 index 00000000..4be07c72 --- /dev/null +++ b/docker/MQTTManager/tests/mqtt_manager_test.cpp @@ -0,0 +1,81 @@ +#include "test_helpers.hpp" +#include +#include +#include +#include + +using nspm_test::ScopedBrokerConnected; + +namespace { + +const std::string LIVE_TOPIC = "test/mqtt_manager/live"; + +std::optional payload() { + return "payload"; +} + +} // namespace + +TEST(MqttManagerTest, publish_if_connected_sends_unretained) { + ScopedBrokerConnected connected; + nspm_test::stomp_mirrored_to(LIVE_TOPIC); + + EXPECT_TRUE(MQTT_Manager::publish_if_connected(LIVE_TOPIC, payload)); + + auto messages = nspm_test::mqtt_sent_to(LIVE_TOPIC); + ASSERT_EQ(messages.size(), 1); + EXPECT_EQ(messages[0].payload, "payload"); + EXPECT_FALSE(messages[0].retain); + + auto mirrored = nspm_test::stomp_mirrored_to(LIVE_TOPIC); + ASSERT_EQ(mirrored.size(), 1); + EXPECT_EQ(mirrored[0].payload, "payload"); + EXPECT_FALSE(mirrored[0].retain); +} + +TEST(MqttManagerTest, publish_if_connected_drops_messages_while_disconnected) { + size_t buffered = MQTT_Manager::test_buffered_message_count(); + + EXPECT_FALSE(MQTT_Manager::publish_if_connected(LIVE_TOPIC, payload)); + EXPECT_EQ(MQTT_Manager::test_buffered_message_count(), buffered); +} + +TEST(MqttManagerTest, publish_if_connected_mirrors_to_stomp_while_disconnected) { + nspm_test::stomp_mirrored_to(LIVE_TOPIC); + + EXPECT_FALSE(MQTT_Manager::publish_if_connected(LIVE_TOPIC, payload)); + + auto mirrored = nspm_test::stomp_mirrored_to(LIVE_TOPIC); + ASSERT_EQ(mirrored.size(), 1); + EXPECT_EQ(mirrored[0].payload, "payload"); + EXPECT_FALSE(mirrored[0].retain); +} + +TEST(MqttManagerTest, publish_if_connected_drops_messages_until_caught_up_after_reconnecting) { + ScopedBrokerConnected connected; + MQTT_Manager::test_set_broker_connected(false); + MQTT_Manager::publish("test/mqtt_manager/buffered", "published while disconnected"); + MQTT_Manager::test_set_broker_connected(true); // Reconnected, but the buffered message is not sent yet. + size_t buffered = MQTT_Manager::test_buffered_message_count(); + nspm_test::stomp_mirrored_to(LIVE_TOPIC); + + EXPECT_FALSE(MQTT_Manager::publish_if_connected(LIVE_TOPIC, payload)); + EXPECT_EQ(MQTT_Manager::test_buffered_message_count(), buffered); + EXPECT_EQ(nspm_test::stomp_mirrored_to(LIVE_TOPIC).size(), 1); + + MQTT_Manager::test_send_buffered_messages(); + EXPECT_TRUE(nspm_test::mqtt_sent_to(LIVE_TOPIC).empty()); +} + +TEST(MqttManagerTest, publish_if_connected_sends_nothing_without_a_payload) { + ScopedBrokerConnected connected; + nspm_test::stomp_mirrored_to(LIVE_TOPIC); + + bool sent = MQTT_Manager::publish_if_connected(LIVE_TOPIC, []() -> std::optional { + return std::nullopt; + }); + + EXPECT_FALSE(sent); + EXPECT_TRUE(nspm_test::mqtt_sent_to(LIVE_TOPIC).empty()); + EXPECT_TRUE(nspm_test::stomp_mirrored_to(LIVE_TOPIC).empty()); +} diff --git a/docker/MQTTManager/tests/nspanel_test.cpp b/docker/MQTTManager/tests/nspanel_test.cpp index 28d818c0..69867602 100644 --- a/docker/MQTTManager/tests/nspanel_test.cpp +++ b/docker/MQTTManager/tests/nspanel_test.cpp @@ -699,6 +699,26 @@ TEST_F(NSPanelTest, commands_go_to_the_mac_and_legacy_name_topics) { EXPECT_EQ(panel->get_state(), MQTT_MANAGER_NSPANEL_STATE::WAITING); } +TEST_F(NSPanelTest, panel_coming_online_is_sent_the_time) { + load_panel(); + nspm_test::ScopedBrokerConnected connected; + + panel->mqtt_callback("nspanel/" + PANEL_MAC + "/status", R"({"state": "online"})"); + + auto messages = nspm_test::mqtt_sent_to(nspm_test::epoch_topic()); + ASSERT_EQ(messages.size(), 1); + EXPECT_FALSE(messages[0].retain); +} + +TEST_F(NSPanelTest, panel_going_offline_is_not_sent_the_time) { + load_panel(); + nspm_test::ScopedBrokerConnected connected; + + panel->mqtt_callback("nspanel/" + PANEL_MAC + "/status", R"({"state": "offline"})"); + + EXPECT_TRUE(nspm_test::mqtt_sent_to(nspm_test::epoch_topic()).empty()); +} + TEST_F(NSPanelTest, mqtt_payload_button_publishes_the_configured_message) { update_row([](auto &settings) { settings.button2_mode = MQTT_PAYLOAD; diff --git a/docker/MQTTManager/tests/test_helpers.hpp b/docker/MQTTManager/tests/test_helpers.hpp index 88e098bd..a2435381 100644 --- a/docker/MQTTManager/tests/test_helpers.hpp +++ b/docker/MQTTManager/tests/test_helpers.hpp @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include #include @@ -220,6 +221,53 @@ inline std::shared_ptr log_capture_sink() { return sink; } +// The topic the manager sends the current time on. +inline std::string epoch_topic() { + return fmt::format("nspanel/mqttmanager_{}/status/epoch", MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::MANAGER_ADDRESS)); +} + +// The messages sent to the MQTT server on a topic since the last call (messages on other topics are +// dropped). Only recorded while connected, see ScopedBrokerConnected. +inline std::vector mqtt_sent_to(const std::string &topic) { + std::vector messages; + for (auto &message : MQTT_Manager::test_take_sent_messages()) { + if (message.topic == topic) { + messages.push_back(message); + } + } + return messages; +} + +// The messages mirrored to STOMP for an MQTT topic since the last call (messages for other topics are +// dropped). Recorded whether or not connected. +inline std::vector stomp_mirrored_to(const std::string &topic) { + std::vector messages; + for (auto &message : MQTT_Manager::test_take_mirrored_messages()) { + if (message.topic == topic) { + messages.push_back(message); + } + } + return messages; +} + +// Connected to the MQTT server, and caught up on what was buffered, while it is in scope. What would +// be sent is recorded instead, see mqtt_sent_to(). Starts with nothing recorded as sent. +class ScopedBrokerConnected { +public: + ScopedBrokerConnected() { + MQTT_Manager::test_set_broker_connected(true); + MQTT_Manager::test_send_buffered_messages(); + MQTT_Manager::test_take_sent_messages(); + } + + ~ScopedBrokerConnected() { + MQTT_Manager::test_set_broker_connected(false); + } + + ScopedBrokerConnected(const ScopedBrokerConnected &) = delete; + ScopedBrokerConnected &operator=(const ScopedBrokerConnected &) = delete; +}; + // Records the errors (or, given a lower level, also the less severe messages) logged while it is in scope. class ScopedErrorLog { public: