MQTTManager: send NSPanels the current time

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()

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 <cstdlib>
+#include <ctime>
+#include <fmt/format.h>
+#include <spdlog/spdlog.h>
+
+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<std::string>(MQTT_MANAGER_SETTING::DATE_FORMAT).c_str(), std::localtime(&time));
+    date_str = date_buffer;
+
+    if (!MqttManagerConfig::get_setting_with_default<bool>(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<std::string>(MQTT_MANAGER_SETTING::MANAGER_ADDRESS)), time_buffer, true);
+    if (!MqttManagerConfig::get_setting_with_default<bool>(MQTT_MANAGER_SETTING::CLOCK_US_STYLE)) [[likely]] {
+      MQTT_Manager::publish(fmt::format("nspanel/mqttmanager_{}/status/ampm", MqttManagerConfig::get_setting_with_default<std::string>(MQTT_MANAGER_SETTING::MANAGER_ADDRESS)), "", true);
+    } else {
+      MQTT_Manager::publish(fmt::format("nspanel/mqttmanager_{}/status/ampm", MqttManagerConfig::get_setting_with_default<std::string>(MQTT_MANAGER_SETTING::MANAGER_ADDRESS)), ampm_buffer, true);
+    }
+
+    MQTT_Manager::publish(fmt::format("nspanel/mqttmanager_{}/status/date", MqttManagerConfig::get_setting_with_default<std::string>(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<std::string>(MQTT_MANAGER_SETTING::MANAGER_ADDRESS));
+  MQTT_Manager::publish_if_connected(topic, [] {
+    return ManagerClock::epoch_payload(std::chrono::system_clock::now());
+  });
+}
+
+std::optional<std::string> 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<std::chrono::milliseconds>(now.time_since_epoch()).count());
+    }
+    return std::nullopt;
+  }
+
+  ManagerClock::_warned_about_unset_clock = false;
+  return std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count());
+}

docker/MQTTManager/include/manager_clock/manager_clock.hpp

@@ -0,0 +1,58 @@
+#ifndef MQTTMANAGER_MANAGER_CLOCK
+#define MQTTMANAGER_MANAGER_CLOCK
+
+#include <atomic>
+#include <chrono>
+#include <optional>
+#include <string>
+#include <thread>
+
+/**
+ * Tells NSPanels the current time.
+ *
+ * Every minute the time and date to display are sent, retained, on nspanel/mqttmanager_<address>/status/time,
+ * status/ampm and status/date, formatted as configured.
+ *
+ * The time is also sent as Unix time in milliseconds on nspanel/mqttmanager_<address>/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<std::string> 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<bool> _warned_about_unset_clock = false;
+};
+
+#endif // !MQTTMANAGER_MANAGER_CLOCK

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<char> 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<std::optional<std::string>()> make_payload) {
+  if (topic.empty()) {
+    SPDLOG_ERROR("Tried to publish to empty topic.");
+    return false;
+  }
+
+  std::lock_guard<std::mutex> 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<std::string> 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<void()> 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<std::mutex> 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<char> 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)) {

docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp

@@ -12,6 +12,7 @@
 #include <mqtt/subscribe_options.h>
 #include <mutex>
 #include <nlohmann/json_fwd.hpp>
+#include <optional>
 #include <spdlog/common.h>
 #include <spdlog/spdlog.h>
 #include <string>
@@ -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<std::optional<std::string>()> 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<void()> 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<std::mutex> 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<std::mutex> mutex_guard(MQTT_Manager::_mqtt_client_mutex);
@@ -155,13 +187,21 @@ class MQTT_Manager {
     std::lock_guard<std::mutex> 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<TestPublishedMessage> test_take_mirrored_messages() {
+    std::lock_guard<std::mutex> lock_guard(MQTT_Manager::_test_published_messages_mutex);
+    return std::exchange(MQTT_Manager::_test_mirrored_messages, {});
+  }
 #endif
 
 private:
 #if defined(TEST_MODE) && TEST_MODE == 1
   static inline std::mutex _test_published_messages_mutex;
   static inline std::vector<TestPublishedMessage> _test_published_messages;
   static inline std::vector<TestPublishedMessage> _test_sent_messages;
+  static inline std::vector<TestPublishedMessage> _test_mirrored_messages;
   static inline std::atomic<bool> _test_broker_connected = false;
 #endif
   static inline boost::lockfree::spsc_queue<MQTTMessage, boost::lockfree::capacity<256>> _mqtt_message_queue;
@@ -193,6 +233,12 @@ class MQTT_Manager {
 
   static inline boost::ptr_map<std::string, boost::signals2::signal<void(std::string, std::string)>> _mqtt_callbacks;
   static inline std::unordered_map<std::string, int> _subscribed_topics;
+  static inline boost::signals2::signal<void()> _connected_callbacks;
+
+  /**
+   * Replicate a sent MQTT message into the websocket STOMP topic "mqtt/<topic>".
+   */
+  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);

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);