[core][mqtt] Build trigger callbacks as stateless lambdas

This commit is contained in:
J. Nick Koston
2026-09-30 00:16:07 +02:00
parent 7187acec04
commit ae7d96a2be
13 changed files with 625 additions and 195 deletions
+18
View File
@@ -384,6 +384,22 @@ file does, and it is the authority when they disagree. The most useful starting
)
```
When the parent callback's parameters are not the automation's arguments, or the trigger should only fire for some values, pass `params`, `forward` and `when`; the helper then generates a capture-less lambda instead of a forwarder (stored inline by `Callback` and `std::function`, nothing is allocated). `params` are the callback's parameters as `[(type, name)]`, `forward` the expressions passed to `trigger()` (default: the parameter names; name the parent with `automation.parent_ref(var)`), `when` a filter: a string, or an `ApplyCall` compared against config values that is skipped when its keys are absent. Compose `forward` from `MockObj` calls, not from f-strings of C++:
```python
# select: the callback carries the index, the automation also gets the option text
parent = automation.parent_ref(var)
index = cg.RawExpression("index")
await automation.build_callback_automation(
var,
"add_on_state_callback",
[(cg.StringRef, "x"), (cg.size_t, "i")],
conf,
params=[(cg.size_t, "index")],
forward=[cg.StringRef(parent.option_at(index)), index],
)
```
A callback with no arguments whose automation receives the parent is one line, `automation.build_parent_callback_automation(var, "add_on_state_callback", (Fan.operator("ptr"), "x"), conf)`. When the registration takes extra arguments (mqtt's topic and qos), `automation.build_trigger_callback(args, conf, params=..., forward=..., when=...)` returns the lambda for the component to register itself. Several callbacks on one parent go in a module-level `_CALLBACK_AUTOMATIONS` tuple of `automation.CallbackAutomation(conf_key, callback_method, args, ...)` entries, applied with `automation.build_callback_automations(var, config, _CALLBACK_AUTOMATIONS)`; each entry takes the same optional `forwarder`, `params`, `forward` and `when`.
**C++ -- no trigger class needed.** The callback registration method must be templatized to accept both `std::function` and lightweight forwarder structs (which avoid heap allocation):
```cpp
class MyComponent : public Component {
@@ -405,6 +421,8 @@ file does, and it is the authority when they disagree. The most useful starting
Use `build_automation()` with a `Trigger<Ts...>` subclass only when the forwarder needs **mutable state beyond a single `Automation*` pointer** (e.g. edge detection tracking previous state, timing logic).
Several such triggers on one parent go in a module-level `_TRIGGER_AUTOMATIONS` tuple of `(conf_key, args)` pairs applied with `automation.build_trigger_automations(var, config, _TRIGGER_AUTOMATIONS)`; it instantiates each entry's class from its `CONF_TRIGGER_ID` with `var` (or nothing when `var` is `None`) and builds the automations.
**Python:**
```python
TurnOnTrigger = my_ns.class_("TurnOnTrigger", automation.Trigger.template())
+154 -30
View File
File diff suppressed because it is too large Load Diff
+1
View File
@@ -38,6 +38,7 @@ from esphome.cpp_generator import ( # noqa: F401
new_variable,
process_lambda,
progmem_array,
progmem_string,
safe_exp,
set_cpp_standard,
statement,
+40 -47
View File
@@ -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,
@@ -118,18 +117,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_(
@@ -283,21 +270,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,
@@ -305,9 +281,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,
}
@@ -342,6 +315,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])
@@ -460,29 +445,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]))
+21 -49
View File
@@ -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
+43 -47
View File
@@ -31,19 +31,16 @@ namespace esphome::mqtt {
using mqtt_on_connect_callback_t = std::function<MQTTBackend::on_connect_callback_t>;
using mqtt_on_disconnect_callback_t = std::function<MQTTBackend::on_disconnect_callback_t>;
/** Callback for MQTT subscriptions.
*
* First parameter is the topic, the second one is the payload.
*/
using mqtt_callback_t = std::function<void(const std::string &, const std::string &)>;
using mqtt_json_callback_t = std::function<void(const std::string &, JsonObject)>;
/// Callback for MQTT subscriptions: the topic, then the payload.
using mqtt_callback_t = Callback<void(const std::string &, const std::string &)>;
/// 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<typename F> void subscribe(std::string topic, F &&callback, uint8_t qos = 0) {
if constexpr (mqtt_callback_t::fits_inline<F>()) {
this->add_subscription_(std::move(topic), mqtt_callback_t::create(std::forward<F>(callback)), false, qos);
} else {
this->add_subscription_(std::move(topic), mqtt_callback_t::create_boxed(std::forward<F>(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<typename F> void subscribe_json(std::string topic, F &&callback, uint8_t qos = 0) {
using DecayF = std::decay_t<F>;
if constexpr (std::is_invocable_v<const DecayF &, const std::string &, JsonObject>) {
this->subscribe(
std::move(topic),
[cb = std::forward<F>(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<F>(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<typename F> 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<std::string>, 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<std::string> payload_;
};
class MQTTJsonMessageTrigger final : public Trigger<JsonObjectConst> {
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<bool> {
public:
explicit MQTTConnectTrigger(MQTTClientComponent *client) {
client->set_on_connect([this](bool session_present) { this->trigger(session_present); });
}
};
class MQTTDisconnectTrigger final : public Trigger<MQTTClientDisconnectReason> {
public:
explicit MQTTDisconnectTrigger(MQTTClientComponent *client) {
client->set_on_disconnect([this](MQTTClientDisconnectReason reason) { this->trigger(reason); });
}
};
template<typename... Ts> class MQTTPublishJsonAction final : public Action<Ts...> {
public:
MQTTPublishJsonAction(MQTTClientComponent *parent) : parent_(parent) {}
@@ -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; }
+6 -2
View File
@@ -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<typename F> void subscribe(std::string topic, F &&callback, uint8_t qos = 0) {
global_mqtt_client->subscribe(std::move(topic), std::forward<F>(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<typename F> void subscribe_json(std::string topic, F &&callback, uint8_t qos = 0) {
global_mqtt_client->subscribe_json(std::move(topic), std::forward<F>(callback), qos);
}
protected:
/// Helper method to get the discovery topic for this component into a buffer.
+31 -1
View File
@@ -1719,11 +1719,19 @@ template<typename... Ts> struct Callback<void(Ts...)> {
/// 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<Ts>(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<typename F> static constexpr bool fits_inline() {
using DecayF = std::decay_t<F>;
return sizeof(DecayF) <= sizeof(void *) && std::is_trivially_copyable_v<DecayF> &&
std::is_invocable_v<const DecayF &, Ts...>;
}
/// Create from any callable. Small trivially-copyable callables (like [this] lambdas)
/// are stored inline in the ctx pointer without heap allocation.
template<typename F> static Callback create(F &&callable) {
using DecayF = std::decay_t<F>;
if constexpr (sizeof(DecayF) <= sizeof(void *) && std::is_trivially_copyable_v<DecayF>) {
if constexpr (fits_inline<F>()) {
// 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).
@@ -1746,6 +1754,28 @@ template<typename... Ts> struct Callback<void(Ts...)> {
return {[](void *c, Ts... args) { (*static_cast<DecayF *>(c))(args...); }, static_cast<void *>(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<typename F> static Callback create_boxed(F &&callable) {
struct Boxed : Box {
std::decay_t<F> fn;
};
auto *box = new Boxed{{[](Box *b) { delete static_cast<Boxed *>(b); }}, // NOLINT(cppcoreguidelines-owning-memory)
std::forward<F>(callable)};
return {[](void *c, Ts... args) { static_cast<Boxed *>(static_cast<Box *>(c))->fn(args...); },
static_cast<Box *>(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<Box *>(this->ctx_);
box->free(box);
}
};
/// Grow a CallbackManager's backing array to exactly size+1. Defined in helpers.cpp.
+7
View File
@@ -264,6 +264,13 @@ class FlashStringLiteral(Literal):
return f"ESPHOME_F({cpp_string_escape(self.string)})"
def progmem_string(value: str) -> Expression:
"""A ``std::string`` argument from a literal that stays in flash on ESP8266."""
if CORE.is_esp8266:
return RawExpression(f"progmem_string({FlashStringLiteral(value)})")
return safe_exp(value)
class IntLiteral(Literal):
__slots__ = ("i",)
@@ -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);
});
@@ -0,0 +1,3 @@
packages:
common: !include common.yaml
triggers: !include common-triggers.yaml
File diff suppressed because it is too large Load Diff