From faa5333481020ac4c7b4c38e5667a9c409617f3c Mon Sep 17 00:00:00 2001 From: James Mulcahy Date: Sat, 17 Oct 2026 13:22:59 -0700 Subject: [PATCH 1/4] MQTTManager: send NSPanels the current time as Unix milliseconds Co-Authored-By: Claude Opus 5.5 --- docker/MQTTManager/CMakeLists.txt | 9 +- .../include/manager_clock/manager_clock.cpp | 28 +++++++ .../include/manager_clock/manager_clock.hpp | 42 ++++++++++ .../include/mqtt_manager/mqtt_manager.cpp | 64 +++++++++++--- .../include/mqtt_manager/mqtt_manager.hpp | 36 ++++++++ .../MQTTManager/include/nspanel/nspanel.cpp | 3 + docker/MQTTManager/src/main.cpp | 3 + docker/MQTTManager/tests/CMakeLists.txt | 2 +- .../MQTTManager/tests/manager_clock_test.cpp | 83 +++++++++++++++++++ .../MQTTManager/tests/mqtt_manager_test.cpp | 67 +++++++++++++++ docker/MQTTManager/tests/nspanel_test.cpp | 20 +++++ docker/MQTTManager/tests/test_helpers.hpp | 36 ++++++++ 12 files changed, 378 insertions(+), 15 deletions(-) create mode 100644 docker/MQTTManager/include/manager_clock/manager_clock.cpp create mode 100644 docker/MQTTManager/include/manager_clock/manager_clock.hpp create mode 100644 docker/MQTTManager/tests/manager_clock_test.cpp create mode 100644 docker/MQTTManager/tests/mqtt_manager_test.cpp 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..af719195 --- /dev/null +++ b/docker/MQTTManager/include/manager_clock/manager_clock.cpp @@ -0,0 +1,28 @@ +#include "manager_clock.hpp" +#include "mqtt_manager/mqtt_manager.hpp" +#include "mqtt_manager_config/mqtt_manager_config.hpp" +#include +#include + +void ManagerClock::init() { + MQTT_Manager::attach_connected_callback(&ManagerClock::publish_epoch); +} + +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..12a98313 --- /dev/null +++ b/docker/MQTTManager/include/manager_clock/manager_clock.hpp @@ -0,0 +1,42 @@ +#ifndef MQTTMANAGER_MANAGER_CLOCK +#define MQTTMANAGER_MANAGER_CLOCK + +#include +#include +#include +#include + +/** + * Tells NSPanels the current time so they can set their own clock. + * + * The time is sent as Unix time in milliseconds on nspanel/mqttmanager_
/status/epoch. + * 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: + /** + * Send the time each time the manager connects to MQTT. Call before MQTT_Manager::init(). + */ + 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: + // 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..cf96db60 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,55 @@ 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); + // 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; + } + + // 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; + } + + try { + SPDLOG_TRACE("Publising '{}' -> '{}'", topic, payload.value()); + MQTT_Manager::_send(mqtt::make_message(topic.c_str(), payload->c_str(), payload->size(), 0, false)); + } catch (std::exception &ex) { + SPDLOG_WARN("Caught exception while trying to publish '{}' -> '{}'. Will not retry.", topic, payload.value()); + return false; + } + MQTT_Manager::_mirror_to_stomp(topic, payload.value(), 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) { + // 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..5b07597a 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,24 @@ class MQTT_Manager { */ static void publish_protobuf(const std::string &topic, google::protobuf::Message &payload, bool retain); + /** + * Will send a non-retained MQTT message, but only if connected right now. The message is never + * buffered for later or retried, so it suits payloads that are only valid at the moment they are + * sent, such as the current time. + * @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. + */ + 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 +148,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); @@ -193,6 +223,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..57fca474 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" @@ -86,6 +87,7 @@ void publish_time_and_date() { } 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(); @@ -215,6 +217,7 @@ int main(int argc, char *argv[]) { EntityManager::init(); SPDLOG_INFO("Entities loaded, connect to MQTT and start listening for commands."); + ManagerClock::init(); // Send the time to NSPanels each time MQTT connects MQTT_Manager::init(); // Initialize MQTT manager to handle everything to do with MQTT CommandManager::init(); 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..180b90a6 --- /dev/null +++ b/docker/MQTTManager/tests/manager_clock_test.cpp @@ -0,0 +1,83 @@ +#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); +} + +// 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..79355b8c --- /dev/null +++ b/docker/MQTTManager/tests/mqtt_manager_test.cpp @@ -0,0 +1,67 @@ +#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; + + 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); +} + +TEST(MqttManagerTest, publish_if_connected_drops_messages_while_disconnected) { + size_t buffered = MQTT_Manager::test_buffered_message_count(); + bool payload_made = false; + + bool sent = MQTT_Manager::publish_if_connected(LIVE_TOPIC, [&]() -> std::optional { + payload_made = true; + return "payload"; + }); + + EXPECT_FALSE(sent); + EXPECT_FALSE(payload_made) << "The payload should only be made when it can be sent at once."; + EXPECT_EQ(MQTT_Manager::test_buffered_message_count(), buffered); +} + +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(); + + EXPECT_FALSE(MQTT_Manager::publish_if_connected(LIVE_TOPIC, payload)); + EXPECT_EQ(MQTT_Manager::test_buffered_message_count(), buffered); + + 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; + + 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()); +} 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..aa69ecc1 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,41 @@ 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; +} + +// 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: From 76b0fd00921fcb59d7e8cfcbc2901898a83d3546 Mon Sep 17 00:00:00 2001 From: James Mulcahy Date: Sun, 18 Oct 2026 08:08:23 -0700 Subject: [PATCH 2/4] MQTTManager: move the time and date thread into ManagerClock publish_time_and_date() moves out of main.cpp unchanged, as ManagerClock::_run_clock_thread(). ManagerClock::start() attaches the MQTT connected callback and starts the thread, so main.cpp makes one call. Co-Authored-By: Claude Opus 5.5 --- .../include/manager_clock/manager_clock.cpp | 68 +++++++++++++++++++ .../include/manager_clock/manager_clock.hpp | 26 +++++-- docker/MQTTManager/src/main.cpp | 63 +---------------- 3 files changed, 90 insertions(+), 67 deletions(-) diff --git a/docker/MQTTManager/include/manager_clock/manager_clock.cpp b/docker/MQTTManager/include/manager_clock/manager_clock.cpp index af719195..7827c776 100644 --- a/docker/MQTTManager/include/manager_clock/manager_clock.cpp +++ b/docker/MQTTManager/include/manager_clock/manager_clock.cpp @@ -1,13 +1,81 @@ #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, [] { diff --git a/docker/MQTTManager/include/manager_clock/manager_clock.hpp b/docker/MQTTManager/include/manager_clock/manager_clock.hpp index 12a98313..41cfa200 100644 --- a/docker/MQTTManager/include/manager_clock/manager_clock.hpp +++ b/docker/MQTTManager/include/manager_clock/manager_clock.hpp @@ -5,18 +5,29 @@ #include #include #include +#include /** - * Tells NSPanels the current time so they can set their own clock. + * Tells NSPanels the current time. * - * The time is sent as Unix time in milliseconds on nspanel/mqttmanager_
/status/epoch. - * 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. + * 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: /** - * Send the time each time the manager connects to MQTT. Call before MQTT_Manager::init(). + * 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(); @@ -32,6 +43,11 @@ class ManagerClock { 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)}; diff --git a/docker/MQTTManager/src/main.cpp b/docker/MQTTManager/src/main.cpp index 57fca474..b163bce9 100644 --- a/docker/MQTTManager/src/main.cpp +++ b/docker/MQTTManager/src/main.cpp @@ -53,65 +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); - 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); - } -} - 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 @@ -199,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); @@ -209,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); @@ -217,7 +157,6 @@ int main(int argc, char *argv[]) { EntityManager::init(); SPDLOG_INFO("Entities loaded, connect to MQTT and start listening for commands."); - ManagerClock::init(); // Send the time to NSPanels each time MQTT connects MQTT_Manager::init(); // Initialize MQTT manager to handle everything to do with MQTT CommandManager::init(); From bd2239da6a499ff1e92b3bb36d474650d7fa43c2 Mon Sep 17 00:00:00 2001 From: James Mulcahy Date: Sun, 18 Oct 2026 08:08:23 -0700 Subject: [PATCH 3/4] MQTTManager: send the epoch with QoS 2 Co-Authored-By: Claude Opus 5.5 --- docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp | 2 +- docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp | 7 ++++--- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp b/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp index cf96db60..6ab7a09b 100644 --- a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp +++ b/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp @@ -389,7 +389,7 @@ bool MQTT_Manager::publish_if_connected(const std::string &topic, std::function< try { SPDLOG_TRACE("Publising '{}' -> '{}'", topic, payload.value()); - MQTT_Manager::_send(mqtt::make_message(topic.c_str(), payload->c_str(), payload->size(), 0, false)); + 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; diff --git a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp b/docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp index 5b07597a..92ad12b9 100644 --- a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp +++ b/docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp @@ -112,9 +112,10 @@ class MQTT_Manager { static void publish_protobuf(const std::string &topic, google::protobuf::Message &payload, bool retain); /** - * Will send a non-retained MQTT message, but only if connected right now. The message is never - * buffered for later or retried, so it suits payloads that are only valid at the moment they are - * sent, such as the current time. + * 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.) * @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. From d7567ff1c04576d7375611dee01dd121c3923fe1 Mon Sep 17 00:00:00 2001 From: James Mulcahy Date: Tue, 20 Oct 2026 07:23:23 -0700 Subject: [PATCH 4/4] MQTTManager: mirror publish_if_connected() to STOMP while disconnected STOMP is a full mirror of what is published, so the message goes to STOMP before the connected check rather than only after a successful MQTT send. The return value still means "sent over MQTT". Co-Authored-By: Claude Opus 5.5 --- .../include/mqtt_manager/mqtt_manager.cpp | 23 +++++++++----- .../include/mqtt_manager/mqtt_manager.hpp | 11 ++++++- .../MQTTManager/tests/manager_clock_test.cpp | 10 +++++++ .../MQTTManager/tests/mqtt_manager_test.cpp | 30 ++++++++++++++----- docker/MQTTManager/tests/test_helpers.hpp | 12 ++++++++ 5 files changed, 70 insertions(+), 16 deletions(-) diff --git a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp b/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp index 6ab7a09b..23b830a8 100644 --- a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp +++ b/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp @@ -374,6 +374,15 @@ bool MQTT_Manager::publish_if_connected(const std::string &topic, std::function< } 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()) { @@ -381,12 +390,6 @@ bool MQTT_Manager::publish_if_connected(const std::string &topic, std::function< return false; } - // 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; - } - try { SPDLOG_TRACE("Publising '{}' -> '{}'", topic, payload.value()); MQTT_Manager::_send(mqtt::make_message(topic.c_str(), payload->c_str(), payload->size(), 2, false)); @@ -394,7 +397,6 @@ bool MQTT_Manager::publish_if_connected(const std::string &topic, std::function< SPDLOG_WARN("Caught exception while trying to publish '{}' -> '{}'. Will not retry.", topic, payload.value()); return false; } - MQTT_Manager::_mirror_to_stomp(topic, payload.value(), false); return true; } @@ -403,6 +405,13 @@ void MQTT_Manager::attach_connected_callback(std::function 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); diff --git a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp b/docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp index 92ad12b9..a8084958 100644 --- a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp +++ b/docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp @@ -116,10 +116,11 @@ class MQTT_Manager { * 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. + * @return true if the message was sent over MQTT. */ static bool publish_if_connected(const std::string &topic, std::function()> make_payload); @@ -186,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: @@ -193,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; diff --git a/docker/MQTTManager/tests/manager_clock_test.cpp b/docker/MQTTManager/tests/manager_clock_test.cpp index 180b90a6..49f5ca4d 100644 --- a/docker/MQTTManager/tests/manager_clock_test.cpp +++ b/docker/MQTTManager/tests/manager_clock_test.cpp @@ -72,6 +72,16 @@ TEST(ManagerClockTest, publish_epoch_does_not_buffer_the_time_while_disconnected 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(); diff --git a/docker/MQTTManager/tests/mqtt_manager_test.cpp b/docker/MQTTManager/tests/mqtt_manager_test.cpp index 79355b8c..4be07c72 100644 --- a/docker/MQTTManager/tests/mqtt_manager_test.cpp +++ b/docker/MQTTManager/tests/mqtt_manager_test.cpp @@ -18,6 +18,7 @@ std::optional payload() { 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)); @@ -25,31 +26,42 @@ TEST(MqttManagerTest, publish_if_connected_sends_unretained) { 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(); - bool payload_made = false; - - bool sent = MQTT_Manager::publish_if_connected(LIVE_TOPIC, [&]() -> std::optional { - payload_made = true; - return "payload"; - }); - EXPECT_FALSE(sent); - EXPECT_FALSE(payload_made) << "The payload should only be made when it can be sent at once."; + 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()); @@ -57,6 +69,7 @@ TEST(MqttManagerTest, publish_if_connected_drops_messages_until_caught_up_after_ 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; @@ -64,4 +77,5 @@ TEST(MqttManagerTest, publish_if_connected_sends_nothing_without_a_payload) { 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/test_helpers.hpp b/docker/MQTTManager/tests/test_helpers.hpp index aa69ecc1..a2435381 100644 --- a/docker/MQTTManager/tests/test_helpers.hpp +++ b/docker/MQTTManager/tests/test_helpers.hpp @@ -238,6 +238,18 @@ inline std::vector mqtt_sent_to(const std::s 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 {