From 4d3d3bcec8361a78061bf8e10135f9ac13c1c188 Mon Sep 17 00:00:00 2001 From: Donald Clark Jackson Date: Fri, 9 Oct 2026 07:14:38 -0700 Subject: [PATCH 1/5] Add the ebus_mqtt target and move PublishHold into it ebus_mqtt is the first component of the split: mqtt/include/ebus/mqtt/ and mqtt/src/, built by mqtt/CMakeLists.txt with only its own include directory and no dependency. ebus_core links it, so consumers keep one target. PublishHold moves from ports/posix/ unchanged in behavior. Its sizes are now EBUS_MQTT_HOLD_ENTRIES, EBUS_MQTT_HOLD_TOPIC_MAX and EBUS_MQTT_HOLD_PAYLOAD_MAX, and it no longer logs: hold() reports EVICTED_OLDEST and DROPPED_TOO_LARGE and the POSIX port logs them. Its test becomes the Unity suite test_mqtt_publish_hold, which links ebus_mqtt alone. The header check compiles each component's headers against that component's target. library.json compiles src/ and mqtt/src/ and adds mqtt/include to the include path, which PlatformIO passes on to the library's dependents. Part of #3 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Npk6WmkpnBRzWkuhK6uCuz --- CMakeLists.txt | 41 +++-- library.json | 9 +- mqtt/CMakeLists.txt | 13 ++ .../include/ebus/mqtt}/publish_hold.h | 39 +++-- {ports/posix => mqtt}/src/publish_hold.cpp | 21 +-- ports/posix/CMakeLists.txt | 6 - .../posix/include/ebus_posix/paho_transport.h | 5 +- ports/posix/src/paho_transport.cpp | 15 +- ports/posix/test/test_publish_hold.cpp | 122 --------------- test/CMakeLists.txt | 7 +- .../test_mqtt_publish_hold.cpp | 144 ++++++++++++++++++ 11 files changed, 247 insertions(+), 175 deletions(-) create mode 100644 mqtt/CMakeLists.txt rename {ports/posix/include/ebus_posix => mqtt/include/ebus/mqtt}/publish_hold.h (56%) rename {ports/posix => mqtt}/src/publish_hold.cpp (69%) delete mode 100644 ports/posix/test/test_publish_hold.cpp create mode 100644 test/test_mqtt_publish_hold/test_mqtt_publish_hold.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 86d2628..e44f3d4 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1,8 +1,11 @@ # Builds the core without PlatformIO, for Zephyr, embedded Linux, or a host check. -# Usage from a parent project: add_subdirectory(path/to/cpp-sdk) and link ebus_core. +# Usage from a parent project: add_subdirectory(path/to/cpp-sdk) and link ebus_core, the +# umbrella, or one component target (ebus_mqtt). cmake_minimum_required(VERSION 3.18) project(ebus_core LANGUAGES CXX) +add_subdirectory(mqtt) + add_library(ebus_core STATIC src/homie/controller.cpp src/homie/controller_inbox.cpp @@ -22,6 +25,7 @@ add_library(ebus_core STATIC ) target_include_directories(ebus_core PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include) target_compile_features(ebus_core PUBLIC cxx_std_17) +target_link_libraries(ebus_core PUBLIC ebus_mqtt) # ArduinoJson, the core's one dependency (header-only). A parent project that already has # an ArduinoJson target keeps it. Otherwise the release PlatformIO uses (library.json, @@ -56,26 +60,31 @@ endif() if(_ebus_core_top_level AND CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang") target_compile_options(ebus_core PRIVATE -Wall -Wextra) + target_compile_options(ebus_mqtt PRIVATE -Wall -Wextra) endif() # Compiles every public header in a translation unit of its own, so a header that needs -# something it does not include (or includes a platform header) fails here. +# something it does not include (or includes a platform header) fails here. A component's +# headers are compiled against that component's target alone. option(EBUS_CORE_CHECK_HEADERS "Compile each public header on its own" ${_ebus_core_top_level}) if(EBUS_CORE_CHECK_HEADERS) - file(GLOB_RECURSE _ebus_core_headers RELATIVE ${CMAKE_CURRENT_SOURCE_DIR}/include - CONFIGURE_DEPENDS ${CMAKE_CURRENT_SOURCE_DIR}/include/*.h) - set(_ebus_core_header_tus "") - foreach(_header IN LISTS _ebus_core_headers) - string(MAKE_C_IDENTIFIER ${_header} _tu_name) - set(_tu ${CMAKE_CURRENT_BINARY_DIR}/header_check/${_tu_name}.cpp) - file(CONFIGURE OUTPUT ${_tu} CONTENT "#include <${_header}>\n") - list(APPEND _ebus_core_header_tus ${_tu}) - endforeach() - add_library(ebus_core_header_check OBJECT ${_ebus_core_header_tus}) - target_link_libraries(ebus_core_header_check PRIVATE ebus_core) - if(CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang") - target_compile_options(ebus_core_header_check PRIVATE -Wall -Wextra) - endif() + function(ebus_header_check target include_dir) + file(GLOB_RECURSE _headers RELATIVE ${include_dir} CONFIGURE_DEPENDS ${include_dir}/*.h) + set(_tus "") + foreach(_header IN LISTS _headers) + string(MAKE_C_IDENTIFIER ${_header} _tu_name) + set(_tu ${CMAKE_CURRENT_BINARY_DIR}/header_check/${target}/${_tu_name}.cpp) + file(CONFIGURE OUTPUT ${_tu} CONTENT "#include <${_header}>\n") + list(APPEND _tus ${_tu}) + endforeach() + add_library(${target}_header_check OBJECT ${_tus}) + target_link_libraries(${target}_header_check PRIVATE ${target}) + if(CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang") + target_compile_options(${target}_header_check PRIVATE -Wall -Wextra) + endif() + endfunction() + ebus_header_check(ebus_mqtt ${CMAKE_CURRENT_SOURCE_DIR}/mqtt/include) + ebus_header_check(ebus_core ${CMAKE_CURRENT_SOURCE_DIR}/include) endif() # The Unity suites in test/, one CTest test each. Off when the core is a subproject. diff --git a/library.json b/library.json index 05a0e77..84cafdc 100644 --- a/library.json +++ b/library.json @@ -19,6 +19,13 @@ ], "build": { "includeDir": "include", - "srcDir": "src" + "srcDir": ".", + "srcFilter": [ + "+", + "+" + ], + "flags": [ + "-I mqtt/include" + ] } } diff --git a/mqtt/CMakeLists.txt b/mqtt/CMakeLists.txt new file mode 100644 index 0000000..7273019 --- /dev/null +++ b/mqtt/CMakeLists.txt @@ -0,0 +1,13 @@ +# ebus_mqtt: the MQTT transport interface and the Homie-agnostic client behavior (the +# disconnected-link hold and the reconnect order). No dependency beyond the C and C++ +# standard libraries, and only its own include directory. See doc/mqtt.md. +# The top-level CMakeLists.txt adds this directory; a project that wants only ebus_mqtt +# can add_subdirectory(/cpp-sdk/mqtt) instead. +cmake_minimum_required(VERSION 3.18) +project(ebus_mqtt LANGUAGES CXX) + +add_library(ebus_mqtt STATIC + src/publish_hold.cpp +) +target_include_directories(ebus_mqtt PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include) +target_compile_features(ebus_mqtt PUBLIC cxx_std_17) diff --git a/ports/posix/include/ebus_posix/publish_hold.h b/mqtt/include/ebus/mqtt/publish_hold.h similarity index 56% rename from ports/posix/include/ebus_posix/publish_hold.h rename to mqtt/include/ebus/mqtt/publish_hold.h index 7f5e208..a60af6a 100644 --- a/ports/posix/include/ebus_posix/publish_hold.h +++ b/mqtt/include/ebus/mqtt/publish_hold.h @@ -7,23 +7,36 @@ // - not retained, QoS 1 or 2: held in order, one entry per publish. // - not retained, QoS 0: dropped. // -// Bounded: when full, the oldest entry is evicted. Fixed storage, no allocation. Not -// thread-safe: the thread that owns the MQTT client is the only one that touches it. -#include +// Bounded: when full, the oldest entry is evicted. Fixed storage in the object, no +// allocation; declare it static, not on a stack. Not thread-safe: only the task that owns +// the MQTT client touches it. +// +// The sizes below set the object's layout, so a -D that overrides one must reach every +// translation unit that includes this header (PlatformIO build_flags, or a PUBLIC +// compile definition on the ebus_mqtt CMake target), not only this library's sources. #include -#ifndef EBUS_POSIX_HOLD_ENTRIES -#define EBUS_POSIX_HOLD_ENTRIES 64 +#ifndef EBUS_MQTT_HOLD_ENTRIES +#define EBUS_MQTT_HOLD_ENTRIES 64 +#endif +// Longest topic held, in characters. A Homie topic is at most HOMIE_TOPIC_MAX (127). +#ifndef EBUS_MQTT_HOLD_TOPIC_MAX +#define EBUS_MQTT_HOLD_TOPIC_MAX 127 #endif -// Longest payload held. Property values are at most Property::VALUE_MAX (256); a longer -// payload (a $description) published while the link is down is dropped and logged. -#ifndef EBUS_POSIX_HOLD_PAYLOAD_MAX -#define EBUS_POSIX_HOLD_PAYLOAD_MAX 1024 +// Longest payload held, in bytes. A longer one (a Homie $description) is dropped. +#ifndef EBUS_MQTT_HOLD_PAYLOAD_MAX +#define EBUS_MQTT_HOLD_PAYLOAD_MAX 1024 #endif class PublishHold { public: - enum Result { HELD, REPLACED, DROPPED_QOS0, DROPPED_TOO_LARGE }; + enum Result { + HELD, + REPLACED, // a held retained value for the same topic was replaced + EVICTED_OLDEST, // held, after evicting the oldest entry to make room + DROPPED_QOS0, + DROPPED_TOO_LARGE // topic or payload over the limits above + }; // Sends one held publish on a live link. Returns true when the entry is done with // (sent, or refused for good), false to keep it held and stop the flush. @@ -48,13 +61,13 @@ class PublishHold { int qos = 0; int length = 0; uint32_t seq = 0; - char topic[HOMIE_TOPIC_MAX + 1] = {0}; - char payload[EBUS_POSIX_HOLD_PAYLOAD_MAX] = {0}; + char topic[EBUS_MQTT_HOLD_TOPIC_MAX + 1] = {0}; + char payload[EBUS_MQTT_HOLD_PAYLOAD_MAX] = {0}; }; int oldest() const; - Entry _entries[EBUS_POSIX_HOLD_ENTRIES]; + Entry _entries[EBUS_MQTT_HOLD_ENTRIES]; int _count = 0; uint32_t _next_seq = 0; uint32_t _evicted = 0; diff --git a/ports/posix/src/publish_hold.cpp b/mqtt/src/publish_hold.cpp similarity index 69% rename from ports/posix/src/publish_hold.cpp rename to mqtt/src/publish_hold.cpp index 197fc29..7571f52 100644 --- a/ports/posix/src/publish_hold.cpp +++ b/mqtt/src/publish_hold.cpp @@ -1,11 +1,10 @@ // SPDX-License-Identifier: MIT -#include -#include +#include #include int PublishHold::oldest() const { int best = -1; - for (int i = 0; i < EBUS_POSIX_HOLD_ENTRIES; i++) { + for (int i = 0; i < EBUS_MQTT_HOLD_ENTRIES; i++) { if (!_entries[i].used) continue; // Sequence numbers are compared as a difference, so a wrap keeps the order. if (best < 0 || (int32_t)(_entries[i].seq - _entries[best].seq) < 0) best = i; @@ -18,16 +17,14 @@ PublishHold::Result PublishHold::hold(const char* topic, const char* payload, in if (!retained && qos == 0) return DROPPED_QOS0; if (length < 0) length = 0; size_t topic_len = strlen(topic); - if (length > EBUS_POSIX_HOLD_PAYLOAD_MAX || topic_len > HOMIE_TOPIC_MAX) { - homie_logf("HOLD: %d-byte publish to '%s' is too large to hold; dropped\n", length, - topic); + if (length > EBUS_MQTT_HOLD_PAYLOAD_MAX || topic_len > EBUS_MQTT_HOLD_TOPIC_MAX) { return DROPPED_TOO_LARGE; } int slot = -1; Result result = HELD; if (retained) { - for (int i = 0; i < EBUS_POSIX_HOLD_ENTRIES; i++) { + for (int i = 0; i < EBUS_MQTT_HOLD_ENTRIES; i++) { if (_entries[i].used && _entries[i].retained && strcmp(_entries[i].topic, topic) == 0) { slot = i; @@ -37,20 +34,16 @@ PublishHold::Result PublishHold::hold(const char* topic, const char* payload, in } } if (slot < 0) { - for (int i = 0; i < EBUS_POSIX_HOLD_ENTRIES; i++) { + for (int i = 0; i < EBUS_MQTT_HOLD_ENTRIES; i++) { if (!_entries[i].used) { slot = i; break; } } } if (slot < 0) { slot = oldest(); _evicted++; - if (_evicted == 1 || _evicted % 100 == 0) { - homie_logf("HOLD: full (%d entries), evicted the oldest ('%s'); %u evicted so " - "far\n", EBUS_POSIX_HOLD_ENTRIES, _entries[slot].topic, - (unsigned)_evicted); - } _entries[slot].used = false; _count--; + result = EVICTED_OLDEST; } Entry& e = _entries[slot]; @@ -78,6 +71,6 @@ int PublishHold::flush(send_fn send, void* ctx) { } void PublishHold::clear() { - for (int i = 0; i < EBUS_POSIX_HOLD_ENTRIES; i++) _entries[i].used = false; + for (int i = 0; i < EBUS_MQTT_HOLD_ENTRIES; i++) _entries[i].used = false; _count = 0; } diff --git a/ports/posix/CMakeLists.txt b/ports/posix/CMakeLists.txt index 4cd92de..a964036 100644 --- a/ports/posix/CMakeLists.txt +++ b/ports/posix/CMakeLists.txt @@ -65,7 +65,6 @@ endfunction() add_library(ebus_posix STATIC src/paho_transport.cpp src/posix_port.cpp - src/publish_hold.cpp ) target_include_directories(ebus_posix PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include) target_link_libraries(ebus_posix PUBLIC ebus_core Threads::Threads PRIVATE ${_paho_target}) @@ -82,11 +81,6 @@ ebus_posix_warnings(ebus-posix-controller) if(EBUS_POSIX_TESTS) enable_testing() - add_executable(test_publish_hold test/test_publish_hold.cpp) - target_link_libraries(test_publish_hold PRIVATE ebus_posix) - ebus_posix_warnings(test_publish_hold) - add_test(NAME publish_hold COMMAND test_publish_hold) - # The end-to-end tests start their own mosquitto on a free localhost port. find_program(EBUS_POSIX_MOSQUITTO mosquitto PATHS /opt/homebrew/sbin /usr/local/sbin /usr/sbin /sbin) diff --git a/ports/posix/include/ebus_posix/paho_transport.h b/ports/posix/include/ebus_posix/paho_transport.h index 562522c..1e787bb 100644 --- a/ports/posix/include/ebus_posix/paho_transport.h +++ b/ports/posix/include/ebus_posix/paho_transport.h @@ -12,7 +12,7 @@ // inbound arena, the queue_publish() ring and the disconnected-link hold. Declare the // transport static, not on a stack. Paho allocates per in-flight message internally, // bounded by its in-flight window. -#include +#include #include #include #include @@ -20,6 +20,9 @@ #include #include +static_assert(EBUS_MQTT_HOLD_TOPIC_MAX >= HOMIE_TOPIC_MAX, + "the publish hold must take every Homie topic"); + #ifndef EBUS_POSIX_INBOX_BYTES #define EBUS_POSIX_INBOX_BYTES (64 * 1024) #endif diff --git a/ports/posix/src/paho_transport.cpp b/ports/posix/src/paho_transport.cpp index 8919bbe..5cf5624 100644 --- a/ports/posix/src/paho_transport.cpp +++ b/ports/posix/src/paho_transport.cpp @@ -124,7 +124,20 @@ bool PahoTransport::send_held(void* ctx, const char* topic, const char* payload, void PahoTransport::hold_or_drop(const char* topic, const char* payload, int length, bool retained, int qos) { if (_stopped) return; - _hold.hold(topic, payload, length, retained, qos); + switch (_hold.hold(topic, payload, length, retained, qos)) { + case PublishHold::DROPPED_TOO_LARGE: + homie_logf("PAHO: %d-byte publish to '%s' is too large to hold; dropped\n", length, + topic); + break; + case PublishHold::EVICTED_OLDEST: + if (_hold.evicted() == 1 || _hold.evicted() % 100 == 0) { + homie_logf("PAHO: hold full (%d entries), evicted the oldest; %u evicted so " + "far\n", EBUS_MQTT_HOLD_ENTRIES, (unsigned)_hold.evicted()); + } + break; + default: + break; + } } bool PahoTransport::publish(const char* topic, const char* payload, int length, diff --git a/ports/posix/test/test_publish_hold.cpp b/ports/posix/test/test_publish_hold.cpp deleted file mode 100644 index 4c3a64e..0000000 --- a/ports/posix/test/test_publish_hold.cpp +++ /dev/null @@ -1,122 +0,0 @@ -// SPDX-License-Identifier: MIT -// The disconnected-link rules (ebus_posix/publish_hold.h), without a broker. -#include -#include -#include -#include - -static int failures = 0; -#define CHECK(cond) \ - do { \ - if (!(cond)) { \ - fprintf(stderr, "%s:%d: FAILED: %s\n", __FILE__, __LINE__, #cond); \ - failures++; \ - } \ - } while (0) - -struct Sent { - std::string topic, payload; - bool retained; - int qos; -}; - -struct Sink { - std::vector sent; - int fail_after = -1; // refuse the send after this many - static bool send(void* ctx, const char* topic, const char* payload, int length, - bool retained, int qos) { - Sink* s = (Sink*)ctx; - if (s->fail_after >= 0 && (int)s->sent.size() >= s->fail_after) return false; - s->sent.push_back({topic, std::string(payload, (size_t)length), retained, qos}); - return true; - } -}; - -static PublishHold hold; // large: keep it off the stack - -static void test_rules() { - hold.clear(); - CHECK(hold.hold("a/v", "1", 1, true, 2) == PublishHold::HELD); - CHECK(hold.hold("e/x", "ev1", 3, false, 1) == PublishHold::HELD); - CHECK(hold.hold("e/q0", "gone", 4, false, 0) == PublishHold::DROPPED_QOS0); - CHECK(hold.hold("b/v", "x", 1, true, 1) == PublishHold::HELD); - CHECK(hold.hold("e/x", "ev2", 3, false, 2) == PublishHold::HELD); // events never merge - CHECK(hold.hold("a/v", "2", 1, true, 2) == PublishHold::REPLACED); // newest per topic - CHECK(hold.count() == 4); - - Sink sink; - CHECK(hold.flush(Sink::send, &sink) == 4); - CHECK(hold.count() == 0); - // The replaced retained value moved to the back. - const char* order[] = {"e/x:ev1", "b/v:x", "e/x:ev2", "a/v:2"}; - CHECK(sink.sent.size() == 4); - for (size_t i = 0; i < sink.sent.size() && i < 4; i++) { - CHECK(sink.sent[i].topic + ":" + sink.sent[i].payload == order[i]); - } - CHECK(sink.sent[3].retained && sink.sent[3].qos == 2); -} - -static void test_empty_string_value_kept() { - hold.clear(); - const char nul = 0; - CHECK(hold.hold("a/s", &nul, 1, true, 2) == PublishHold::HELD); - Sink sink; - hold.flush(Sink::send, &sink); - CHECK(sink.sent.size() == 1 && sink.sent[0].payload.size() == 1 && - sink.sent[0].payload[0] == 0); -} - -static void test_bounded_evicts_oldest() { - hold.clear(); - uint32_t evicted = hold.evicted(); - char topic[32]; - for (int i = 0; i < EBUS_POSIX_HOLD_ENTRIES + 3; i++) { - snprintf(topic, sizeof(topic), "t/%d", i); - hold.hold(topic, "v", 1, true, 2); - } - CHECK(hold.count() == EBUS_POSIX_HOLD_ENTRIES); - CHECK(hold.evicted() - evicted == 3); - Sink sink; - hold.flush(Sink::send, &sink); - CHECK(!sink.sent.empty() && sink.sent.front().topic == "t/3"); - snprintf(topic, sizeof(topic), "t/%d", EBUS_POSIX_HOLD_ENTRIES + 2); - CHECK(!sink.sent.empty() && sink.sent.back().topic == topic); -} - -static void test_too_large_dropped() { - hold.clear(); - static char big[EBUS_POSIX_HOLD_PAYLOAD_MAX + 1]; - CHECK(hold.hold("d/$description", big, sizeof(big), true, 2) == - PublishHold::DROPPED_TOO_LARGE); - CHECK(hold.count() == 0); -} - -static void test_interrupted_flush_keeps_the_rest() { - hold.clear(); - hold.hold("a", "1", 1, true, 2); - hold.hold("b", "2", 1, true, 2); - hold.hold("c", "3", 1, true, 2); - Sink sink; - sink.fail_after = 1; - CHECK(hold.flush(Sink::send, &sink) == 1); - CHECK(hold.count() == 2); - // A newer value for a still-held topic replaces it. - CHECK(hold.hold("c", "4", 1, true, 2) == PublishHold::REPLACED); - Sink again; - CHECK(hold.flush(Sink::send, &again) == 2); - CHECK(again.sent.size() == 2 && again.sent[0].topic == "b" && again.sent[1].payload == "4"); -} - -int main() { - test_rules(); - test_empty_string_value_kept(); - test_bounded_evicts_oldest(); - test_too_large_dropped(); - test_interrupted_flush_keeps_the_rest(); - if (failures) { - fprintf(stderr, "%d check(s) failed\n", failures); - return 1; - } - printf("publish_hold: all checks passed\n"); - return 0; -} diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index c6d1836..2b5d170 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -1,5 +1,6 @@ # The core's Unity suites, built and run without PlatformIO. Each test_/ directory is # one executable and one CTest test; the suites include only core headers and support/. +# A test_mqtt_/ suite links ebus_mqtt alone, so it cannot reach a Homie header. # Unity 2.6.1, the release PlatformIO's `test_framework = unity` resolves to, so a suite # behaves the same under `pio test` in esp32-sdk and under CTest here. To build offline, @@ -24,7 +25,11 @@ foreach(_suite IN LISTS _ebus_core_suites) endif() file(GLOB _sources CONFIGURE_DEPENDS ${CMAKE_CURRENT_SOURCE_DIR}/${_suite}/*.cpp) add_executable(${_suite} ${_sources}) - target_link_libraries(${_suite} PRIVATE ebus_core unity) + if(_suite MATCHES "^test_mqtt_") + target_link_libraries(${_suite} PRIVATE ebus_mqtt unity) + else() + target_link_libraries(${_suite} PRIVATE ebus_core unity) + endif() if(CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang") target_compile_options(${_suite} PRIVATE -Wall -Wextra) endif() diff --git a/test/test_mqtt_publish_hold/test_mqtt_publish_hold.cpp b/test/test_mqtt_publish_hold/test_mqtt_publish_hold.cpp new file mode 100644 index 0000000..eee6151 --- /dev/null +++ b/test/test_mqtt_publish_hold/test_mqtt_publish_hold.cpp @@ -0,0 +1,144 @@ +// SPDX-License-Identifier: MIT +// The disconnected-link rules (ebus/mqtt/publish_hold.h), without a broker. Builds against +// ebus_mqtt alone. +#include +#include +#include +#include + +struct Sent { + char topic[EBUS_MQTT_HOLD_TOPIC_MAX + 1]; + char payload[16]; + int length; + bool retained; + int qos; +}; + +struct Sink { + static const int MAX = EBUS_MQTT_HOLD_ENTRIES + 8; + Sent sent[MAX]; + int count = 0; + int fail_after = -1; // refuse the send after this many + + static bool send(void* ctx, const char* topic, const char* payload, int length, + bool retained, int qos) { + Sink* s = (Sink*)ctx; + if (s->fail_after >= 0 && s->count >= s->fail_after) return false; + if (s->count >= MAX) return false; + Sent& e = s->sent[s->count++]; + snprintf(e.topic, sizeof(e.topic), "%s", topic); + int n = length < (int)sizeof(e.payload) ? length : (int)sizeof(e.payload); + memcpy(e.payload, payload, (size_t)n); + e.length = length; + e.retained = retained; + e.qos = qos; + return true; + } +}; + +static PublishHold hold; // large: keep it off the stack +static Sink sink; + +void setUp(void) { + hold.clear(); + sink = Sink(); +} +void tearDown(void) {} + +static void assert_sent(int i, const char* topic, const char* payload) { + TEST_ASSERT_EQUAL_STRING(topic, sink.sent[i].topic); + TEST_ASSERT_EQUAL_INT((int)strlen(payload), sink.sent[i].length); + TEST_ASSERT_EQUAL_MEMORY(payload, sink.sent[i].payload, strlen(payload)); +} + +static void test_rules(void) { + TEST_ASSERT_EQUAL_INT(PublishHold::HELD, hold.hold("a/v", "1", 1, true, 2)); + TEST_ASSERT_EQUAL_INT(PublishHold::HELD, hold.hold("e/x", "ev1", 3, false, 1)); + TEST_ASSERT_EQUAL_INT(PublishHold::DROPPED_QOS0, hold.hold("e/q0", "gone", 4, false, 0)); + TEST_ASSERT_EQUAL_INT(PublishHold::HELD, hold.hold("b/v", "x", 1, true, 1)); + // Events never merge; a retained value is newest per topic. + TEST_ASSERT_EQUAL_INT(PublishHold::HELD, hold.hold("e/x", "ev2", 3, false, 2)); + TEST_ASSERT_EQUAL_INT(PublishHold::REPLACED, hold.hold("a/v", "2", 1, true, 2)); + TEST_ASSERT_EQUAL_INT(4, hold.count()); + + TEST_ASSERT_EQUAL_INT(4, hold.flush(Sink::send, &sink)); + TEST_ASSERT_EQUAL_INT(0, hold.count()); + // The replaced retained value moved to the back. + TEST_ASSERT_EQUAL_INT(4, sink.count); + assert_sent(0, "e/x", "ev1"); + assert_sent(1, "b/v", "x"); + assert_sent(2, "e/x", "ev2"); + assert_sent(3, "a/v", "2"); + TEST_ASSERT_TRUE(sink.sent[3].retained); + TEST_ASSERT_EQUAL_INT(2, sink.sent[3].qos); + TEST_ASSERT_FALSE(sink.sent[0].retained); + TEST_ASSERT_EQUAL_INT(1, sink.sent[0].qos); +} + +static void test_empty_string_value_kept(void) { + const char nul = 0; + TEST_ASSERT_EQUAL_INT(PublishHold::HELD, hold.hold("a/s", &nul, 1, true, 2)); + hold.flush(Sink::send, &sink); + TEST_ASSERT_EQUAL_INT(1, sink.count); + TEST_ASSERT_EQUAL_INT(1, sink.sent[0].length); + TEST_ASSERT_EQUAL_INT8(0, sink.sent[0].payload[0]); +} + +static void test_bounded_evicts_oldest(void) { + uint32_t evicted = hold.evicted(); + char topic[32]; + for (int i = 0; i < EBUS_MQTT_HOLD_ENTRIES; i++) { + snprintf(topic, sizeof(topic), "t/%d", i); + TEST_ASSERT_EQUAL_INT(PublishHold::HELD, hold.hold(topic, "v", 1, true, 2)); + } + for (int i = EBUS_MQTT_HOLD_ENTRIES; i < EBUS_MQTT_HOLD_ENTRIES + 3; i++) { + snprintf(topic, sizeof(topic), "t/%d", i); + TEST_ASSERT_EQUAL_INT(PublishHold::EVICTED_OLDEST, hold.hold(topic, "v", 1, true, 2)); + } + TEST_ASSERT_EQUAL_INT(EBUS_MQTT_HOLD_ENTRIES, hold.count()); + TEST_ASSERT_EQUAL_UINT32(3, hold.evicted() - evicted); + hold.flush(Sink::send, &sink); + TEST_ASSERT_EQUAL_INT(EBUS_MQTT_HOLD_ENTRIES, sink.count); + TEST_ASSERT_EQUAL_STRING("t/3", sink.sent[0].topic); + snprintf(topic, sizeof(topic), "t/%d", EBUS_MQTT_HOLD_ENTRIES + 2); + TEST_ASSERT_EQUAL_STRING(topic, sink.sent[sink.count - 1].topic); +} + +static void test_too_large_dropped(void) { + static char big[EBUS_MQTT_HOLD_PAYLOAD_MAX + 1]; + TEST_ASSERT_EQUAL_INT(PublishHold::DROPPED_TOO_LARGE, + hold.hold("d/$description", big, sizeof(big), true, 2)); + static char long_topic[EBUS_MQTT_HOLD_TOPIC_MAX + 2]; + memset(long_topic, 'x', sizeof(long_topic) - 1); + TEST_ASSERT_EQUAL_INT(PublishHold::DROPPED_TOO_LARGE, + hold.hold(long_topic, "v", 1, true, 2)); + TEST_ASSERT_EQUAL_INT(0, hold.count()); +} + +static void test_interrupted_flush_keeps_the_rest(void) { + hold.hold("a", "1", 1, true, 2); + hold.hold("b", "2", 1, true, 2); + hold.hold("c", "3", 1, true, 2); + sink.fail_after = 1; + TEST_ASSERT_EQUAL_INT(1, hold.flush(Sink::send, &sink)); + TEST_ASSERT_EQUAL_INT(2, hold.count()); + // A newer value for a still-held topic replaces it. + TEST_ASSERT_EQUAL_INT(PublishHold::REPLACED, hold.hold("c", "4", 1, true, 2)); + sink = Sink(); + TEST_ASSERT_EQUAL_INT(2, hold.flush(Sink::send, &sink)); + TEST_ASSERT_EQUAL_INT(2, sink.count); + assert_sent(0, "b", "2"); + assert_sent(1, "c", "4"); +} + +int main(int argc, char** argv) { + (void)argc; + (void)argv; + UNITY_BEGIN(); + RUN_TEST(test_rules); + RUN_TEST(test_empty_string_value_kept); + RUN_TEST(test_bounded_evicts_oldest); + RUN_TEST(test_too_large_dropped); + RUN_TEST(test_interrupted_flush_keeps_the_rest); + return UNITY_END(); +} From 15517bbdffd805c56aa15ca0725dcda918d43a48 Mon Sep 17 00:00:00 2001 From: Donald Clark Jackson Date: Fri, 9 Oct 2026 07:15:59 -0700 Subject: [PATCH 2/5] Add mqtt_after_connect(), the reconnect order, and use it in the POSIX port After every connect a port flushes the hold, re-subscribes, then notifies the application, as ebus-mqtt-client does. ebus/mqtt/reconnect.h documents that contract and mqtt_after_connect() runs it from three hooks, stopping before the later steps when the flush is interrupted or the link drops during the re-subscribe. PahoTransport::try_connect() now calls it; its log lines, which the e2e reconnect test checks, are unchanged. Part of #3 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Npk6WmkpnBRzWkuhK6uCuz --- mqtt/CMakeLists.txt | 1 + mqtt/include/ebus/mqtt/reconnect.h | 36 ++++++++ mqtt/src/reconnect.cpp | 16 ++++ .../posix/include/ebus_posix/paho_transport.h | 4 + ports/posix/src/paho_transport.cpp | 46 ++++++---- .../test_mqtt_reconnect.cpp | 91 +++++++++++++++++++ 6 files changed, 176 insertions(+), 18 deletions(-) create mode 100644 mqtt/include/ebus/mqtt/reconnect.h create mode 100644 mqtt/src/reconnect.cpp create mode 100644 test/test_mqtt_reconnect/test_mqtt_reconnect.cpp diff --git a/mqtt/CMakeLists.txt b/mqtt/CMakeLists.txt index 7273019..2b1b276 100644 --- a/mqtt/CMakeLists.txt +++ b/mqtt/CMakeLists.txt @@ -8,6 +8,7 @@ project(ebus_mqtt LANGUAGES CXX) add_library(ebus_mqtt STATIC src/publish_hold.cpp + src/reconnect.cpp ) target_include_directories(ebus_mqtt PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include) target_compile_features(ebus_mqtt PUBLIC cxx_std_17) diff --git a/mqtt/include/ebus/mqtt/reconnect.h b/mqtt/include/ebus/mqtt/reconnect.h new file mode 100644 index 0000000..669f6f5 --- /dev/null +++ b/mqtt/include/ebus/mqtt/reconnect.h @@ -0,0 +1,36 @@ +#pragma once +// The order of work after every successful connect, ebus-mqtt-client's: +// +// 1. Flush the hold. What was published while the link was down describes state that +// already exists, so it reaches the broker before anything reacts to the connect. +// 2. Subscribe every topic again (a clean session keeps none). +// 3. Notify the application, which may now publish (a Homie device republishes its +// state) and subscribe. +// +// Until step 1 has finished, the port keeps treating the link as down, so a publish made +// meanwhile joins the hold behind the older values instead of overtaking them. +// mqtt_after_connect() runs the three steps on the task that owns the client. +#include + +struct MqttConnectSteps { + // Step 1: send one held publish on the live link, as PublishHold::flush() calls it. + PublishHold::send_fn send; + // Step 2: mark the link usable for publish() and subscribe(), then subscribe every + // topic. Returns false when the link dropped meanwhile. May be null. + bool (*resubscribe)(void* ctx); + // Step 3: `first` is true for the first connect. May be null. + void (*notify)(void* ctx, bool first); + void* ctx; +}; + +struct MqttConnectReport { + int held; // entries in the hold when the connect came up + int flushed; // entries sent by step 1 +}; + +// Runs the three steps in order. Returns true when all three ran. Returns false, without +// running the later steps, when the flush stopped early (the rest stays held) or +// resubscribe() reported the link down; the port then treats the link as down and runs +// this again after its next connect. `hold` may be null for a port that holds nothing. +bool mqtt_after_connect(PublishHold* hold, const MqttConnectSteps& steps, bool first, + MqttConnectReport* report = nullptr); diff --git a/mqtt/src/reconnect.cpp b/mqtt/src/reconnect.cpp new file mode 100644 index 0000000..82a1f78 --- /dev/null +++ b/mqtt/src/reconnect.cpp @@ -0,0 +1,16 @@ +// SPDX-License-Identifier: MIT +#include + +bool mqtt_after_connect(PublishHold* hold, const MqttConnectSteps& steps, bool first, + MqttConnectReport* report) { + int held = hold ? hold->count() : 0; + int flushed = hold ? hold->flush(steps.send, steps.ctx) : 0; + if (report) { + report->held = held; + report->flushed = flushed; + } + if (flushed < held) return false; + if (steps.resubscribe && !steps.resubscribe(steps.ctx)) return false; + if (steps.notify) steps.notify(steps.ctx, first); + return true; +} diff --git a/ports/posix/include/ebus_posix/paho_transport.h b/ports/posix/include/ebus_posix/paho_transport.h index 1e787bb..ae35b82 100644 --- a/ports/posix/include/ebus_posix/paho_transport.h +++ b/ports/posix/include/ebus_posix/paho_transport.h @@ -93,8 +93,11 @@ class PahoTransport final : public HomieTransport { }; friend struct PahoCallbacks; // Paho's callbacks, in paho_transport.cpp + // The steps of mqtt_after_connect(), in paho_transport.cpp. static bool send_held(void* ctx, const char* topic, const char* payload, int length, bool retained, int qos); + static bool resubscribe_all(void* ctx); + static void notify_connected(void* ctx, bool first); bool try_connect(); bool send(const char* topic, const char* payload, int length, bool retained, int qos); @@ -123,6 +126,7 @@ class PahoTransport final : public HomieTransport { bool _was_up = false; bool _stopped = false; bool _attempted = false; + int _held_at_connect = 0; uint32_t _next_attempt_ms = 0; int _last_error = 0; diff --git a/ports/posix/src/paho_transport.cpp b/ports/posix/src/paho_transport.cpp index 5cf5624..63533dd 100644 --- a/ports/posix/src/paho_transport.cpp +++ b/ports/posix/src/paho_transport.cpp @@ -1,5 +1,6 @@ // SPDX-License-Identifier: MIT #include +#include #include #include #include @@ -164,8 +165,8 @@ bool PahoTransport::connected() { return _client && _link_ready.load() && MQTTClient_isConnected(CLIENT(_client)); } -// One connect attempt. On success: flush the hold, re-subscribe every settable topic, -// then tell the application, in that order (ebus-mqtt-client's reconnect order). +// One connect attempt. On success, mqtt_after_connect() flushes the hold, re-subscribes +// every settable topic, then tells the application, in that order. bool PahoTransport::try_connect() { MQTTClient c = CLIENT(_client); if (MQTTClient_isConnected(c)) MQTTClient_disconnect(c, 0); // a half-up session @@ -199,33 +200,42 @@ bool PahoTransport::try_connect() { } if (_will_topic[0]) homie_logf("PAHO: will '%s' = '%s'\n", _will_topic, _will_payload); - int held = _hold.count(); - int sent = _hold.flush(send_held, this); - homie_logf("PAHO: flushed %d of %d held publishes (%u evicted while down)\n", sent, held, - (unsigned)_hold.evicted()); - if (sent < held) { - homie_logln("PAHO: link dropped during the flush; the rest stays held"); - return false; + MqttConnectSteps steps = {send_held, resubscribe_all, notify_connected, this}; + MqttConnectReport report; + _held_at_connect = _hold.count(); + bool first = !_ever_connected; + bool ready = mqtt_after_connect(&_hold, steps, first, &report); + if (report.flushed < report.held) { + homie_logf("PAHO: flushed %d of %d held publishes; the link dropped during the " + "flush, the rest stays held\n", report.flushed, report.held); } + if (ready) _ever_connected = true; + return ready; +} - // Subscribes are made with the link marked up, so subscribe() is usable from here on. - _link_ready.store(true); +bool PahoTransport::resubscribe_all(void* ctx) { + PahoTransport* self = (PahoTransport*)ctx; + homie_logf("PAHO: flushed %d of %d held publishes (%u evicted while down)\n", + self->_held_at_connect, self->_held_at_connect, (unsigned)self->_hold.evicted()); + // From here publish() and subscribe() go to the wire. + self->_link_ready.store(true); int subscribed = 0; for (int i = 0; i < settable_count(); i++) { - if (subscribe(settable_topic(i), 0)) { // /set is QoS 0 + if (self->subscribe(settable_topic(i), 0)) { // /set is QoS 0 subscribed++; } else { - homie_logf("PAHO: subscribe '%s' failed (%d)\n", settable_topic(i), _last_error); + homie_logf("PAHO: subscribe '%s' failed (%d)\n", settable_topic(i), + self->_last_error); } } homie_logf("PAHO: resubscribed %d of %d settable topics\n", subscribed, settable_count()); - if (!_link_ready.load()) return false; + return self->_link_ready.load(); +} - bool first = !_ever_connected; - _ever_connected = true; +void PahoTransport::notify_connected(void* ctx, bool first) { + PahoTransport* self = (PahoTransport*)ctx; homie_logf("PAHO: connected (%s)\n", first ? "first" : "reconnect"); - if (_connected_fn) _connected_fn(_connected_ctx, first); - return true; + if (self->_connected_fn) self->_connected_fn(self->_connected_ctx, first); } void PahoTransport::drain_inbound() { diff --git a/test/test_mqtt_reconnect/test_mqtt_reconnect.cpp b/test/test_mqtt_reconnect/test_mqtt_reconnect.cpp new file mode 100644 index 0000000..a3de70e --- /dev/null +++ b/test/test_mqtt_reconnect/test_mqtt_reconnect.cpp @@ -0,0 +1,91 @@ +// SPDX-License-Identifier: MIT +// The reconnect order (ebus/mqtt/reconnect.h): flush, re-subscribe, notify. Builds against +// ebus_mqtt alone. +#include +#include +#include + +// Records each step as one letter: 'p' per publish sent, 's' for the re-subscribe, +// 'n' (first connect) or 'r' (later) for the notify. +struct Port { + char steps[32]; + int n = 0; + int send_ok = 100; // sends that succeed before the link drops + bool resubscribe_ok = true; + + void add(char c) { + if (n < (int)sizeof(steps) - 1) steps[n++] = c; + steps[n] = '\0'; + } + static bool send(void* ctx, const char*, const char*, int, bool, int) { + Port* p = (Port*)ctx; + if (p->send_ok-- <= 0) return false; + p->add('p'); + return true; + } + static bool resubscribe(void* ctx) { + Port* p = (Port*)ctx; + p->add('s'); + return p->resubscribe_ok; + } + static void notify(void* ctx, bool first) { ((Port*)ctx)->add(first ? 'n' : 'r'); } + + MqttConnectSteps hooks() { return MqttConnectSteps{send, resubscribe, notify, this}; } +}; + +static PublishHold hold; +static Port port; + +void setUp(void) { + hold.clear(); + port = Port(); + port.steps[0] = '\0'; +} +void tearDown(void) {} + +static void test_flush_then_resubscribe_then_notify(void) { + hold.hold("a", "1", 1, true, 2); + hold.hold("b", "2", 1, true, 2); + MqttConnectReport report; + TEST_ASSERT_TRUE(mqtt_after_connect(&hold, port.hooks(), true, &report)); + TEST_ASSERT_EQUAL_STRING("ppsn", port.steps); + TEST_ASSERT_EQUAL_INT(2, report.held); + TEST_ASSERT_EQUAL_INT(2, report.flushed); + TEST_ASSERT_EQUAL_INT(0, hold.count()); +} + +static void test_interrupted_flush_stops_before_resubscribe(void) { + hold.hold("a", "1", 1, true, 2); + hold.hold("b", "2", 1, true, 2); + port.send_ok = 1; + MqttConnectReport report; + TEST_ASSERT_FALSE(mqtt_after_connect(&hold, port.hooks(), false, &report)); + TEST_ASSERT_EQUAL_STRING("p", port.steps); + TEST_ASSERT_EQUAL_INT(2, report.held); + TEST_ASSERT_EQUAL_INT(1, report.flushed); + TEST_ASSERT_EQUAL_INT(1, hold.count()); +} + +static void test_link_lost_during_resubscribe_skips_notify(void) { + port.resubscribe_ok = false; + TEST_ASSERT_FALSE(mqtt_after_connect(&hold, port.hooks(), false)); + TEST_ASSERT_EQUAL_STRING("s", port.steps); +} + +static void test_no_hold_and_optional_steps(void) { + TEST_ASSERT_TRUE(mqtt_after_connect(nullptr, port.hooks(), false)); + TEST_ASSERT_EQUAL_STRING("sr", port.steps); + MqttConnectSteps bare = {Port::send, nullptr, nullptr, &port}; + TEST_ASSERT_TRUE(mqtt_after_connect(&hold, bare, true)); +} + +int main(int argc, char** argv) { + (void)argc; + (void)argv; + UNITY_BEGIN(); + RUN_TEST(test_flush_then_resubscribe_then_notify); + RUN_TEST(test_interrupted_flush_stops_before_resubscribe); + RUN_TEST(test_link_lost_during_resubscribe_skips_notify); + RUN_TEST(test_no_hold_and_optional_steps); + return UNITY_END(); +} From 57a56c1cd45af3649ea1a9a279a3e0860d74b585 Mon Sep 17 00:00:00 2001 From: Donald Clark Jackson Date: Fri, 9 Oct 2026 07:17:43 -0700 Subject: [PATCH 3/5] Move the transport interface into ebus_mqtt as MqttTransport ebus/mqtt/transport.h holds the Homie-agnostic interface: publish(), subscribe(), connected(), last_error(), and a queue_publish() that takes an explicit QoS and a completion callback (mqtt_publish_done_fn plus a void* context) in place of Homie's Property*. HomieTransport, in homie/homie_transport.h as before, derives from it and keeps the Property overload as the Homie adapter: by default it forwards to the generic overload with a callback that calls Property::queued_publish_done(), so the publish-on-change memo works over either. A port overrides one of the two; a port written against 0.1.0 (esp32-sdk's MqttClientTransport overrides the Property overload) builds and behaves as before. MAX_DATA_LEN, mqtt_qos and homie_qos() stay on the Homie side. The POSIX port and the test FakeTransport now implement the generic overload. New tests cover a transport that overrides only the Property overload and one that overrides neither. Part of #3 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Npk6WmkpnBRzWkuhK6uCuz --- include/homie/homie_transport.h | 47 ++++++------ mqtt/include/ebus/mqtt/transport.h | 42 +++++++++++ .../posix/include/ebus_posix/paho_transport.h | 7 +- ports/posix/src/paho_transport.cpp | 14 ++-- src/homie/homie_transport.cpp | 23 ++++++ test/support/fake_transport.h | 23 +++--- .../test_property_publish.cpp | 72 +++++++++++++++++-- 7 files changed, 180 insertions(+), 48 deletions(-) create mode 100644 mqtt/include/ebus/mqtt/transport.h diff --git a/include/homie/homie_transport.h b/include/homie/homie_transport.h index 81e679a..28187e4 100644 --- a/include/homie/homie_transport.h +++ b/include/homie/homie_transport.h @@ -1,9 +1,10 @@ #pragma once -// The MQTT client as the Homie layer sees it. Device, Node, Property, the settable table -// and the controller publish and subscribe only through this interface; a port +// The MQTT client as the Homie layer sees it: ebus_mqtt's MqttTransport plus the Homie +// queued publish that reports back to a Property. Device, Node, Property, the settable +// table and the controller publish and subscribe only through this interface; a port // implements it over its own MQTT client (src/platform/mqtt_client.h on the ESP32). +#include #include -#include class Property; @@ -14,33 +15,25 @@ class Property; #define MAX_DATA_LEN 8192 #endif -class HomieTransport { +// A port overrides one of the two queue_publish() overloads: +// - MqttTransport's, with a completion callback (a new port), or +// - the Property one below (a port written before ebus_mqtt). +// Each default forwards to, or stands in for, the other. Property calls the Property one. +class HomieTransport : public MqttTransport { public: - // Publish exactly `length` bytes (a single 0x00 byte is the Homie empty-string value; - // a zero-length retained payload deletes the topic). Main task only. Returns false - // when the client is not connected or the publish failed. - virtual bool publish(const char* topic, const char* payload, int length, bool retained, - int qos) = 0; - // NUL-terminated payload. - bool publish(const char* topic, const char* payload, bool retained, int qos) { - return publish(topic, payload, (int)strlen(payload), retained, qos); - } + using MqttTransport::queue_publish; - // Publish from ANY task: the port hands the message to the task that owns the client. - // QoS follows homie_qos(retained). When `source` is not null the port calls - // source->queued_publish_done(payload, length, sent) exactly once, when the message - // is sent or dropped, including when this call itself fails. - virtual bool queue_publish(const char* topic, const char* payload, int length, - bool retained, Property* source) = 0; - - // Subscribe one topic filter. Main task only, and never from inside the receive - // callback: a client that waits for the SUBACK delivers other messages meanwhile. - virtual bool subscribe(const char* topic, int qos) = 0; + // Default for a port that overrides only the Property overload: no queue, so the + // message is refused and `done` is told so. + bool queue_publish(const char* topic, const char* payload, int length, bool retained, + int qos, mqtt_publish_done_fn done, void* ctx) override; - virtual bool connected() = 0; - - // The client's error code for the last failed operation, for logs only. - virtual int last_error() = 0; + // Publish from ANY task at homie_qos(retained). When `source` is not null, + // source->queued_publish_done(payload, length, sent) is called exactly once, when the + // message is sent or dropped, including when this call itself fails. The default + // forwards to the overload above with a callback that does that. + virtual bool queue_publish(const char* topic, const char* payload, int length, + bool retained, Property* source); protected: ~HomieTransport() {} diff --git a/mqtt/include/ebus/mqtt/transport.h b/mqtt/include/ebus/mqtt/transport.h new file mode 100644 index 0000000..94191ef --- /dev/null +++ b/mqtt/include/ebus/mqtt/transport.h @@ -0,0 +1,42 @@ +#pragma once +// The MQTT client as a library sees it: publish, a publish that is safe from any task, +// subscribe. A port implements it over its own MQTT client. +#include + +// Reports what became of a queue_publish(): called exactly once per call, with the +// payload that was queued, when the message is sent (sent=true) or dropped. +typedef void (*mqtt_publish_done_fn)(void* ctx, const char* payload, int length, bool sent); + +class MqttTransport { + public: + // Publish exactly `length` bytes (a zero-length retained payload deletes the topic). + // Only on the task that owns the client. Returns false when the client is not + // connected or the publish failed. + virtual bool publish(const char* topic, const char* payload, int length, bool retained, + int qos) = 0; + // NUL-terminated payload. + bool publish(const char* topic, const char* payload, bool retained, int qos) { + return publish(topic, payload, (int)strlen(payload), retained, qos); + } + + // Publish from ANY task: the port copies the message and hands it to the task that + // owns the client. When `done` is not null the port calls done(ctx, payload, length, + // sent) exactly once, on any task, when the message is sent or dropped, including + // when this call itself fails. + virtual bool queue_publish(const char* topic, const char* payload, int length, + bool retained, int qos, mqtt_publish_done_fn done, + void* ctx) = 0; + + // Subscribe one topic filter. Only on the task that owns the client, and never from + // inside the receive callback: a client that waits for the SUBACK delivers other + // messages meanwhile. + virtual bool subscribe(const char* topic, int qos) = 0; + + virtual bool connected() = 0; + + // The client's error code for the last failed operation, for logs only. + virtual int last_error() = 0; + + protected: + ~MqttTransport() {} +}; diff --git a/ports/posix/include/ebus_posix/paho_transport.h b/ports/posix/include/ebus_posix/paho_transport.h index ae35b82..54da0bd 100644 --- a/ports/posix/include/ebus_posix/paho_transport.h +++ b/ports/posix/include/ebus_posix/paho_transport.h @@ -72,10 +72,11 @@ class PahoTransport final : public HomieTransport { // HomieTransport. publish() and subscribe(): loop() thread only. While the link is // down publish() holds or drops the message (see PublishHold) and returns false. using HomieTransport::publish; + using HomieTransport::queue_publish; bool publish(const char* topic, const char* payload, int length, bool retained, int qos) override; bool queue_publish(const char* topic, const char* payload, int length, bool retained, - Property* source) override; + int qos, mqtt_publish_done_fn done, void* ctx) override; bool subscribe(const char* topic, int qos) override; bool connected() override; int last_error() override { return _last_error; } @@ -89,7 +90,9 @@ class PahoTransport final : public HomieTransport { char payload[Property::VALUE_MAX]; int length; bool retained; - Property* source; + int qos; + mqtt_publish_done_fn done; + void* ctx; }; friend struct PahoCallbacks; // Paho's callbacks, in paho_transport.cpp diff --git a/ports/posix/src/paho_transport.cpp b/ports/posix/src/paho_transport.cpp index 63533dd..e0db726 100644 --- a/ports/posix/src/paho_transport.cpp +++ b/ports/posix/src/paho_transport.cpp @@ -274,14 +274,14 @@ void PahoTransport::drain_queue() { _queue_head = (_queue_head + 1) % EBUS_POSIX_PUBLISH_QUEUE; _queue_count--; } - bool sent = publish(item.topic, item.payload, item.length, item.retained, - homie_qos(item.retained)); - if (item.source) item.source->queued_publish_done(item.payload, item.length, sent); + bool sent = publish(item.topic, item.payload, item.length, item.retained, item.qos); + if (item.done) item.done(item.ctx, item.payload, item.length, sent); } } bool PahoTransport::queue_publish(const char* topic, const char* payload, int length, - bool retained, Property* source) { + bool retained, int qos, mqtt_publish_done_fn done, + void* ctx) { if (length < 0) length = 0; size_t topic_len = strlen(topic); bool ok = false; @@ -299,7 +299,9 @@ bool PahoTransport::queue_publish(const char* topic, const char* payload, int le if (length > 0) memcpy(q.payload, payload, (size_t)length); q.length = length; q.retained = retained; - q.source = source; + q.qos = qos; + q.done = done; + q.ctx = ctx; _queue_count++; ok = true; } @@ -310,7 +312,7 @@ bool PahoTransport::queue_publish(const char* topic, const char* payload, int le homie_logf("PAHO: publish queue full, '%s' dropped (%u dropped)\n", topic, (unsigned)n); } - if (source) source->queued_publish_done(payload, length, false); + if (done) done(ctx, payload, length, false); } return ok; } diff --git a/src/homie/homie_transport.cpp b/src/homie/homie_transport.cpp index c8bb0d1..335b5f3 100644 --- a/src/homie/homie_transport.cpp +++ b/src/homie/homie_transport.cpp @@ -1,3 +1,26 @@ #include +#include +#include uint8_t mqtt_qos = 2; // QoS for retained Homie publications; Homie 5 recommends 2 (C4) + +static void property_publish_done(void* ctx, const char* payload, int length, bool sent) { + ((Property*)ctx)->queued_publish_done(payload, length, sent); +} + +bool HomieTransport::queue_publish(const char* topic, const char* payload, int length, + bool retained, int qos, mqtt_publish_done_fn done, + void* ctx) { + (void)retained; + (void)qos; + homie_logf("TRANSPORT: **ERROR -- this transport implements no queue_publish(); '%s' " + "dropped\n", topic); + if (done) done(ctx, payload, length, false); + return false; +} + +bool HomieTransport::queue_publish(const char* topic, const char* payload, int length, + bool retained, Property* source) { + return queue_publish(topic, payload, length, retained, homie_qos(retained), + source ? property_publish_done : nullptr, source); +} diff --git a/test/support/fake_transport.h b/test/support/fake_transport.h index 0dcd83b..042bc35 100644 --- a/test/support/fake_transport.h +++ b/test/support/fake_transport.h @@ -3,8 +3,9 @@ // order, so a test can assert exactly what would have gone on the wire. // // queue_publish() only records. The real queue reports back later, when its owner task -// sends or drops the message; a test plays that part by calling -// Property::queued_publish_done() itself (see FakeTransport::Record::source). +// sends or drops the message; a test plays that part by calling the recorded callback, +// Record::done(Record::ctx, ...). A Property's queued publish reaches it through +// HomieTransport's adapter, so ctx is the Property. #include #include #include @@ -24,7 +25,8 @@ class FakeTransport final : public HomieTransport { int length; // bytes the caller asked to publish (0 for a subscribe) bool retained; int qos; - Property* source; // queue_publish() only + mqtt_publish_done_fn done; // queue_publish() only + void* ctx; }; bool is_connected = true; @@ -33,18 +35,20 @@ class FakeTransport final : public HomieTransport { bool subscribe_result = true; using HomieTransport::publish; + using HomieTransport::queue_publish; bool publish(const char* topic, const char* payload, int length, bool retained, int qos) override { - record(PUBLISH, topic, payload, length, retained, qos, nullptr); + record(PUBLISH, topic, payload, length, retained, qos, nullptr, nullptr); return is_connected && publish_result; } bool queue_publish(const char* topic, const char* payload, int length, bool retained, - Property* source) override { - record(QUEUED, topic, payload, length, retained, homie_qos(retained), source); + int qos, mqtt_publish_done_fn done, void* ctx) override { + record(QUEUED, topic, payload, length, retained, qos, done, ctx); + if (!queue_result && done) done(ctx, payload, length, false); return queue_result; } bool subscribe(const char* topic, int qos) override { - record(SUBSCRIBE, topic, "", 0, false, qos, nullptr); + record(SUBSCRIBE, topic, "", 0, false, qos, nullptr, nullptr); return is_connected && subscribe_result; } bool connected() override { return is_connected; } @@ -68,7 +72,7 @@ class FakeTransport final : public HomieTransport { private: void record(Kind kind, const char* topic, const char* payload, int length, bool retained, - int qos, Property* source) { + int qos, mqtt_publish_done_fn done, void* ctx) { if (_count < MAX_RECORDS) { Record& r = _records[_count]; r.kind = kind; @@ -80,7 +84,8 @@ class FakeTransport final : public HomieTransport { r.length = length; r.retained = retained; r.qos = qos; - r.source = source; + r.done = done; + r.ctx = ctx; } _count++; } diff --git a/test/test_property_publish/test_property_publish.cpp b/test/test_property_publish/test_property_publish.cpp index 7797131..015c292 100644 --- a/test/test_property_publish/test_property_publish.cpp +++ b/test/test_property_publish/test_property_publish.cpp @@ -116,7 +116,8 @@ static void test_queued_publish_is_not_gated_while_in_flight(void) { TEST_ASSERT_TRUE(p.publish_queued()); TEST_ASSERT_EQUAL_INT(1, t.count_of(FakeTransport::QUEUED)); const FakeTransport::Record* q = t.nth(FakeTransport::QUEUED, 0); - TEST_ASSERT_EQUAL_PTR(&p, q->source); + TEST_ASSERT_EQUAL_PTR(&p, q->ctx); + TEST_ASSERT_NOT_NULL(q->done); TEST_ASSERT_EQUAL_STRING("80", q->payload); TEST_ASSERT_EQUAL_INT(2, q->qos); TEST_ASSERT_EQUAL_UINT32(1, p.queued_count()); @@ -125,9 +126,9 @@ static void test_queued_publish_is_not_gated_while_in_flight(void) { TEST_ASSERT_TRUE(p.publish_queued()); TEST_ASSERT_EQUAL_INT(2, t.count_of(FakeTransport::QUEUED)); - // The queue reports both as sent; now the gate holds. - p.queued_publish_done("80", 2, true); - p.queued_publish_done("80", 2, true); + // The queue reports both as sent, through the adapter's callback; now the gate holds. + q->done(q->ctx, "80", 2, true); + t.nth(FakeTransport::QUEUED, 1)->done(q->ctx, "80", 2, true); TEST_ASSERT_TRUE(p.publish_queued()); TEST_ASSERT_EQUAL_INT(2, t.count_of(FakeTransport::QUEUED)); TEST_ASSERT_EQUAL_STRING("80", p.last_published()); @@ -143,6 +144,67 @@ static void test_refused_queue_publish_is_not_counted(void) { TEST_ASSERT_EQUAL_UINT32(0, p.queued_count()); } +// A port written before ebus_mqtt overrides only the Property overload (esp32-sdk's +// MqttClientTransport does); Property's queued publish must still reach it. +class PropertyQueueTransport final : public HomieTransport { + public: + Property* source = nullptr; + int queued = 0; + + using HomieTransport::publish; + bool publish(const char*, const char*, int, bool, int) override { return true; } + bool queue_publish(const char*, const char* payload, int length, bool, + Property* src) override { + queued++; + source = src; + if (src) src->queued_publish_done(payload, length, true); + return true; + } + bool subscribe(const char*, int) override { return true; } + bool connected() override { return true; } + int last_error() override { return 0; } +}; + +// Overrides neither queue_publish(): the default refuses and still reports back. +class NoQueueTransport final : public HomieTransport { + public: + using HomieTransport::publish; + bool publish(const char*, const char*, int, bool, int) override { return true; } + bool subscribe(const char*, int) override { return true; } + bool connected() override { return true; } + int last_error() override { return 0; } +}; + +static void test_property_overload_override_still_receives_queued_publishes(void) { + static Property p; + add(p, "amps", "integer"); + PropertyQueueTransport legacy; + p.setMQTTClient(&legacy); + p.setValue(7); + + TEST_ASSERT_TRUE(p.publish_queued()); + TEST_ASSERT_EQUAL_INT(1, legacy.queued); + TEST_ASSERT_EQUAL_PTR(&p, legacy.source); + TEST_ASSERT_EQUAL_STRING("7", p.last_published()); + TEST_ASSERT_TRUE(p.publish_queued()); // sent and memoized: gated + TEST_ASSERT_EQUAL_INT(1, legacy.queued); + p.setMQTTClient(&t); +} + +static void test_transport_without_a_queue_refuses_and_reports(void) { + static Property p; + add(p, "watts", "integer"); + NoQueueTransport none; + p.setMQTTClient(&none); + p.setValue(9); + + TEST_ASSERT_FALSE(p.publish_queued()); + TEST_ASSERT_EQUAL_UINT32(0, p.queued_count()); + TEST_ASSERT_NULL(p.last_published()); + TEST_ASSERT_NOT_NULL(strstr(g_log.last_error, "implements no queue_publish()")); + p.setMQTTClient(&t); +} + static void test_subscribe_registers_the_set_topic_at_qos_0(void) { static Property p; add(p, "relay", "boolean"); @@ -194,6 +256,8 @@ int main(int argc, char** argv) { RUN_TEST(test_clear_value_retracts_and_forgets_the_memo); RUN_TEST(test_queued_publish_is_not_gated_while_in_flight); RUN_TEST(test_refused_queue_publish_is_not_counted); + RUN_TEST(test_property_overload_override_still_receives_queued_publishes); + RUN_TEST(test_transport_without_a_queue_refuses_and_reports); RUN_TEST(test_subscribe_registers_the_set_topic_at_qos_0); RUN_TEST(test_subscribe_retries_twice_then_gives_up); RUN_TEST(test_subscribe_while_disconnected_does_nothing); From 60896fa95bb6365c8199f7c42b2fa72f0ec73e65 Mon Sep 17 00:00:00 2001 From: Donald Clark Jackson Date: Fri, 9 Oct 2026 07:18:07 -0700 Subject: [PATCH 4/5] CI: compile ebus_mqtt with its own include directory only Each mqtt/src/ source and mqtt/include/ header is compiled with -I mqtt/include as the only include path, with GCC and Clang, so an include of a Homie or ArduinoJson header in ebus_mqtt fails the build. Part of #3 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Npk6WmkpnBRzWkuhK6uCuz --- .github/workflows/ci.yml | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7a72b8b..485ab35 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -33,6 +33,21 @@ jobs: - name: Test run: ctest --test-dir build-core --output-on-failure + # ebus_mqtt depends on nothing: each of its sources and headers must compile with + # its own include directory as the only one, so a Homie or ArduinoJson include fails. + - name: ebus_mqtt alone + run: | + set -e + flags="-std=c++17 -Wall -Wextra -Werror -I mqtt/include" + for f in mqtt/src/*.cpp; do + echo "$f" + ${{ matrix.cxx }} $flags -c "$f" -o /dev/null + done + for h in mqtt/include/ebus/mqtt/*.h; do + echo "$h" + echo "#include <${h#mqtt/include/}>" | ${{ matrix.cxx }} $flags -x c++ -c - -o /dev/null + done + # The POSIX port (ports/posix): the core on Linux over Paho MQTT C, its demo programs, # and end-to-end tests against a mosquitto each test starts on a free localhost port. posix-port: From 2ff30a975446c19bd46ed01096badacf06451920 Mon Sep 17 00:00:00 2001 From: Donald Clark Jackson Date: Fri, 9 Oct 2026 07:21:36 -0700 Subject: [PATCH 5/5] Document ebus_mqtt: targets, the transport interface, the reconnect order doc/mqtt.md covers MqttTransport and the HomieTransport adapter (which overload a port overrides), PublishHold's rules and sizes, and mqtt_after_connect(). The README gains a table of the CMake targets and the PlatformIO layout; doc/core.md, the POSIX port README and CONTRIBUTING point to the new pieces; CHANGELOG has an Unreleased entry. Closes #4 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Npk6WmkpnBRzWkuhK6uCuz --- CHANGELOG.md | 11 ++++++++ CONTRIBUTING.md | 6 ++-- README.md | 15 +++++++--- doc/core.md | 12 ++++---- doc/mqtt.md | 65 +++++++++++++++++++++++++++++++++++++++++++ ports/posix/README.md | 19 ++++--------- 6 files changed, 102 insertions(+), 26 deletions(-) create mode 100644 doc/mqtt.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 7cc14e0..45a9d24 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,17 @@ All notable changes to `cpp-sdk` are recorded here. Format follows [Keep a Chang ## [Unreleased] +### Added + +- `ebus_mqtt`, a CMake target with no dependency (`mqtt/`, [doc/mqtt.md](doc/mqtt.md)): `MqttTransport`, the transport interface, whose `queue_publish()` takes a QoS and a completion callback (`mqtt_publish_done_fn` and a `void*` context); `PublishHold`, the disconnected-link rules, moved from the POSIX port; and `mqtt_after_connect()`, which runs the reconnect order (flush, re-subscribe, notify). `ebus_core` links it. PlatformIO still builds one library. +- Unity suites `test_mqtt_publish_hold` and `test_mqtt_reconnect`, which link `ebus_mqtt` alone, and a CI step that compiles `ebus_mqtt` with only its own include directory. + +### Changed + +- `HomieTransport` derives from `MqttTransport`. Its `queue_publish(..., Property* source)` is no longer pure: it forwards to the generic overload with a callback that calls `Property::queued_publish_done()`. A port that overrides it, as esp32-sdk's does, builds unchanged; a new port overrides the generic overload instead. +- `PublishHold` sizes are `EBUS_MQTT_HOLD_ENTRIES`, `EBUS_MQTT_HOLD_TOPIC_MAX` and `EBUS_MQTT_HOLD_PAYLOAD_MAX` (were `EBUS_POSIX_HOLD_ENTRIES` and `EBUS_POSIX_HOLD_PAYLOAD_MAX`). It no longer logs; `hold()` returns `EVICTED_OLDEST` or `DROPPED_TOO_LARGE` and the port logs. Its header is ``. +- The POSIX port's `PahoTransport` overrides the generic `queue_publish()` and runs its connect through `mqtt_after_connect()`. + ## [0.1.0] - 2026-10-08 ### Added diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 83622fe..74bc182 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -34,13 +34,13 @@ Pull requests are welcome. - For substantive changes (new public API, changes to `HomieTransport` or the other port interfaces, new ports, changes that alter Homie/eBus wire behavior), open a Discussion or Issue first so we can align on scope before you invest the effort. - **Spec conformance is the north star.** The [Homie 5 convention](https://homieiot.github.io/) and the [Electrification Bus specification](https://github.com/electrification-bus/specification) are the source of truth for on-the-wire behavior: topic structure, datatypes, `$state` semantics, the nested/proxied device model, the eBus device- and capability-type vocabularies. When the code and the spec disagree, the spec wins; fix the code rather than codifying the deviation. - **Update `.ebus-spec.json` on every spec sync.** When you reconcile the core with a newer specification commit, bump `synced_commit`, `synced_date`, and `framework`, and edit `supports` to match the framework features the code now implements (see the specification's `conventions/spec-provenance.md`). -- **House style for the core (`include/`, `src/`).** It runs on microcontrollers, so: +- **House style for the core (`include/`, `src/`, `mqtt/`).** It runs on microcontrollers, so: - No dynamic allocation in steady state. Use fixed-size buffers and static or caller-owned storage; what allocates today is listed in [doc/core.md](doc/core.md) ("Rules for code in the core"), and new code adds nothing to that list. - No STL, no Arduino `String`, no lambdas. Use `snprintf` over string concatenation, and the C function-pointer + `void*` context idiom for callbacks. Members are underscore-prefixed. - - No Arduino, ESP-IDF, FreeRTOS or OS header. The core includes only standard headers, ArduinoJson and other core headers; anything platform-specific goes through the port interfaces. CI builds the core with only its own include directory and fails on anything else. + - No Arduino, ESP-IDF, FreeRTOS or OS header. The core includes only standard headers, ArduinoJson and other core headers, and `ebus_mqtt` (`mqtt/`) only standard headers and its own; anything platform-specific goes through the port interfaces. CI builds each with only its own include directory and fails on anything else. - C++17, GCC or Clang, warnings as errors in CI. - **Ports (`ports/`) may use their platform's facilities**, but keep the core's contract as documented in [doc/core.md](doc/core.md), including the rules for publishes while the broker link is down. -- **Run the tests.** `cmake -S . -B build && cmake --build build -j && ctest --test-dir build --output-on-failure` runs the core's Unity suites; the POSIX port's tests (`ctest --test-dir ports/posix/build`) need `mosquitto`. See the [README](README.md). To add a core suite, create `test/test_/test_.cpp` with its own `main()`; `test/CMakeLists.txt` picks it up. The fakes in `test/support/` stand in for a port. CI runs both test sets with GCC and Clang on every push. +- **Run the tests.** `cmake -S . -B build && cmake --build build -j && ctest --test-dir build --output-on-failure` runs the core's Unity suites; the POSIX port's tests (`ctest --test-dir ports/posix/build`) need `mosquitto`. See the [README](README.md). To add a core suite, create `test/test_/test_.cpp` with its own `main()`; `test/CMakeLists.txt` picks it up. A suite named `test_mqtt_` links `ebus_mqtt` alone. The fakes in `test/support/` stand in for a port. CI runs both test sets with GCC and Clang on every push. - **Test wire changes against a real broker.** A unit test is necessary but not sufficient for anything that changes what goes on the wire; the POSIX port's end-to-end tests and its demo programs run the core against mosquitto. - **Respect upstream licenses: reference, don't copy.** This SDK is MIT. Do **not** paste source from copyleft projects into it. PRs that copy GPL/copyleft code will be asked to reimplement. - **Keep comments to a minimum.** Write self-explanatory code and reserve comments for non-obvious *why* (a spec corner, a platform quirk). Don't add comments that just restate the code. diff --git a/README.md b/README.md index 7e68505..8f12099 100644 --- a/README.md +++ b/README.md @@ -52,7 +52,14 @@ cmake --build build -j ctest --test-dir build --output-on-failure ``` -As the top-level project this builds the `ebus_core` library, compiles each public header on its own, and runs the Unity suites in [`test/`](test/). ArduinoJson 7.4.3 and Unity 2.6.1 are fetched and checked against their SHA-256. A parent CMake project uses `add_subdirectory(/cpp-sdk)` and links `ebus_core`; the tests are then off. +As the top-level project this builds the libraries below, compiles each public header on its own, and runs the Unity suites in [`test/`](test/). ArduinoJson 7.4.3 and Unity 2.6.1 are fetched and checked against their SHA-256. A parent CMake project uses `add_subdirectory(/cpp-sdk)` and links `ebus_core`; the tests are then off. + +| Target | Contents | Depends on | Docs | +|---|---|---|---| +| `ebus_mqtt` | The MQTT transport interface (`MqttTransport`), the publishes held while the link is down (`PublishHold`), the reconnect order (`mqtt_after_connect()`) | nothing | [`doc/mqtt.md`](doc/mqtt.md) | +| `ebus_core` | Everything: the Homie model, `/set` dispatch, the controller, validation, broker discovery, and `ebus_mqtt` | `ebus_mqtt`, ArduinoJson | [`doc/core.md`](doc/core.md) | + +`ebus_mqtt` builds with only its own include directory (`mqtt/include`), so a project can take it alone with `add_subdirectory(/cpp-sdk/mqtt)`. ## Writing a port @@ -60,16 +67,16 @@ A port binds four things, all documented in [`doc/core.md`](doc/core.md) ("What | Piece | Header | What the port does | |---|---|---| -| MQTT transport | `homie/homie_transport.h` | Subclasses `HomieTransport`: `publish()`, `subscribe()` and `queue_publish()` | +| MQTT transport | `homie/homie_transport.h` | Subclasses `HomieTransport`, which is `ebus_mqtt`'s `MqttTransport` plus a Homie adapter: `publish()`, `subscribe()` and `queue_publish()` with a completion callback ([`doc/mqtt.md`](doc/mqtt.md)) | | Clock | `homie/homie_clock.h` | `homie_clock_bind(now_ms, sleep_ms)` | | Console | `homie/homie_log.h` | `homie_log_bind(vprintf_sink)`; optional, stdout otherwise | | Settable table | `homie/homie_settable.h` | Supplies storage to `settable_table_bind()`, routes each claimed `/set` message to `settable_dispatch()` on its own task, and subscribes every settable topic after each connect | -The POSIX port is the reference: [`ports/posix/src/posix_port.cpp`](ports/posix/src/posix_port.cpp) binds the clock, console and settable table, and [`ports/posix/src/paho_transport.cpp`](ports/posix/src/paho_transport.cpp) is the transport. A port should also follow the rules for publishes made while the broker link is down, which the POSIX port takes from [ebus-mqtt-client](https://github.com/electrification-bus/ebus-mqtt-client). +The POSIX port is the reference: [`ports/posix/src/posix_port.cpp`](ports/posix/src/posix_port.cpp) binds the clock, console and settable table, and [`ports/posix/src/paho_transport.cpp`](ports/posix/src/paho_transport.cpp) is the transport. A port should also follow the rules for publishes made while the broker link is down and the reconnect order, both from [ebus-mqtt-client](https://github.com/electrification-bus/ebus-mqtt-client); `ebus_mqtt` provides them as `PublishHold` and `mqtt_after_connect()`, which the POSIX port uses. ## Using the core from PlatformIO -[`library.json`](library.json) makes this repository a PlatformIO library named `ebus_core` that compiles `include/` and `src/` and depends on ArduinoJson 7.4.3. Add it with a git URL: +[`library.json`](library.json) makes this repository one PlatformIO library named `ebus_core` that compiles `src/` and `mqtt/src/`, puts `include/` and `mqtt/include/` on the include path, and depends on ArduinoJson 7.4.3. Add it with a git URL: ```ini lib_deps = diff --git a/doc/core.md b/doc/core.md index 389d738..b121e77 100644 --- a/doc/core.md +++ b/doc/core.md @@ -7,7 +7,7 @@ The portable part of the eBus / Homie 5 SDK: the Homie device model, `/set` disp | `homie/Device.h`, `homie/Node.h`, `homie/Property.h` | The Homie object model: device tree, `$state` / `$description` lifecycle, property values and the publish-on-change gate | | `homie/homie_settable.h` | The settable table: `/set` registration, dispatch, `deliver_local_set()` | | `homie/controller.h` | Controller role: discovery, device tree from `$description`, `controller_set_property()` | -| `homie/homie_transport.h` | `HomieTransport`, the MQTT interface a port implements; `MAX_DATA_LEN`; `mqtt_qos` and `homie_qos()` | +| `homie/homie_transport.h` | `HomieTransport`, the MQTT interface a port implements: `ebus_mqtt`'s `MqttTransport` ([`mqtt.md`](mqtt.md)) plus the Property `queue_publish()` adapter; `MAX_DATA_LEN`; `mqtt_qos` and `homie_qos()` | | `homie/homie_log.h` | `homie_logf()` / `homie_logln()` and the sink a port binds | | `homie/homie_clock.h` | `homie_now_ms()` / `homie_sleep_ms()` and the hooks a port binds | | `homie/homie_json.h`, `util/jsonUtils.h` | `$description` (de)serialization helpers (ArduinoJson) | @@ -26,10 +26,10 @@ The portable part of the eBus / Homie 5 SDK: the Homie device model, `/set` disp | Piece | Contract | ESP32 port ([esp32-sdk](https://github.com/electrification-bus/esp32-sdk)) | |---|---|---| -| MQTT transport | A `HomieTransport` subclass. `publish()` and `subscribe()` run on the task that owns the client and never inside its receive callback. `queue_publish()` may be called from any task and must call `source->queued_publish_done(payload, length, sent)` exactly once per call when `source` is set, including when it fails. `last_error()` is for logs only. | `MqttClientTransport` in `include/platform/mqtt_client.h`, over arduino-mqtt and the FreeRTOS publish queue; the instance is `mqtt_transport` | +| MQTT transport | A `HomieTransport` subclass. `publish()` and `subscribe()` run on the task that owns the client and never inside its receive callback. `queue_publish()` may be called from any task; the port overrides either the generic overload, which must call `done(ctx, payload, length, sent)` exactly once per call when `done` is set, or the Property overload, which must call `source->queued_publish_done(payload, length, sent)` exactly once per call when `source` is set, in both cases including when it fails ([`mqtt.md`](mqtt.md)). `last_error()` is for logs only. | `MqttClientTransport` in `include/platform/mqtt_client.h`, over arduino-mqtt and the FreeRTOS publish queue; the instance is `mqtt_transport` | | Clock | `homie_clock_bind(now_ms, sleep_ms)`. Required: unbound, the tick reads 0 and the sleep returns at once. | `millis()` / `delay()`, bound in `src/platform/homie_port.cpp` | | Console | `homie_log_bind(vprintf_sink)`. Optional: unbound, output goes to `vprintf()` on stdout. | The UART + TCP tee, bound in `src/platform/homie_port.cpp` | -| Settable table | Storage for `capacity` entries, passed with the transport and the driver call to `settable_table_bind()` before anything registers. The receive callback asks `settable_is_registered(topic)` and, if so, copies the message for the owning task, which runs `settable_dispatch(topic, value)`. After every connect, subscribe `settable_topic(i)` for `i < settable_count()`. | `mqtt_queue_init()`, `subscriber_callback()`, `mqtt_process_queue()` and `broker_connect()` in `src/platform/mqtt_client.cpp` | +| Settable table | Storage for `capacity` entries, passed with the transport and the driver call to `settable_table_bind()` before anything registers. The receive callback asks `settable_is_registered(topic)` and, if so, copies the message for the owning task, which runs `settable_dispatch(topic, value)`. After every connect, subscribe `settable_topic(i)` for `i < settable_count()` (step 2 of the reconnect order in [`mqtt.md`](mqtt.md)). | `mqtt_queue_init()`, `subscriber_callback()`, `mqtt_process_queue()` and `broker_connect()` in `src/platform/mqtt_client.cpp` | | Driver call | A `settable_entity_call_t` that calls a driver's settable callback, defined where `NodeEntity` is. | `node_entity_settable_call()` in `src/node/NodeEntity.cpp` | | Root device | `Device::init(name, id, type, &transport)`; child devices share the root's `mqttClient()`. | `theDevice.init(..., &mqtt_transport)` in `src/main.cpp` | | QoS | Set `mqtt_qos` (default 2) for retained publications, if it is configurable. | `src/platform/config.cpp`, from the `mqtt_qos` config key | @@ -40,14 +40,14 @@ The POSIX port in [`ports/posix/`](../ports/posix/) implements the same pieces o ## Rules for code in the core -- Include only standard headers, ArduinoJson and other core headers. CI enforces this. +- Include only standard headers, ArduinoJson and other core headers; `ebus_mqtt` (`mqtt/`) includes only standard headers and its own. CI enforces both. - No Arduino `String`, no STL containers, no lambdas: callers pass buffers. - No new dynamic allocation. What allocates today predates the move: `Device::addNode()` (each `Node`), `Device::addNodePropertiesFromConfigJson()` (each `Property`, controller path only), the `$description` buffer (`MAX_DATA_LEN` bytes, once, on first use), the controller's `Device` per discovered device, and ArduinoJson's `JsonDocument` pools. - C++17, GCC or Clang (`__atomic` builtins, `__attribute__((format))`). ## Building -**PlatformIO**: `library.json` makes this repository a library that declares its ArduinoJson dependency and compiles only `include/` and `src/`; a project adds it with a `lib_deps` git URL. A library is compiled without the project's `build_src_flags`, so a macro the core reads (`MAX_DATA_LEN`, `CONTROLLER_INBOX_BYTES`, `USE_EBUS_TOPIC`) must be set in `build_flags`. +**PlatformIO**: `library.json` makes this repository one library that declares its ArduinoJson dependency, compiles only `src/` and `mqtt/src/`, and puts `include/` and `mqtt/include/` on the include path of the library and of the project that uses it; a project adds it with a `lib_deps` git URL. A library is compiled without the project's `build_src_flags`, so a macro the core reads (`MAX_DATA_LEN`, `CONTROLLER_INBOX_BYTES`, `USE_EBUS_TOPIC`) must be set in `build_flags`. **CMake**, for any other build system or host: @@ -57,7 +57,7 @@ cmake --build build ctest --test-dir build --output-on-failure ``` -A parent project uses `add_subdirectory(/cpp-sdk)` and links the `ebus_core` target, which carries its include directory, C++17 and ArduinoJson. A parent that already defines an `ArduinoJson` target keeps it; otherwise `FetchContent` downloads ArduinoJson 7.4.3, the release esp32-sdk's `platformio.ini` uses, checked against its SHA-256, so both builds serialize `$description` with the same code. `find_package()` was not used because few systems package ArduinoJson and none would pin that release. To build offline, pass `-DFETCHCONTENT_SOURCE_DIR_ARDUINOJSON=`. `-DEBUS_CORE_EBUS_TOPIC=ON` defines `USE_EBUS_TOPIC`, which moves the topic root from `homie/5` to `ebus/5`; the ESP32 firmware sets it in `platformio.ini`. Built as the top-level project, the target `ebus_core_header_check` also compiles each public header in a translation unit of its own, and the Unity suites in `test/` are built and registered with CTest (`-DEBUS_CORE_TESTS=OFF` skips them). +A parent project uses `add_subdirectory(/cpp-sdk)` and links the `ebus_core` target, which carries its include directory, C++17, ArduinoJson and `ebus_mqtt`. A parent that already defines an `ArduinoJson` target keeps it; otherwise `FetchContent` downloads ArduinoJson 7.4.3, the release esp32-sdk's `platformio.ini` uses, checked against its SHA-256, so both builds serialize `$description` with the same code. `find_package()` was not used because few systems package ArduinoJson and none would pin that release. To build offline, pass `-DFETCHCONTENT_SOURCE_DIR_ARDUINOJSON=`. `-DEBUS_CORE_EBUS_TOPIC=ON` defines `USE_EBUS_TOPIC`, which moves the topic root from `homie/5` to `ebus/5`; the ESP32 firmware sets it in `platformio.ini`. Built as the top-level project, the targets `ebus_core_header_check` and `ebus_mqtt_header_check` also compile each public header in a translation unit of its own (an `ebus_mqtt` header against `ebus_mqtt` alone), and the Unity suites in `test/` are built and registered with CTest (`-DEBUS_CORE_TESTS=OFF` skips them). CI (`core-host-build` in `.github/workflows/ci.yml`) runs that CMake build and the tests with GCC and Clang, warnings as errors, and only the core's include directory and ArduinoJson on the core's path. diff --git a/doc/mqtt.md b/doc/mqtt.md new file mode 100644 index 0000000..da73a4b --- /dev/null +++ b/doc/mqtt.md @@ -0,0 +1,65 @@ +# ebus_mqtt + +The MQTT layer under the Homie model, with no Homie in it: the transport interface a port implements, the publishes held while the broker link is down, and the order of work after a connect. It depends on the C and C++ standard libraries only, the same split as the Python [ebus-mqtt-client](https://github.com/electrification-bus/ebus-mqtt-client). Sources are in `mqtt/`; CMake builds them as the `ebus_mqtt` target, which `ebus_core` links. + +| Header | Contents | +|---|---| +| `ebus/mqtt/transport.h` | `MqttTransport`, the interface a port implements; `mqtt_publish_done_fn` | +| `ebus/mqtt/publish_hold.h` | `PublishHold`, the disconnected-link rules, in fixed storage | +| `ebus/mqtt/reconnect.h` | `mqtt_after_connect()`, the reconnect order | + +## The transport + +```cpp +typedef void (*mqtt_publish_done_fn)(void* ctx, const char* payload, int length, bool sent); + +class MqttTransport { + public: + virtual bool publish(const char* topic, const char* payload, int length, bool retained, int qos) = 0; + virtual bool queue_publish(const char* topic, const char* payload, int length, bool retained, int qos, + mqtt_publish_done_fn done, void* ctx) = 0; + virtual bool subscribe(const char* topic, int qos) = 0; + virtual bool connected() = 0; + virtual int last_error() = 0; +}; +``` + +`publish()` and `subscribe()` run on the task that owns the client, never inside its receive callback. `queue_publish()` may be called from any task; the port copies the message for the owning task and calls `done(ctx, payload, length, sent)` exactly once per call when `done` is set, including when the call itself fails. + +The Homie layer uses `HomieTransport` (`homie/homie_transport.h`), which derives from `MqttTransport` and adds `queue_publish(topic, payload, length, retained, Property* source)`. Property calls that overload; by default it forwards to the generic one at `homie_qos(retained)`, with a callback that calls `source->queued_publish_done()`. A port overrides one of the two: + +| Port | Overrides | Example | +|---|---|---| +| New | `queue_publish(..., qos, done, ctx)` | `PahoTransport` in `ports/posix/` | +| Written against 0.1.0 | `queue_publish(..., Property* source)` | esp32-sdk's `MqttClientTransport` | + +A `HomieTransport` that overrides neither refuses every queued publish, logs an error, and reports it to `done` as not sent. + +## While the link is down + +`PublishHold` applies ebus-mqtt-client's rules ("Publishing before the connection is up"): + +- Retained, any QoS: held, newest value per topic; a newer value moves to the back of the flush order. +- Not retained, QoS 1 or 2: held in order, one entry per publish. +- Not retained, QoS 0: dropped (`DROPPED_QOS0`). Every non-retained Homie publication is QoS 0, so on a Homie device only retained values are held. +- At most `EBUS_MQTT_HOLD_ENTRIES` (64) entries, evicting the oldest (`EVICTED_OLDEST`). A topic over `EBUS_MQTT_HOLD_TOPIC_MAX` (127 characters) or a payload over `EBUS_MQTT_HOLD_PAYLOAD_MAX` (1024 bytes, so a `$description`) is dropped (`DROPPED_TOO_LARGE`). + +`hold()` returns what it did and does not log, so the port decides what to log. `flush(send, ctx)` sends the oldest first and stops at the first send that returns false, keeping that entry and the rest. The storage is in the object (about 75 KB with the defaults), so declare it static. The three sizes set the class layout: an override goes in the flags of every translation unit that includes the header (PlatformIO `build_flags`; CMake `target_compile_definitions(ebus_mqtt PUBLIC ...)`). Only the task that owns the client may touch it. + +A port marks the link down when it is lost and, while it is down, passes each `publish()` to `hold()` instead of the client. + +## After a connect + +Every successful connect runs, in order: + +1. Flush the hold. What was published while the link was down describes state that already exists, so it reaches the broker before anything reacts to the connect. +2. Subscribe every topic again. +3. Notify the application, which may now publish and subscribe. + +Until the flush has finished the port keeps treating the link as down, so a publish made meanwhile joins the hold behind older values instead of overtaking them. `mqtt_after_connect(hold, steps, first, report)` runs the three steps from hooks in `MqttConnectSteps` (`send`, `resubscribe`, `notify`, `ctx`). The `resubscribe` hook marks the link usable for `publish()` and `subscribe()` before subscribing. The function returns false, without the later steps, when the flush stops early or `resubscribe` reports the link lost; the port then treats the link as down and runs it again after the next connect. `report` gives the number held and the number flushed. `PahoTransport::try_connect()` in [`ports/posix/src/paho_transport.cpp`](../ports/posix/src/paho_transport.cpp) is the reference. + +## Building + +`ebus_mqtt` has `mqtt/include` as its only include directory. CI (`core-host-build`, step "ebus_mqtt alone") compiles each source and header in `mqtt/` with that one `-I` and nothing else, and the `ebus_mqtt_header_check` target compiles each header against `ebus_mqtt` alone. Its Unity suites are `test/test_mqtt_*`, which link `ebus_mqtt` without `ebus_core`. A project that wants only this layer can `add_subdirectory(/cpp-sdk/mqtt)`. + +PlatformIO builds it as part of the single `ebus_core` library: `library.json` compiles `src/` and `mqtt/src/` and adds `mqtt/include` to the include path, which PlatformIO also gives the library's dependents. diff --git a/ports/posix/README.md b/ports/posix/README.md index d5c3698..6744b20 100644 --- a/ports/posix/README.md +++ b/ports/posix/README.md @@ -1,15 +1,14 @@ # POSIX port -The eBus core ([`doc/core.md`](../../doc/core.md)) on Linux and macOS: a `HomieTransport` over the Eclipse Paho MQTT C client, the clock and log bindings, a demo device, a demo controller, and end-to-end tests against a real mosquitto. CMake only; PlatformIO never builds this directory, because the root `library.json` limits a PlatformIO build to `include/` and `src/`. +The eBus core ([`doc/core.md`](../../doc/core.md)) on Linux and macOS: a `HomieTransport` over the Eclipse Paho MQTT C client, the clock and log bindings, a demo device, a demo controller, and end-to-end tests against a real mosquitto. CMake only; PlatformIO never builds this directory, because the root `library.json` limits a PlatformIO build to `src/` and `mqtt/src/`. | Path | Contents | |---|---| | `include/ebus_posix/paho_transport.h`, `src/paho_transport.cpp` | `PahoTransport`: connect, reconnect, Last Will, inbound deferral, `queue_publish()` from any thread | -| `include/ebus_posix/publish_hold.h`, `src/publish_hold.cpp` | What is held while the broker link is down | | `include/ebus_posix/posix_port.h`, `src/posix_port.cpp` | Clock (`CLOCK_MONOTONIC`, `nanosleep`), log (stderr), settable table storage, `ebus_posix_register_settable()` | | `apps/device_main.cpp` | `ebus-posix-device` | | `apps/controller_main.cpp` | `ebus-posix-controller` | -| `test/` | `test_publish_hold` (no broker) and `ebus_posix_e2e` (mosquitto) | +| `test/` | `ebus_posix_e2e` (mosquitto) | ## Build @@ -21,7 +20,7 @@ ctest --test-dir ports/posix/build --output-on-failure Paho MQTT C v1.3.16 is fetched and checked against its SHA-256, and built as a static, plain-TCP library. An installed Paho is used instead when CMake finds its `eclipse-paho-mqtt-c` package; `-DEBUS_POSIX_PAHO=fetch` or `=system` forces one or the other. To build offline, pass `-DFETCHCONTENT_SOURCE_DIR_PAHO_MQTT_C=` (and `FETCHCONTENT_SOURCE_DIR_ARDUINOJSON` for the core). `-DEBUS_POSIX_WERROR=ON` makes warnings in the port, demos and tests errors, as CI does. -The end-to-end tests need `mosquitto` (Homebrew: `brew install mosquitto`; Debian/Ubuntu: `apt-get install mosquitto`). Each test starts its own on a free 127.0.0.1 port and leaves its logs in `build/e2e//`. Without mosquitto, CMake warns and registers only `publish_hold`. +The end-to-end tests need `mosquitto` (Homebrew: `brew install mosquitto`; Debian/Ubuntu: `apt-get install mosquitto`). Each test starts its own on a free 127.0.0.1 port and leaves its logs in `build/e2e//`. Without mosquitto, CMake warns and registers no test. ## Run @@ -40,20 +39,14 @@ The device publishes `/5/` with `switch/on` (boolean, settable) and | Piece | Here | |---|---| | Inbound messages | Paho's receive thread copies each message into a mutex-guarded arena (`EBUS_POSIX_INBOX_BYTES`, 64 KB) and returns. `PahoTransport::loop()`, on the application's thread, hands each one to `settable_dispatch()` if the settable table claims the topic, else to the fallback (`controller_mqtt_callback()` for a controller). Nothing that publishes or subscribes runs on Paho's thread. | -| `queue_publish()` | Any thread: copies into a fixed ring (`EBUS_POSIX_PUBLISH_QUEUE`, 64 entries of `Property::VALUE_MAX`), which `loop()` sends. `queued_publish_done()` is called exactly once per call, from `loop()`, or at once when the ring is full or the message too long. | -| Connect | Clean session, keepalive 60 s, Last Will `/$state` = `lost`, retained, at `homie_qos(true)`. After each connect, in order: flush the hold, subscribe every settable topic, call the application's `on_connected(first)`. The device publishes the tree on the first and `publishStateTree()` on later ones, as the ESP32 firmware does. | +| `queue_publish()` | `PahoTransport` overrides the generic overload (`ebus_mqtt`'s `MqttTransport`); a Property reaches it through `HomieTransport`'s adapter. Any thread: copies into a fixed ring (`EBUS_POSIX_PUBLISH_QUEUE`, 64 entries of `Property::VALUE_MAX`), which `loop()` sends. The completion callback is called exactly once per call, from `loop()`, or at once when the ring is full or the message too long. | +| Connect | Clean session, keepalive 60 s, Last Will `/$state` = `lost`, retained, at `homie_qos(true)`. After each connect, `mqtt_after_connect()` runs, in order: flush the hold, subscribe every settable topic, call the application's `on_connected(first)`. The device publishes the tree on the first and `publishStateTree()` on later ones, as the ESP32 firmware does. | | Link down | `publish()` holds or drops (below) and returns false; `loop()` reconnects at once, then every `--reconnect-ms`. | | Settable table | `EBUS_POSIX_SETTABLE_CAPACITY` (64) static entries. No `NodeEntity` on this port, so no driver call is bound; a settable reaches the application through a `settable_handler_t`, registered with `ebus_posix_register_settable()`, which also works before the first connect. | ### While the link is down -The rules of [ebus-mqtt-client](https://github.com/electrification-bus/ebus-mqtt-client) ("Publishing before the connection is up"), in `PublishHold`: - -- Retained, any QoS: held, newest value per topic; a newer value moves to the back of the flush order. -- Not retained, QoS 1 or 2: held in order, one entry per publish. -- Not retained, QoS 0: dropped. Every non-retained Homie publication is QoS 0, so in practice only retained values are held. -- At most `EBUS_POSIX_HOLD_ENTRIES` (64), evicting the oldest; a payload over `EBUS_POSIX_HOLD_PAYLOAD_MAX` (1024 bytes, so a `$description`) is dropped and logged. -- Flushed on connect before the subscriptions and before `on_connected`. A flush the link interrupts keeps what it did not send. `disconnect()` discards the hold. +`publish()` passes each message to `ebus_mqtt`'s `PublishHold`, which applies the rules of [ebus-mqtt-client](https://github.com/electrification-bus/ebus-mqtt-client) ([`doc/mqtt.md`](../../doc/mqtt.md#while-the-link-is-down)), and logs what it evicts or drops as too large. The hold is flushed on connect before the subscriptions and before `on_connected`; a flush the link interrupts keeps what it did not send. `disconnect()` discards the hold. ## Known gaps