From ca33731b54d2ffc5bc2bf6a90d1934a4ec3dffda Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 9 Oct 2026 07:35:01 -1000 Subject: [PATCH] [mqtt] Build the mqtt triggers with build_trigger_callback --- esphome/components/mqtt/__init__.py | 87 +++++++++--------- esphome/components/mqtt/mqtt_client.cpp | 70 +++++---------- esphome/components/mqtt/mqtt_client.h | 90 +++++++++---------- esphome/components/mqtt/mqtt_component.cpp | 8 -- esphome/components/mqtt/mqtt_component.h | 8 +- esphome/core/helpers.h | 32 ++++++- tests/components/mqtt/common-triggers.yaml | 25 ++++++ .../mqtt/test-triggers.esp32-idf.yaml | 3 + 8 files changed, 169 insertions(+), 154 deletions(-) create mode 100644 tests/components/mqtt/common-triggers.yaml create mode 100644 tests/components/mqtt/test-triggers.esp32-idf.yaml diff --git a/esphome/components/mqtt/__init__.py b/esphome/components/mqtt/__init__.py index b21dc2e2f0..77aaf031dd 100644 --- a/esphome/components/mqtt/__init__.py +++ b/esphome/components/mqtt/__init__.py @@ -53,7 +53,6 @@ from esphome.const import ( CONF_SUBSCRIBE_QOS, CONF_TOPIC, CONF_TOPIC_PREFIX, - CONF_TRIGGER_ID, CONF_USE_ABBREVIATIONS, CONF_USERNAME, CONF_WILL_MESSAGE, @@ -119,18 +118,6 @@ MQTTMessage = mqtt_ns.struct("MQTTMessage") MQTTClientDisconnectReason = mqtt_ns.enum("MQTTClientDisconnectReason") MQTTClientComponent = mqtt_ns.class_("MQTTClientComponent", cg.Component) MQTTPublishJsonAction = mqtt_ns.class_("MQTTPublishJsonAction", automation.Action) -MQTTMessageTrigger = mqtt_ns.class_( - "MQTTMessageTrigger", automation.Trigger.template(cg.std_string), cg.Component -) -MQTTJsonMessageTrigger = mqtt_ns.class_( - "MQTTJsonMessageTrigger", automation.Trigger.template(cg.JsonObjectConst) -) -MQTTConnectTrigger = mqtt_ns.class_( - "MQTTConnectTrigger", automation.Trigger.template(cg.bool_) -) -MQTTDisconnectTrigger = mqtt_ns.class_( - "MQTTDisconnectTrigger", automation.Trigger.template(MQTTClientDisconnectReason) -) MQTTComponent = mqtt_ns.class_("MQTTComponent", cg.Component) MQTTAlarmControlPanelComponent = mqtt_ns.class_( @@ -284,21 +271,10 @@ CONFIG_SCHEMA = cv.All( cv.Optional( CONF_REBOOT_TIMEOUT, default="15min" ): cv.positive_time_period_milliseconds, - cv.Optional(CONF_ON_CONNECT): automation.validate_automation( - { - cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(MQTTConnectTrigger), - } - ), - cv.Optional(CONF_ON_DISCONNECT): automation.validate_automation( - { - cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id( - MQTTDisconnectTrigger - ), - } - ), + cv.Optional(CONF_ON_CONNECT): automation.validate_automation(), + cv.Optional(CONF_ON_DISCONNECT): automation.validate_automation(), cv.Optional(CONF_ON_MESSAGE): automation.validate_automation( { - cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(MQTTMessageTrigger), cv.Required(CONF_TOPIC): cv.subscribe_topic, cv.Optional(CONF_QOS, default=0): cv.mqtt_qos, cv.Optional(CONF_PAYLOAD): cv.string_strict, @@ -306,9 +282,6 @@ CONFIG_SCHEMA = cv.All( ), cv.Optional(CONF_ON_JSON_MESSAGE): automation.validate_automation( { - cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id( - MQTTJsonMessageTrigger - ), cv.Required(CONF_TOPIC): cv.subscribe_topic, cv.Optional(CONF_QOS, default=0): cv.mqtt_qos, } @@ -343,6 +316,18 @@ def exp_mqtt_message(config): ) +_CALLBACK_AUTOMATIONS = ( + automation.CallbackAutomation( + CONF_ON_CONNECT, "set_on_connect", [(cg.bool_, "session_present")] + ), + automation.CallbackAutomation( + CONF_ON_DISCONNECT, + "set_on_disconnect", + [(MQTTClientDisconnectReason, "reason")], + ), +) + + @coroutine_with_priority(CoroPriority.WEB) async def to_code(config): var = cg.new_Pvariable(config[CONF_ID]) @@ -457,29 +442,37 @@ async def to_code(config): cg.add_define("USE_MQTT_IDF_ENQUEUE") # end esp-idf + # The client queues subscriptions until it connects, so they can be made at construction. for conf in config.get(CONF_ON_MESSAGE, []): - trig = cg.new_Pvariable(conf[CONF_TRIGGER_ID], conf[CONF_TOPIC]) - cg.add(trig.set_qos(conf[CONF_QOS])) - if CONF_PAYLOAD in conf: - cg.add(trig.set_payload(conf[CONF_PAYLOAD])) - await cg.register_component(trig, conf) - await automation.build_automation(trig, [(cg.std_string, "x")], conf) + callback = await automation.build_trigger_callback( + [(cg.std_string, "x")], + conf, + params=[(cg.std_string, "topic"), (cg.std_string, "payload")], + forward=["payload"], + # The length is compared first; the literal stays in flash on ESP8266. + when=automation.ApplyCall( + "StringRef(payload) == {}", + ((CONF_PAYLOAD, cg.std_string, automation.string_ref_literal),), + ), + ) + cg.add( + var.subscribe(cg.progmem_string(conf[CONF_TOPIC]), callback, conf[CONF_QOS]) + ) for conf in config.get(CONF_ON_JSON_MESSAGE, []): - trig = cg.new_Pvariable(conf[CONF_TRIGGER_ID], conf[CONF_TOPIC], conf[CONF_QOS]) - await automation.build_automation(trig, [(cg.JsonObjectConst, "x")], conf) - - for conf in config.get(CONF_ON_CONNECT, []): - trigger = cg.new_Pvariable(conf[CONF_TRIGGER_ID], var) - await automation.build_automation( - trigger, [(cg.bool_, "session_present")], conf + callback = await automation.build_trigger_callback( + [(cg.JsonObjectConst, "x")], + conf, + params=[(cg.std_string, "topic"), (cg.JsonObject, "root")], + forward=["root"], + ) + cg.add( + var.subscribe_json( + cg.progmem_string(conf[CONF_TOPIC]), callback, conf[CONF_QOS] + ) ) - for conf in config.get(CONF_ON_DISCONNECT, []): - trigger = cg.new_Pvariable(conf[CONF_TRIGGER_ID], var) - await automation.build_automation( - trigger, [(MQTTClientDisconnectReason, "reason")], conf - ) + await automation.build_callback_automations(var, config, _CALLBACK_AUTOMATIONS) cg.add(var.set_publish_nan_as_none(config[CONF_PUBLISH_NAN_AS_NONE])) diff --git a/esphome/components/mqtt/mqtt_client.cpp b/esphome/components/mqtt/mqtt_client.cpp index b3d8ab80b1..c849beb1cc 100644 --- a/esphome/components/mqtt/mqtt_client.cpp +++ b/esphome/components/mqtt/mqtt_client.cpp @@ -461,37 +461,24 @@ void MQTTClientComponent::resubscribe_subscriptions_() { } } -void MQTTClientComponent::subscribe(const std::string &topic, mqtt_callback_t callback, uint8_t qos) { - MQTTSubscription subscription{ - .topic = topic, +void MQTTClientComponent::add_subscription_(std::string &&topic, mqtt_callback_t callback, bool boxed, uint8_t qos) { + auto &subscription = this->subscriptions_.emplace_back(MQTTSubscription{ + .topic = std::move(topic), .qos = qos, - .callback = std::move(callback), .subscribed = false, + .boxed = boxed, + .callback = callback, .resubscribe_timeout = 0, - }; + }); this->resubscribe_subscription_(&subscription); - this->subscriptions_.push_back(subscription); -} - -void MQTTClientComponent::subscribe_json(const std::string &topic, const mqtt_json_callback_t &callback, uint8_t qos) { - auto f = [callback](const std::string &topic, const std::string &payload) { - json::parse_json(payload, [topic, callback](JsonObject root) -> bool { - callback(topic, root); - return true; - }); - }; - MQTTSubscription subscription{ - .topic = topic, - .qos = qos, - .callback = f, - .subscribed = false, - .resubscribe_timeout = 0, - }; - this->resubscribe_subscription_(&subscription); - this->subscriptions_.push_back(subscription); } void MQTTClientComponent::unsubscribe(const std::string &topic) { + if (this->dispatching_) { + // Would erase, and for a boxed callback free, the subscription that may be running right now. + ESP_LOGE(TAG, "Cannot unsubscribe from '%s' inside a subscription callback", topic.c_str()); + return; + } bool ret = this->mqtt_backend_.unsubscribe(topic.c_str()); yield(); if (ret) { @@ -505,6 +492,8 @@ void MQTTClientComponent::unsubscribe(const std::string &topic) { auto it = subscriptions_.begin(); while (it != subscriptions_.end()) { if (it->topic == topic) { + if (it->boxed) + it->callback.free_boxed(); it = subscriptions_.erase(it); } else { ++it; @@ -660,10 +649,16 @@ void MQTTClientComponent::on_message(const std::string &topic, const std::string // in simple tests but will cause crashes with complex automations. this->defer([this, topic, payload]() { #endif - for (auto &subscription : this->subscriptions_) { + // Indexed with the count taken up front: a callback may subscribe, which can reallocate the + // vector, and the new entry waits for the next message. + this->dispatching_ = true; + const size_t count = this->subscriptions_.size(); + for (size_t i = 0; i < count; i++) { + const auto &subscription = this->subscriptions_[i]; if (topic_match(topic.c_str(), subscription.topic.c_str())) - subscription.callback(topic, payload); + subscription.callback.call(topic, payload); } + this->dispatching_ = false; #ifdef USE_ESP8266 }); #endif @@ -762,29 +757,6 @@ void MQTTClientComponent::set_on_disconnect(mqtt_on_disconnect_callback_t &&call MQTTClientComponent *global_mqtt_client = nullptr; // NOLINT(cppcoreguidelines-avoid-non-const-global-variables) -// MQTTMessageTrigger -MQTTMessageTrigger::MQTTMessageTrigger(std::string topic) : topic_(std::move(topic)) {} -void MQTTMessageTrigger::setup() { - global_mqtt_client->subscribe( - this->topic_, - [this](const std::string &topic, const std::string &payload) { - if (this->payload_.has_value() && payload != *this->payload_) { - return; - } - - this->trigger(payload); - }, - this->qos_); -} -void MQTTMessageTrigger::dump_config() { - ESP_LOGCONFIG(TAG, - "MQTT Message Trigger:\n" - " Topic: '%s'\n" - " QoS: %u", - this->topic_.c_str(), this->qos_); -} -float MQTTMessageTrigger::get_setup_priority() const { return setup_priority::AFTER_CONNECTION; } - } // namespace esphome::mqtt #endif // USE_MQTT diff --git a/esphome/components/mqtt/mqtt_client.h b/esphome/components/mqtt/mqtt_client.h index ced9c84e10..f16a285eed 100644 --- a/esphome/components/mqtt/mqtt_client.h +++ b/esphome/components/mqtt/mqtt_client.h @@ -31,19 +31,16 @@ namespace esphome::mqtt { using mqtt_on_connect_callback_t = std::function; using mqtt_on_disconnect_callback_t = std::function; -/** Callback for MQTT subscriptions. - * - * First parameter is the topic, the second one is the payload. - */ -using mqtt_callback_t = std::function; -using mqtt_json_callback_t = std::function; +/// Callback for MQTT subscriptions: the topic, then the payload. +using mqtt_callback_t = Callback; /// internal struct for MQTT subscriptions. struct MQTTSubscription { std::string topic; uint8_t qos; - mqtt_callback_t callback; bool subscribed; + bool boxed; // made by create_boxed(), freed on unsubscribe + mqtt_callback_t callback; uint32_t resubscribe_timeout; }; @@ -165,12 +162,21 @@ class MQTTClientComponent final : public Component { bool is_log_message_enabled() const; /** Subscribe to an MQTT topic and call callback when a message is received. + * + * A callback that fits in a pointer (capture at most `this`) is stored inline; a larger one is + * boxed on the heap and freed by unsubscribe(). * * @param topic The topic. Wildcards are currently not supported. * @param callback The callback function. * @param qos The QoS of this subscription. */ - void subscribe(const std::string &topic, mqtt_callback_t callback, uint8_t qos = 0); + template void subscribe(std::string topic, F &&callback, uint8_t qos = 0) { + if constexpr (mqtt_callback_t::fits_inline()) { + this->add_subscription_(std::move(topic), mqtt_callback_t::create(std::forward(callback)), false, qos); + } else { + this->add_subscription_(std::move(topic), mqtt_callback_t::create_boxed(std::forward(callback)), true, qos); + } + } /** Subscribe to a MQTT topic and automatically parse JSON payload. * @@ -181,11 +187,30 @@ class MQTTClientComponent final : public Component { * received. * @param qos The QoS of this subscription. */ - void subscribe_json(const std::string &topic, const mqtt_json_callback_t &callback, uint8_t qos = 0); + template void subscribe_json(std::string topic, F &&callback, uint8_t qos = 0) { + using DecayF = std::decay_t; + if constexpr (std::is_invocable_v) { + this->subscribe( + std::move(topic), + [cb = std::forward(callback)](const std::string &topic, const std::string &payload) { + call_json(cb, topic, payload); + }, + qos); + } else { + // A mutable callback keeps its state; the wrapper is then boxed rather than copied per call. + this->subscribe( + std::move(topic), + [cb = std::forward(callback)](const std::string &topic, const std::string &payload) mutable { + call_json(cb, topic, payload); + }, + qos); + } + } /** Unsubscribe from an MQTT topic. * * If multiple existing subscriptions to the same topic exist, all of them will be removed. + * Not allowed from inside a subscription callback; such a call is logged and ignored. * * @param topic The topic to unsubscribe from. * Must match the topic in the original subscribe or subscribe_json call exactly. @@ -285,6 +310,14 @@ class MQTTClientComponent final : public Component { bool subscribe_(const char *topic, uint8_t qos); void resubscribe_subscription_(MQTTSubscription *sub); + // parse_json runs its callback before returning, so the captures can be references. + template static void call_json(F &callback, const std::string &topic, const std::string &payload) { + json::parse_json(payload, [&topic, &callback](JsonObject root) -> bool { + callback(topic, root); + return true; + }); + } + void add_subscription_(std::string &&topic, mqtt_callback_t callback, bool boxed, uint8_t qos); void resubscribe_subscriptions_(); MQTTCredentials credentials_; @@ -336,48 +369,11 @@ class MQTTClientComponent final : public Component { bool publish_nan_as_none_{false}; bool wait_for_connection_{false}; + bool dispatching_{false}; }; extern MQTTClientComponent *global_mqtt_client; // NOLINT(cppcoreguidelines-avoid-non-const-global-variables) -class MQTTMessageTrigger final : public Trigger, public Component { - public: - explicit MQTTMessageTrigger(std::string topic); - - void set_qos(uint8_t qos) { this->qos_ = qos; } - void set_payload(const std::string &payload) { this->payload_ = payload; } - void setup() override; - void dump_config() override; - float get_setup_priority() const override; - - protected: - std::string topic_; - uint8_t qos_{0}; - optional payload_; -}; - -class MQTTJsonMessageTrigger final : public Trigger { - public: - explicit MQTTJsonMessageTrigger(const std::string &topic, uint8_t qos) { - global_mqtt_client->subscribe_json( - topic, [this](const std::string &topic, JsonObject root) { this->trigger(root); }, qos); - } -}; - -class MQTTConnectTrigger final : public Trigger { - public: - explicit MQTTConnectTrigger(MQTTClientComponent *client) { - client->set_on_connect([this](bool session_present) { this->trigger(session_present); }); - } -}; - -class MQTTDisconnectTrigger final : public Trigger { - public: - explicit MQTTDisconnectTrigger(MQTTClientComponent *client) { - client->set_on_disconnect([this](MQTTClientDisconnectReason reason) { this->trigger(reason); }); - } -}; - template class MQTTPublishJsonAction final : public Action { public: MQTTPublishJsonAction(MQTTClientComponent *parent) : parent_(parent) {} diff --git a/esphome/components/mqtt/mqtt_component.cpp b/esphome/components/mqtt/mqtt_component.cpp index adb5a95828..7e4b8be725 100644 --- a/esphome/components/mqtt/mqtt_component.cpp +++ b/esphome/components/mqtt/mqtt_component.cpp @@ -342,14 +342,6 @@ bool MQTTComponent::is_discovery_enabled() const { return this->discovery_enabled_ && global_mqtt_client->is_discovery_enabled(); } -void MQTTComponent::subscribe(const std::string &topic, mqtt_callback_t callback, uint8_t qos) { - global_mqtt_client->subscribe(topic, std::move(callback), qos); -} - -void MQTTComponent::subscribe_json(const std::string &topic, const mqtt_json_callback_t &callback, uint8_t qos) { - global_mqtt_client->subscribe_json(topic, callback, qos); -} - MQTTComponent::MQTTComponent() = default; float MQTTComponent::get_setup_priority() const { return setup_priority::AFTER_CONNECTION; } diff --git a/esphome/components/mqtt/mqtt_component.h b/esphome/components/mqtt/mqtt_component.h index 141b5f12b3..461db5d3e7 100644 --- a/esphome/components/mqtt/mqtt_component.h +++ b/esphome/components/mqtt/mqtt_component.h @@ -259,7 +259,9 @@ class MQTTComponent : public Component { * @param callback The callback that will be called when a message with matching topic is received. * @param qos The MQTT quality of service. Defaults to 0. */ - void subscribe(const std::string &topic, mqtt_callback_t callback, uint8_t qos = 0); + template void subscribe(std::string topic, F &&callback, uint8_t qos = 0) { + global_mqtt_client->subscribe(std::move(topic), std::forward(callback), qos); + } /** Subscribe to a MQTT topic and automatically parse JSON payload. * @@ -270,7 +272,9 @@ class MQTTComponent : public Component { * received. * @param qos The MQTT quality of service. Defaults to 0. */ - void subscribe_json(const std::string &topic, const mqtt_json_callback_t &callback, uint8_t qos = 0); + template void subscribe_json(std::string topic, F &&callback, uint8_t qos = 0) { + global_mqtt_client->subscribe_json(std::move(topic), std::forward(callback), qos); + } protected: static constexpr uint32_t SEND_DEFER_ID = 0; diff --git a/esphome/core/helpers.h b/esphome/core/helpers.h index b88a9d70e9..f1079f1855 100644 --- a/esphome/core/helpers.h +++ b/esphome/core/helpers.h @@ -1778,11 +1778,19 @@ template struct Callback { /// Invoke the callback. Only valid on Callbacks created via create(), never on default-constructed instances. void call(Ts... args) const { this->fn_(this->ctx_, std::forward(args)...); } + /// Whether create() stores F inline in the ctx pointer. The inline path invokes a copy, so a + /// callable that mutates its captures (a mutable lambda) is kept whole on the heap instead. + template static constexpr bool fits_inline() { + using DecayF = std::decay_t; + return sizeof(DecayF) <= sizeof(void *) && std::is_trivially_copyable_v && + std::is_invocable_v; + } + /// Create from any callable. Small trivially-copyable callables (like [this] lambdas) /// are stored inline in the ctx pointer without heap allocation. template static Callback create(F &&callable) { using DecayF = std::decay_t; - if constexpr (sizeof(DecayF) <= sizeof(void *) && std::is_trivially_copyable_v) { + if constexpr (fits_inline()) { // Small trivial callable (e.g. [this]() { this->method(); }) - store inline in ctx. // Safe under C++20 (P0593R6): byte copy into aligned storage implicitly // creates objects of implicit-lifetime types (trivially copyable qualifies). @@ -1805,6 +1813,28 @@ template struct Callback { return {[](void *c, Ts... args) { (*static_cast(c))(args...); }, static_cast(stored)}; } } + + /// Heap home of a callable stored by create_boxed(); the header knows how to free it. + struct Box { + void (*free)(Box *box); + }; + + /// Store any callable on the heap with a deleter, for an owner that can later free_boxed() it. + template static Callback create_boxed(F &&callable) { + struct Boxed : Box { + std::decay_t fn; + }; + auto *box = new Boxed{{[](Box *b) { delete static_cast(b); }}, // NOLINT(cppcoreguidelines-owning-memory) + std::forward(callable)}; + return {[](void *c, Ts... args) { static_cast(static_cast(c))->fn(args...); }, + static_cast(box)}; + } + + /// Free the callable of a Callback made by create_boxed(); the Callback must not be called afterwards. + void free_boxed() const { + auto *box = static_cast(this->ctx_); + box->free(box); + } }; /// Grow a CallbackManager's backing array to exactly size+1. Defined in helpers.cpp. diff --git a/tests/components/mqtt/common-triggers.yaml b/tests/components/mqtt/common-triggers.yaml new file mode 100644 index 0000000000..dadfc6f53b --- /dev/null +++ b/tests/components/mqtt/common-triggers.yaml @@ -0,0 +1,25 @@ +# on_message with a payload filter, a subscription callback larger than a pointer with its unsubscribe, +# and mutable callbacks, which keep their state instead of being copied per call. +mqtt: + id: mqtt_client + on_message: + - topic: livingroom/ota_mode + payload: "ON" + qos: 1 + then: + - logger.log: Got livingroom/ota_mode ON + +esphome: + on_boot: + - lambda: |- + std::string tag = "boxed"; + id(mqtt_client).subscribe("some/topic/boxed", [tag](const std::string &topic, const std::string &payload) { + ESP_LOGD("test", "%s %s", tag.c_str(), payload.c_str()); + }); + id(mqtt_client).unsubscribe("some/topic/boxed"); + id(mqtt_client).subscribe("some/topic/counted", [n = 0](const std::string &topic, const std::string &payload) mutable { + ESP_LOGD("test", "message %d: %s", ++n, payload.c_str()); + }); + id(mqtt_client).subscribe_json("some/topic/counted_json", [n = 0](const std::string &topic, JsonObject root) mutable { + ESP_LOGD("test", "json %d", ++n); + }); diff --git a/tests/components/mqtt/test-triggers.esp32-idf.yaml b/tests/components/mqtt/test-triggers.esp32-idf.yaml new file mode 100644 index 0000000000..b34863e873 --- /dev/null +++ b/tests/components/mqtt/test-triggers.esp32-idf.yaml @@ -0,0 +1,3 @@ +packages: + common: !include common.yaml + triggers: !include common-triggers.yaml