github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-10-18 12:12:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

Grow the AsyncMqttClient topic buffer at connect instead of rejecting long topics

Browse Source
This commit is contained in:
J. Nick Koston
2026-10-14 10:37:26 -10:00
parent 555db9d1ca
commit fb52e1679d
6 changed files with 27 additions and 250 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/components/mqtt/__init__.py
-110
View File
@@ -1,5 +1,3 @@
from collections.abc import Iterator
from esphome import automation
import esphome.codegen as cg
from esphome.components import logger, socket
@@ -25,7 +23,6 @@ from esphome.const import (
CONF_CLIENT_ID,
CONF_COMMAND_RETAIN,
CONF_COMMAND_TOPIC,
CONF_DEVICE_ID,
CONF_DISCOVER_IP,
CONF_DISCOVERY,
CONF_DISCOVERY_OBJECT_ID_GENERATOR,
@@ -33,17 +30,10 @@ from esphome.const import (
CONF_DISCOVERY_RETAIN,
CONF_DISCOVERY_UNIQUE_ID_GENERATOR,
CONF_ENABLE_ON_BOOT,
CONF_ESPHOME,
CONF_ID,
CONF_INTERNAL,
CONF_KEEPALIVE,
CONF_LEVEL,
CONF_LOG_TOPIC,
CONF_MODE_COMMAND_TOPIC,
CONF_MQTT as CONF_MQTT,
CONF_MQTT_ID,
CONF_NAME,
CONF_NAME_ADD_MAC_SUFFIX,
CONF_ON_CONNECT,
CONF_ON_DISCONNECT,
CONF_ON_JSON_MESSAGE,
@@ -52,7 +42,6 @@ from esphome.const import (
CONF_PAYLOAD,
CONF_PAYLOAD_AVAILABLE,
CONF_PAYLOAD_NOT_AVAILABLE,
CONF_PLATFORM,
CONF_PORT,
CONF_PUBLISH_NAN_AS_NONE,
CONF_QOS,
@@ -64,7 +53,6 @@ from esphome.const import (
CONF_SUBSCRIBE_QOS,
CONF_TOPIC,
CONF_TOPIC_PREFIX,
CONF_TYPE,
CONF_USE_ABBREVIATIONS,
CONF_USERNAME,
CONF_WILL_MESSAGE,
@@ -76,8 +64,6 @@ from esphome.const import (
PlatformFramework,
)
from esphome.core import CORE, CoroPriority, coroutine_with_priority
from esphome.core.entity_helpers import get_base_entity_object_id
import esphome.final_validate as fv
from esphome.types import ConfigType
DEPENDENCIES = ["network"]
@@ -322,102 +308,6 @@ CONFIG_SCHEMA = cv.All(
_consume_mqtt_sockets,
)
# AsyncMqttClient, the ESP8266 and LibreTiny backend, drops any inbound message whose
# topic is longer than this (its default maxTopicLength)
ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN = 128
# name_add_mac_suffix appends "-" plus 6 hex digits to the name at runtime
MAC_SUFFIX = "-xxxxxx"
# Entity domains whose MQTT component always subscribes to its command topic
_COMMAND_TOPIC_DOMAINS = (
"alarm_control_panel",
"button",
"climate",
"cover",
"datetime",
"fan",
"light",
"lock",
"number",
"select",
"switch",
"text",
"update",
"valve",
)
def _entity_command_topic(
domain: str, config: ConfigType, topic_prefix: str, mac_suffix: bool
) -> str | None:
"""Return the topic an entity's MQTT component always subscribes to, if known."""
if domain == "climate":
suffix, custom_key = "mode/command", CONF_MODE_COMMAND_TOPIC
else:
suffix, custom_key = "command", CONF_COMMAND_TOPIC
if (custom := config.get(custom_key)) is not None:
# A lambda topic is only known at runtime
return custom if isinstance(custom, str) else None
# Mirrors MQTTComponent::compute_is_internal_(): internal components never subscribe
if CONF_STATE_TOPIC in config:
internal = config[CONF_STATE_TOPIC] == ""
else:
internal = config.get(CONF_INTERNAL, False)
if internal:
return None
device = config.get(CONF_DEVICE_ID)
object_id = get_base_entity_object_id(
config.get(CONF_NAME) or "", CORE.friendly_name, device.id if device else None
)
if mac_suffix and not config.get(CONF_NAME):
object_id += MAC_SUFFIX
component_type = config[CONF_TYPE].lower() if domain == "datetime" else domain
return f"{topic_prefix}/{component_type}/{object_id}/{suffix}"
def _subscribed_topics(config: ConfigType) -> Iterator[str]:
"""Yield every subscribed topic that is known at build time."""
full_config = fv.full_config.get()
mac_suffix = full_config[CONF_ESPHOME].get(CONF_NAME_ADD_MAC_SUFFIX, False)
if topic_prefix := config[CONF_TOPIC_PREFIX]:
if mac_suffix and topic_prefix == CORE.name:
topic_prefix += MAC_SUFFIX
for domain in _COMMAND_TOPIC_DOMAINS:
for entity in full_config.get(domain) or []:
if CONF_MQTT_ID in entity and (
topic := _entity_command_topic(
domain, entity, topic_prefix, mac_suffix
)
):
yield topic
for key in (CONF_ON_MESSAGE, CONF_ON_JSON_MESSAGE):
for trigger in config.get(key, []):
yield trigger[CONF_TOPIC]
for domain in ("sensor", "text_sensor"):
for entity in full_config.get(domain) or []:
if entity[CONF_PLATFORM] == "mqtt_subscribe":
yield entity[CONF_TOPIC]
def _validate_topic_lengths(config: ConfigType) -> ConfigType:
"""Reject topics that the ESP8266 and LibreTiny MQTT client would silently drop."""
if not (CORE.is_esp8266 or CORE.is_libretiny):
return config
for topic in _subscribed_topics(config):
# A wildcard filter matches topics of any length, so only literals are checked
if (
len(topic) > ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN
and "+" not in topic
and "#" not in topic
):
raise cv.Invalid(
f"MQTT topic '{topic}' is {len(topic)} characters; the MQTT client on "
f"this platform drops topics longer than {ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN}"
)
return config
FINAL_VALIDATE_SCHEMA = _validate_topic_lengths
def exp_mqtt_message(config):
if config is None:
esphome/components/mqtt/mqtt_backend_esp8266.h
+2
View File
@@ -49,6 +49,8 @@ class MQTTBackendESP8266 final : public MQTTBackend {
bool connected() const final { return mqtt_client_.connected(); }
void connect() final { mqtt_client_.connect(); }
/// Only call while disconnected: the parser writes inbound topics into this buffer.
void set_max_topic_length(uint16_t length) { mqtt_client_.setMaxTopicLength(length); }
void disconnect() final { mqtt_client_.disconnect(true); }
bool subscribe(const char *topic, uint8_t qos) final { return mqtt_client_.subscribe(topic, qos) != 0; }
bool unsubscribe(const char *topic) final { return mqtt_client_.unsubscribe(topic) != 0; }
esphome/components/mqtt/mqtt_backend_libretiny.h
+2
View File
@@ -49,6 +49,8 @@ class MQTTBackendLibreTiny final : public MQTTBackend {
bool connected() const final { return mqtt_client_.connected(); }
void connect() final { mqtt_client_.connect(); }
/// Only call while disconnected: the parser writes inbound topics into this buffer.
void set_max_topic_length(uint16_t length) { mqtt_client_.setMaxTopicLength(length); }
void disconnect() final { mqtt_client_.disconnect(true); }
bool subscribe(const char *topic, uint8_t qos) final { return mqtt_client_.subscribe(topic, qos) != 0; }
bool unsubscribe(const char *topic) final { return mqtt_client_.unsubscribe(topic) != 0; }
esphome/components/mqtt/mqtt_client.cpp
+19
View File
@@ -2,6 +2,7 @@
#ifdef USE_MQTT
#include <algorithm>
#include <utility>
#include "esphome/components/network/util.h"
#include "esphome/core/application.h"
@@ -312,6 +313,17 @@ void MQTTClientComponent::start_connect_() {
// Force disconnect first
this->mqtt_backend_.disconnect();
#if defined(USE_ESP8266) || defined(USE_LIBRETINY)
// Grow the topic buffer to the longest subscription. Only here, while no connection parses into it.
size_t longest = this->max_topic_length_;
for (const auto &subscription : this->subscriptions_)
longest = std::max(longest, subscription.topic.size());
if (longest > this->max_topic_length_ && longest <= UINT16_MAX) {
this->max_topic_length_ = longest;
this->mqtt_backend_.set_max_topic_length(longest);
}
#endif
this->mqtt_backend_.set_client_id(this->credentials_.client_id.c_str());
this->mqtt_backend_.set_clean_session(this->credentials_.clean_session);
const char *username = nullptr;
@@ -472,6 +484,13 @@ void MQTTClientComponent::add_subscription_(std::string &&topic, mqtt_subscripti
.callback = callback,
.resubscribe_timeout = 0,
});
#if defined(USE_ESP8266) || defined(USE_LIBRETINY)
if (subscription.topic.size() > this->max_topic_length_ &&
(this->state_ == MQTT_CLIENT_CONNECTING || this->state_ == MQTT_CLIENT_CONNECTED)) {
ESP_LOGW(TAG, "Topic '%s' is longer than %u characters; its messages arrive after the next reconnect",
subscription.topic.c_str(), this->max_topic_length_);
}
#endif
this->resubscribe_subscription_(&subscription);
}
esphome/components/mqtt/mqtt_client.h
+4
View File
@@ -370,6 +370,10 @@ class MQTTClientComponent final : public Component {
bool wait_for_connection_{false};
bool dispatching_{false};
bool removed_while_dispatching_{false};
#if defined(USE_ESP8266) || defined(USE_LIBRETINY)
// AsyncMqttClient drops inbound topics longer than this; it starts at the library's default
uint16_t max_topic_length_{128};
#endif
};
extern MQTTClientComponent *global_mqtt_client; // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
tests/unit_tests/components/test_mqtt_topic_length.py
-140
View File
@@ -1,140 +0,0 @@
"""Tests for the mqtt check that subscribed topics fit the AsyncMqttClient limit."""
import pytest
from esphome.components.mqtt import (
ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN,
_validate_topic_lengths,
)
import esphome.config_validation as cv
from esphome.const import (
CONF_ESPHOME,
CONF_MQTT_ID,
CONF_NAME,
CONF_NAME_ADD_MAC_SUFFIX,
KEY_CORE,
KEY_TARGET_PLATFORM,
PLATFORM_BK72XX,
PLATFORM_ESP32,
PLATFORM_ESP8266,
)
from esphome.core import CORE, Lambda
import esphome.final_validate as fv
from esphome.types import ConfigType
# "dev/switch/<object_id>/command" is 19 characters around the object id
FITS = "a" * (ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN - 19)
TOO_LONG = FITS + "a"
LONG_LITERAL = "t" * (ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN + 1)
def _switch(name: str, **extra) -> dict:
return {"switch": [{CONF_NAME: name, CONF_MQTT_ID: "mqtt_switch", **extra}]}
def _run(
monkeypatch: pytest.MonkeyPatch,
full_config: dict,
mqtt: ConfigType | None = None,
platform: str = PLATFORM_ESP8266,
) -> ConfigType:
monkeypatch.setitem(CORE.data, KEY_CORE, {KEY_TARGET_PLATFORM: platform})
monkeypatch.setattr(CORE, "name", "dev")
monkeypatch.setattr(CORE, "friendly_name", None)
config = {"topic_prefix": "dev", **(mqtt or {})}
token = fv.full_config.set({CONF_ESPHOME: {}, **full_config})
try:
return _validate_topic_lengths(config)
finally:
fv.full_config.reset(token)
@pytest.mark.parametrize(
("full_config", "mqtt", "platform"),
[
pytest.param(_switch(TOO_LONG), None, PLATFORM_ESP8266, id="command_topic"),
pytest.param(_switch(TOO_LONG), None, PLATFORM_BK72XX, id="libretiny"),
pytest.param(
{CONF_ESPHOME: {CONF_NAME_ADD_MAC_SUFFIX: True}, **_switch(FITS)},
None,
PLATFORM_ESP8266,
id="mac_suffix",
),
pytest.param(
_switch("s", command_topic=LONG_LITERAL),
None,
PLATFORM_ESP8266,
id="custom_command_topic",
),
pytest.param(
{
"climate": [
# "dev/climate/<object_id>/mode/command" is 25 characters around it
{CONF_NAME: "a" * 104, CONF_MQTT_ID: "mqtt_climate"}
]
},
None,
PLATFORM_ESP8266,
id="climate_mode_command",
),
pytest.param(
{},
{"on_message": [{"topic": LONG_LITERAL}]},
PLATFORM_ESP8266,
id="trigger",
),
pytest.param(
{"sensor": [{"platform": "mqtt_subscribe", "topic": LONG_LITERAL}]},
None,
PLATFORM_ESP8266,
id="mqtt_subscribe",
),
],
)
def test_rejected(
monkeypatch: pytest.MonkeyPatch, full_config: dict, mqtt: dict | None, platform: str
) -> None:
with pytest.raises(cv.Invalid, match="characters"):
_run(monkeypatch, full_config, mqtt, platform)
@pytest.mark.parametrize(
("full_config", "mqtt", "platform"),
[
pytest.param(_switch(FITS), None, PLATFORM_ESP8266, id="at_the_limit"),
pytest.param(_switch(TOO_LONG), None, PLATFORM_ESP32, id="esp32"),
pytest.param(
_switch(TOO_LONG, command_topic="short/cmd"),
None,
PLATFORM_ESP8266,
id="short_custom_topic",
),
pytest.param(
_switch(TOO_LONG, command_topic=Lambda('return "x";')),
None,
PLATFORM_ESP8266,
id="lambda_topic",
),
pytest.param(
_switch(TOO_LONG, internal=True), None, PLATFORM_ESP8266, id="internal"
),
pytest.param(
{"switch": [{CONF_NAME: TOO_LONG}]}, None, PLATFORM_ESP8266, id="no_mqtt"
),
pytest.param(
{},
{"on_message": [{"topic": LONG_LITERAL + "/#"}]},
PLATFORM_ESP8266,
id="wildcard",
),
],
)
def test_accepted(
monkeypatch: pytest.MonkeyPatch, full_config: dict, mqtt: dict | None, platform: str
) -> None:
_run(monkeypatch, full_config, mqtt, platform)
def test_message_names_the_topic_and_length(monkeypatch: pytest.MonkeyPatch) -> None:
with pytest.raises(cv.Invalid, match=rf"'dev/switch/{TOO_LONG}/command' is 129"):
_run(monkeypatch, _switch(TOO_LONG))