mqtt_client.cpp

Go to the documentation of this file.

#include "mqtt_client.h"
 
#ifdef USE_MQTT
 
#include <utility>
#include "esphome/components/network/util.h"
#include "esphome/core/application.h"
#include "esphome/core/entity_base.h"
#include "esphome/core/helpers.h"
#include "esphome/core/log.h"
#include "esphome/core/progmem.h"
#include "esphome/core/version.h"
#ifdef USE_LOGGER
#include "esphome/components/logger/logger.h"
#endif
#include "lwip/dns.h"
#include "lwip/err.h"
#include "mqtt_component.h"
#include "mqtt_topic_match.h"
 
#ifdef USE_API
#include "esphome/components/api/api_server.h"
#endif
#ifdef USE_DASHBOARD_IMPORT
#include "esphome/components/dashboard_import/dashboard_import.h"
#endif
 
namespace esphome::mqtt {
 
ESPHOME_LOG_TAG(TAG, "mqtt");
 
// Maximum number of MQTT component resends per loop iteration.
// Limits work to avoid triggering the task watchdog on reconnect.
static constexpr uint8_t MAX_RESENDS_PER_LOOP = 8;
 
// Disconnect reason strings indexed by MQTTClientDisconnectReason enum (0-8)
PROGMEM_STRING_TABLE(MQTTDisconnectReasonStrings, "TCP disconnected", "Unacceptable Protocol Version",
                     "Identifier Rejected", "Server Unavailable", "Malformed Credentials", "Not Authorized",
                     "Not Enough Space", "TLS Bad Fingerprint", "DNS Resolve Error", "Unknown");
 
MQTTClientComponent::MQTTClientComponent() {
  global_mqtt_client = this;
  char mac_addr[MAC_ADDRESS_BUFFER_SIZE];
  get_mac_address_into_buffer(mac_addr);
  const StringRef &name = App.get_name();
  char client_id[MAX_NAME_WITH_SUFFIX_SIZE];
  size_t len = make_name_with_suffix_to(client_id, sizeof(client_id), name.c_str(), name.size(), '-', mac_addr,
                                        MAC_ADDRESS_BUFFER_SIZE - 1);
  this->credentials_.client_id.assign(client_id, len);
}
 
// Connection
// Room for "esphome/<kind>/<name>" with the longest kind, "discover"
static constexpr size_t ESPHOME_TOPIC_BUFFER_SIZE = sizeof("esphome/discover/") + ESPHOME_DEVICE_NAME_MAX_LEN;
 
static void format_esphome_topic(char (&topic)[ESPHOME_TOPIC_BUFFER_SIZE], const char *kind) {
  buf_append_printf(topic, sizeof(topic), 0, "esphome/%s/%s", kind, App.get_name().c_str());
}
 
void MQTTClientComponent::setup() {
  this->mqtt_backend_.set_on_message(
      [this](const char *topic, const char *payload, size_t len, size_t index, size_t total) {
        if (index == 0) {
          this->payload_buffer_.clear();
          this->payload_buffer_.reserve(total);
        }
 
        // append new payload, may contain incomplete MQTT message
        this->payload_buffer_.append(payload, len);
 
        // MQTT fully received
        if (len + index == total) {
          this->on_message(topic, this->payload_buffer_);
          this->payload_buffer_.clear();
        }
      });
  this->mqtt_backend_.set_on_disconnect([this](MQTTClientDisconnectReason reason) {
    if (this->state_ == MQTT_CLIENT_DISABLED)
      return;
    this->state_ = MQTT_CLIENT_DISCONNECTED;
    this->disconnect_reason_ = reason;
  });
#ifdef USE_LOGGER
  if (this->is_log_message_enabled() && logger::global_logger != nullptr) {
    logger::global_logger->add_log_callback(
        this, [](void *self, uint8_t level, const char *tag, const char *message, size_t message_len) {
          static_cast<MQTTClientComponent *>(self)->on_log(level, tag, message, message_len);
        });
#ifdef USE_MQTT_CRASH_LOG
    // `esphome logs` announces itself here, like an API client's log subscription, and the record stays
    // loggable until the next boot, so the offer does too. A log topic that drops errors would swallow the
    // replay, so it is not offered. The topic is rebuilt from the device name when needed, so nothing is
    // stored for it.
    this->crash_log_offered_ = logger::crash_log_pending() && this->log_level_ >= ESPHOME_LOG_LEVEL_ERROR;
#endif
  }
#endif
 
  if (this->is_discovery_ip_enabled()) {
    this->subscribe(
        static_topic("esphome/discover"),
        [this](const std::string &topic, const std::string &payload) { this->send_device_info_(); }, 2);
 
    static_assert(sizeof("esphome/ping/") + ESPHOME_DEVICE_NAME_MAX_LEN <= MQTT_DEFAULT_TOPIC_MAX_LEN);
    MQTTTopicSource ping_topic{[](const void *, std::span<char, MQTT_DEFAULT_TOPIC_MAX_LEN> buf) {
                                 size_t len = buf_append_printf(buf.data(), buf.size(), 0, "esphome/ping/%s",
                                                                App.get_name().c_str());
                                 return StringRef(buf.data(), len);
                               },
                               nullptr};
    this->subscribe(
        ping_topic, [this](const std::string &topic, const std::string &payload) { this->send_device_info_(); }, 2);
  }
 
  if (this->enable_on_boot_) {
    this->enable();
  }
}
 
void MQTTClientComponent::send_device_info_() {
  if (!this->is_connected() or !this->is_discovery_ip_enabled()) {
    return;
  }
  // Format topic on stack to avoid heap allocation
  char topic[ESPHOME_TOPIC_BUFFER_SIZE];
  format_esphome_topic(topic, "discover");
 
  // NOLINTBEGIN(clang-analyzer-cplusplus.NewDeleteLeaks) false positive with ArduinoJson
  this->publish_json(
      topic,
      [](JsonObject root) {
        uint8_t index = 0;
        for (auto &ip : network::get_ip_addresses()) {
          if (ip.is_set()) {
            char key[8];  // "ip" + up to 3 digits + null
            char ip_buf[network::IP_ADDRESS_BUFFER_SIZE];
            if (index == 0) {
              key[0] = 'i';
              key[1] = 'p';
              key[2] = '\0';
            } else {
              buf_append_printf(key, sizeof(key), 0, "ip%u", index);
            }
            ip.str_to(ip_buf);
            root[key] = ip_buf;
            index++;
          }
        }
        root[ESPHOME_F("name")] = App.get_name();
        if (!App.get_friendly_name().empty()) {
          root[ESPHOME_F("friendly_name")] = App.get_friendly_name();
        }
#ifdef USE_API
        root[ESPHOME_F("port")] = api::global_api_server->get_port();
#endif
        root[ESPHOME_F("version")] = ESPHOME_VERSION;
        char mac_buf[MAC_ADDRESS_BUFFER_SIZE];
        get_mac_address_into_buffer(mac_buf);
        root[ESPHOME_F("mac")] = mac_buf;
 
#ifdef USE_ESP8266
        root[ESPHOME_F("platform")] = ESPHOME_F("ESP8266");
#endif
#ifdef USE_ESP32
        root[ESPHOME_F("platform")] = ESPHOME_F("ESP32");
#endif
#ifdef USE_LIBRETINY
        root[ESPHOME_F("platform")] = lt_cpu_get_model_name();
#endif
 
        root[ESPHOME_F("board")] = ESPHOME_BOARD;
#if defined(USE_WIFI)
        root[ESPHOME_F("network")] = ESPHOME_F("wifi");
#elif defined(USE_ETHERNET)
        root[ESPHOME_F("network")] = ESPHOME_F("ethernet");
#endif
 
#ifdef ESPHOME_PROJECT_NAME
        root[ESPHOME_F("project_name")] = ESPHOME_PROJECT_NAME;
        root[ESPHOME_F("project_version")] = ESPHOME_PROJECT_VERSION;
#endif  // ESPHOME_PROJECT_NAME
 
#ifdef USE_DASHBOARD_IMPORT
        root[ESPHOME_F("package_import_url")] = dashboard_import::get_package_import_url();
#endif
 
#ifdef USE_API_NOISE
        root[api::global_api_server->get_noise_ctx().has_psk() ? ESPHOME_F("api_encryption")
                                                               : ESPHOME_F("api_encryption_supported")] =
            ESPHOME_F("Noise_NNpsk0_25519_ChaChaPoly_SHA256");
#endif
      },
      2, this->discovery_info_.retain);
  // NOLINTEND(clang-analyzer-cplusplus.NewDeleteLeaks)
}
 
#ifdef USE_LOGGER
void MQTTClientComponent::on_log(uint8_t level, const char *tag, const char *message, size_t message_len) {
  (void) tag;
  if (level <= this->log_level_ && this->is_connected()) {
    this->publish(this->log_message_.topic.c_str(), message, message_len, this->log_message_.qos,
                  this->log_message_.retain);
  }
}
#endif
 
void MQTTClientComponent::dump_config() {
  char ip_buf[network::IP_ADDRESS_BUFFER_SIZE];
  // clang-format off
  ESP_LOGCONFIG(TAG,
                "MQTT:\n"
                "  Server Address: %s:%u (%s)\n"
                "  Username: " LOG_SECRET("'%s'") "\n"
                "  Client ID: " LOG_SECRET("'%s'") "\n"
                "  Clean Session: %s",
                this->credentials_.address.c_str(), this->credentials_.port, this->ip_.str_to(ip_buf),
                this->credentials_.username.c_str(), this->credentials_.client_id.c_str(),
                YESNO(this->credentials_.clean_session));
  // clang-format on
#ifdef USE_ESP32
  if (const MQTTTransport transport = this->mqtt_backend_.get_transport();
      transport != MQTTTransport::MQTT_TRANSPORT_TCP) {
    ESP_LOGCONFIG(TAG,
                  "  Transport: %s\n"
                  "  WebSocket Path: '%s'",
                  transport == MQTTTransport::MQTT_TRANSPORT_WSS ? LOG_STR_LITERAL("wss") : LOG_STR_LITERAL("ws"),
                  this->mqtt_backend_.get_ws_path());
  }
#endif
#if defined(USE_MQTT_CERT_BUNDLE)
  ESP_LOGCONFIG(TAG, "  TLS: certificate bundle");
#elif defined(USE_MQTT_TLS_INSECURE)
  ESP_LOGCONFIG(TAG, "  TLS: no verification");
#elif defined(USE_MQTT_TLS)
  ESP_LOGCONFIG(TAG, "  TLS: CA certificate");
#endif
  if (this->is_discovery_ip_enabled()) {
    ESP_LOGCONFIG(TAG, "  Discovery IP enabled");
  }
  if (!this->discovery_info_.prefix.empty()) {
    ESP_LOGCONFIG(TAG,
                  "  Discovery prefix: '%s'\n"
                  "  Discovery retain: %s",
                  this->discovery_info_.prefix.c_str(), YESNO(this->discovery_info_.retain));
  }
  ESP_LOGCONFIG(TAG, "  Topic Prefix: '%s'", this->topic_prefix_.c_str());
  if (!this->log_message_.topic.empty()) {
    ESP_LOGCONFIG(TAG, "  Log Topic: '%s'", this->log_message_.topic.c_str());
  }
  if (!this->availability_.topic.empty()) {
    ESP_LOGCONFIG(TAG, "  Availability: '%s'", this->availability_.topic.c_str());
  }
}
bool MQTTClientComponent::can_proceed() {
  return network::is_disabled() || this->state_ == MQTT_CLIENT_DISABLED || this->is_connected() ||
         !this->wait_for_connection_;
}
 
void MQTTClientComponent::start_dnslookup_() {
  for (auto &subscription : this->subscriptions_) {
    subscription.subscribed = false;
    subscription.resubscribe_timeout = 0;
  }
 
  this->status_set_warning();
  this->dns_resolve_error_ = false;
  this->dns_resolved_ = false;
  ip_addr_t addr;
  err_t err;
  {
    LwIPLock lock;
#if USE_NETWORK_IPV6
    err = dns_gethostbyname_addrtype(this->credentials_.address.c_str(), &addr, MQTTClientComponent::dns_found_callback,
                                     this, LWIP_DNS_ADDRTYPE_IPV6_IPV4);
#else
    err = dns_gethostbyname_addrtype(this->credentials_.address.c_str(), &addr, MQTTClientComponent::dns_found_callback,
                                     this, LWIP_DNS_ADDRTYPE_IPV4);
#endif /* USE_NETWORK_IPV6 */
  }
  switch (err) {
    case ERR_OK: {
      // Got IP immediately
      this->dns_resolved_ = true;
      this->ip_ = network::IPAddress(&addr);
      this->start_connect_();
      return;
    }
    case ERR_INPROGRESS: {
      // wait for callback
      ESP_LOGD(TAG, "Resolving broker IP address");
      break;
    }
    default:
    case ERR_ARG: {
      // error
      ESP_LOGW(TAG, "Error resolving broker IP address: %d", err);
      break;
    }
  }
 
  this->state_ = MQTT_CLIENT_RESOLVING_ADDRESS;
  this->connect_begin_ = millis();
}
void MQTTClientComponent::check_dnslookup_() {
  if (!this->dns_resolved_ && millis() - this->connect_begin_ > 20000) {
    this->dns_resolve_error_ = true;
  }
 
  if (this->dns_resolve_error_) {
    ESP_LOGW(TAG, "Couldn't resolve IP address for '%s'", this->credentials_.address.c_str());
    this->state_ = MQTT_CLIENT_DISCONNECTED;
    this->disconnect_reason_ = MQTTClientDisconnectReason::DNS_RESOLVE_ERROR;
    this->on_disconnect_.call(MQTTClientDisconnectReason::DNS_RESOLVE_ERROR);
    return;
  }
 
  if (!this->dns_resolved_) {
    return;
  }
 
  char ip_buf[network::IP_ADDRESS_BUFFER_SIZE];
  ESP_LOGD(TAG, "Resolved broker IP address to %s", this->ip_.str_to(ip_buf));
  this->start_connect_();
}
#if defined(USE_ESP8266) && LWIP_VERSION_MAJOR == 1
void MQTTClientComponent::dns_found_callback(const char *name, ip_addr_t *ipaddr, void *callback_arg) {
#else
void MQTTClientComponent::dns_found_callback(const char *name, const ip_addr_t *ipaddr, void *callback_arg) {
#endif
  auto *a_this = (MQTTClientComponent *) callback_arg;
  if (ipaddr == nullptr) {
    a_this->dns_resolve_error_ = true;
  } else {
    a_this->ip_ = network::IPAddress(ipaddr);
    a_this->dns_resolved_ = true;
  }
}
 
void MQTTClientComponent::start_connect_() {
  if (!network::is_connected())
    return;
 
  ESP_LOGI(TAG, "Connecting");
  // Force disconnect first
  this->mqtt_backend_.disconnect();
 
  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;
  if (!this->credentials_.username.empty())
    username = this->credentials_.username.c_str();
  const char *password = nullptr;
  if (!this->credentials_.password.empty())
    password = this->credentials_.password.c_str();
 
  this->mqtt_backend_.set_credentials(username, password);
 
  this->mqtt_backend_.set_server(this->credentials_.address.c_str(), this->credentials_.port);
  if (!this->last_will_.topic.empty()) {
    this->mqtt_backend_.set_will(this->last_will_.topic.c_str(), this->last_will_.qos, this->last_will_.retain,
                                 this->last_will_.payload.c_str());
  }
 
  this->mqtt_backend_.connect();
  this->state_ = MQTT_CLIENT_CONNECTING;
  this->connect_begin_ = millis();
}
bool MQTTClientComponent::is_connected() {
  return this->state_ == MQTT_CLIENT_CONNECTED && this->mqtt_backend_.connected();
}
 
void MQTTClientComponent::check_connected() {
  if (!this->mqtt_backend_.connected()) {
    if (millis() - this->connect_begin_ > 60000) {
      this->state_ = MQTT_CLIENT_DISCONNECTED;
      this->start_dnslookup_();
    }
    return;
  }
 
  this->state_ = MQTT_CLIENT_CONNECTED;
  this->sent_birth_message_ = false;
  this->status_clear_warning();
  ESP_LOGI(TAG, "Connected");
  // MQTT Client needs some time to be fully set up.
  delay(100);  // NOLINT
 
  this->resubscribe_subscriptions_();
#ifdef USE_MQTT_CRASH_LOG
  // Best effort: a refused subscription is tried again at the next connection
  if (this->crash_log_offered_) {
    char topic[ESPHOME_TOPIC_BUFFER_SIZE];
    format_esphome_topic(topic, "logs");
    this->subscribe_(topic, 0);
  }
#endif
  this->send_device_info_();
 
  for (MQTTComponent *component : this->children_)
    component->schedule_resend_state();
}
 
void MQTTClientComponent::loop() {
  // Call the backend loop first
  mqtt_backend_.loop();
 
  if (this->disconnect_reason_.has_value()) {
    const LogString *reason_s = MQTTDisconnectReasonStrings::get_log_str(
        static_cast<uint8_t>(*this->disconnect_reason_), MQTTDisconnectReasonStrings::LAST_INDEX);
    if (!network::is_connected()) {
      reason_s = LOG_STR("WiFi disconnected");
    }
    ESP_LOGW(TAG, "Disconnected: %s", LOG_STR_ARG(reason_s));
    this->disconnect_reason_.reset();
  }
 
  const uint32_t now = App.get_loop_component_start_time();
 
  switch (this->state_) {
    case MQTT_CLIENT_DISABLED:
      return;  // Return to avoid a reboot when disabled
    case MQTT_CLIENT_DISCONNECTED:
      if (now - this->connect_begin_ > 5000) {
        this->start_dnslookup_();
      }
      break;
    case MQTT_CLIENT_RESOLVING_ADDRESS:
      this->check_dnslookup_();
      break;
    case MQTT_CLIENT_CONNECTING:
      this->check_connected();
      break;
    case MQTT_CLIENT_CONNECTED:
      if (!this->mqtt_backend_.connected()) {
        this->state_ = MQTT_CLIENT_DISCONNECTED;
        ESP_LOGW(TAG, "Lost client connection");
        this->start_dnslookup_();
      } else {
        if (!this->birth_message_.topic.empty() && !this->sent_birth_message_) {
          this->sent_birth_message_ = this->publish(this->birth_message_);
        }
 
        this->last_connected_ = now;
        this->resubscribe_subscriptions_();
 
        // Process pending resends for all MQTT components centrally
        // Limit work per loop iteration to avoid triggering task WDT on reconnect
        {
          uint8_t resend_count = 0;
          for (MQTTComponent *component : this->children_) {
            if (component->is_resend_pending()) {
              component->process_resend();
              if (++resend_count >= MAX_RESENDS_PER_LOOP)
                break;
            }
          }
        }
      }
      break;
  }
 
  if (millis() - this->last_connected_ > this->reboot_timeout_ && this->reboot_timeout_ != 0) {
    ESP_LOGE(TAG, "Can't connect; restarting");
    App.reboot();
  }
}
float MQTTClientComponent::get_setup_priority() const { return setup_priority::AFTER_WIFI; }
 
// Subscribe
bool MQTTClientComponent::subscribe_(const char *topic, uint8_t qos) {
  if (!this->is_connected())
    return false;
 
  bool ret = this->mqtt_backend_.subscribe(topic, qos);
  yield();
 
  if (ret) {
    ESP_LOGV(TAG, "subscribe(topic='%s')", topic);
  } else {
    delay(5);
    ESP_LOGV(TAG, "Subscribe failed for topic='%s'. Will retry", topic);
    this->status_momentary_warning(1000);
  }
  return ret != 0;
}
// Out of line so the topic buffer is not part of loop()'s stack frame
__attribute__((noinline)) bool MQTTClientComponent::subscribe_topic_(const MQTTTopicSource &topic, uint8_t qos) {
  char buf[MQTT_DEFAULT_TOPIC_MAX_LEN];
  return this->subscribe_(topic.build(buf).c_str(), qos);
}
 
void MQTTClientComponent::resubscribe_subscription_(MQTTSubscription *sub) {
  if (sub->subscribed)
    return;
 
  const uint32_t now = millis();
  bool do_resub = sub->resubscribe_timeout == 0 || now - sub->resubscribe_timeout > 1000;
 
  if (do_resub) {
    sub->subscribed = this->subscribe_topic_(sub->topic, sub->qos);
    sub->resubscribe_timeout = now;
  }
}
void MQTTClientComponent::resubscribe_subscriptions_() {
  for (auto &subscription : this->subscriptions_) {
    this->resubscribe_subscription_(&subscription);
  }
}
 
#ifdef USE_MQTT_CRASH_LOG
void MQTTClientComponent::handle_crash_log_request_(const std::string &topic, const std::string &payload) {
  // Only the CLI's empty request counts: a log_topic pointing at this topic would otherwise replay the crash
  // log for every line it logs
  if (!this->crash_log_offered_ || !payload.empty())
    return;
  char logs_topic[ESPHOME_TOPIC_BUFFER_SIZE];
  format_esphome_topic(logs_topic, "logs");
  if (topic == logs_topic)
    logger::log_crash_to_client();
}
#endif
 
static StringRef static_topic_builder(const void *context, std::span<char, MQTT_DEFAULT_TOPIC_MAX_LEN>) {
  return StringRef(static_cast<const char *>(context));
}
// Same as static_topic_builder, but a distinct function so its address marks the topics to free
static StringRef copied_topic_builder(const void *context, std::span<char, MQTT_DEFAULT_TOPIC_MAX_LEN>) {
  return StringRef(static_cast<const char *>(context));
}
 
MQTTTopicSource static_topic(const char *topic) { return {static_topic_builder, topic}; }
 
static_assert(MQTT_DEFAULT_TOPIC_MAX_LEN == 248, "update MQTT_DEFAULT_TOPIC_MAX_LEN in mqtt/__init__.py");
 
#ifdef USE_ESP8266
// mqtt/__init__.py picks this source only for topics that fit
static StringRef progmem_topic_builder(const void *context, std::span<char, MQTT_DEFAULT_TOPIC_MAX_LEN> buf) {
  size_t len = buf_append_str_p(buf.data(), buf.size(), 0, static_cast<const char *>(context));
  return StringRef(buf.data(), len);
}
 
MQTTTopicSource progmem_topic(ProgmemStr topic) {
  return {progmem_topic_builder, reinterpret_cast<const char *>(topic)};
}
#endif
 
MQTTTopicSource MQTTClientComponent::copy_topic(StringRef topic) {
  char *copy = new char[topic.size() + 1];  // NOLINT(cppcoreguidelines-owning-memory)
  memcpy(copy, topic.c_str(), topic.size());
  copy[topic.size()] = '\0';
  return {copied_topic_builder, copy};
}
 
void MQTTClientComponent::add_subscription_(MQTTTopicSource topic, mqtt_subscription_callback_t callback, bool boxed,
                                            uint8_t qos) {
  auto &subscription = this->subscriptions_.emplace_back(MQTTSubscription{
      .topic = topic,
      .callback = callback,
      .resubscribe_timeout = 0,
      .qos = qos,
      .subscribed = false,
      .boxed = boxed,
      .removed = false,
  });
  this->resubscribe_subscription_(&subscription);
}
 
// Out of line so the topic buffer is not part of unsubscribe()'s stack frame
__attribute__((noinline)) static bool topic_equals(const MQTTTopicSource &source, const std::string &topic) {
  char buf[MQTT_DEFAULT_TOPIC_MAX_LEN];
  return source.build(buf) == topic;
}
 
void MQTTClientComponent::unsubscribe(const std::string &topic) {
  bool ret = this->mqtt_backend_.unsubscribe(topic.c_str());
  yield();
  if (ret) {
    ESP_LOGV(TAG, "unsubscribe(topic='%s')", topic.c_str());
  } else {
    delay(5);
    ESP_LOGV(TAG, "Unsubscribe failed for topic='%s'.", topic.c_str());
    this->status_momentary_warning(1000);
  }
 
  for (auto &subscription : this->subscriptions_) {
    if (topic_equals(subscription.topic, topic))
      subscription.removed = true;
  }
  // A callback may be running; the outermost dispatch erases them once it returns.
  if (this->dispatching_) {
    this->removed_while_dispatching_ = true;
  } else {
    this->erase_removed_subscriptions_();
  }
}
 
void MQTTClientComponent::erase_removed_subscriptions_() {
  auto it = this->subscriptions_.begin();
  while (it != this->subscriptions_.end()) {
    if (it->removed) {
      if (it->boxed)
        it->callback.free_boxed();
      if (it->topic.builder == copied_topic_builder)
        delete[] static_cast<const char *>(it->topic.context);  // NOLINT(cppcoreguidelines-owning-memory)
      it = this->subscriptions_.erase(it);
    } else {
      ++it;
    }
  }
}
 
// Publish
bool MQTTClientComponent::publish(const std::string &topic, const std::string &payload, uint8_t qos, bool retain) {
  return this->publish(topic, payload.data(), payload.size(), qos, retain);
}
 
bool MQTTClientComponent::publish(const std::string &topic, const char *payload, size_t payload_length, uint8_t qos,
                                  bool retain) {
  return this->publish(topic.c_str(), payload, payload_length, qos, retain);
}
 
bool MQTTClientComponent::publish(const MQTTMessage &message) {
  return this->publish(message.topic.c_str(), message.payload.c_str(), message.payload.length(), message.qos,
                       message.retain);
}
bool MQTTClientComponent::publish_json(const std::string &topic, const json::json_build_t &f, uint8_t qos,
                                       bool retain) {
  return this->publish_json(topic.c_str(), f, qos, retain);
}
 
bool MQTTClientComponent::publish(const char *topic, const char *payload, size_t payload_length, uint8_t qos,
                                  bool retain) {
  if (!this->is_connected()) {
    return false;
  }
  size_t topic_len = strlen(topic);
  bool logging_topic = (topic_len == this->log_message_.topic.size()) &&
                       (memcmp(this->log_message_.topic.c_str(), topic, topic_len) == 0);
  bool ret = this->mqtt_backend_.publish(topic, payload, payload_length, qos, retain);
  delay(0);
  if (!ret && !logging_topic && this->is_connected()) {
    delay(0);
    ret = this->mqtt_backend_.publish(topic, payload, payload_length, qos, retain);
    delay(0);
  }
 
  if (!logging_topic) {
    if (ret) {
      ESP_LOGV(TAG, "Publish(topic='%s' retain=%d qos=%d)", topic, retain, qos);
      ESP_LOGVV(TAG, "Publish payload (len=%u): '%.*s'", payload_length, static_cast<int>(payload_length), payload);
    } else {
      ESP_LOGV(TAG, "Publish failed for topic='%s' (len=%u). Will retry", topic, payload_length);
      this->status_momentary_warning(1000);
    }
  }
  return ret != 0;
}
 
bool MQTTClientComponent::publish_json(const char *topic, const json::json_build_t &f, uint8_t qos, bool retain) {
  auto message = json::build_json(f);
  return this->publish(topic, message.c_str(), message.size(), qos, retain);
}
 
void MQTTClientComponent::enable() {
  if (this->state_ != MQTT_CLIENT_DISABLED)
    return;
  ESP_LOGD(TAG, "Enabling");
  this->state_ = MQTT_CLIENT_DISCONNECTED;
  this->last_connected_ = millis();
  this->start_dnslookup_();
}
 
void MQTTClientComponent::disable() {
  if (this->state_ == MQTT_CLIENT_DISABLED)
    return;
  ESP_LOGD(TAG, "Disabling");
  this->state_ = MQTT_CLIENT_DISABLED;
  this->on_shutdown();
}
 
// Out of line so the topic buffer is off the stack while the matched callback runs
__attribute__((noinline)) static bool topic_matches(const char *message, const MQTTTopicSource &topic) {
  char buf[MQTT_DEFAULT_TOPIC_MAX_LEN];
  return topic_match(message, topic.build(buf).c_str());
}
 
void MQTTClientComponent::on_message(const std::string &topic, const std::string &payload) {
#ifdef USE_ESP8266
  // IMPORTANT: This defer is REQUIRED to prevent stack overflow crashes on ESP8266.
  //
  // On ESP8266, this callback is invoked directly from the lwIP/AsyncTCP network stack
  // which runs in the "sys" context with a very limited stack (~4KB). By the time we
  // reach this function, the stack is already partially consumed by the network
  // processing chain: tcp_input -> AsyncClient::_recv -> AsyncMqttClient::_onMessage -> here.
  //
  // MQTT subscription callbacks can trigger arbitrary user actions (automations, HTTP
  // requests, sensor updates, etc.) which may have deep call stacks of their own.
  // For example, an HTTP request action requires: DNS lookup -> TCP connect -> TLS
  // handshake (if HTTPS) -> request formatting. This easily overflows the remaining
  // system stack space, causing a LoadStoreAlignmentCause exception or silent corruption.
  //
  // By deferring to the main loop, we ensure callbacks execute with a fresh, full-size
  // stack in the normal application context rather than the constrained network task.
  //
  // DO NOT REMOVE THIS DEFER without understanding the above. It may appear to work
  // in simple tests but will cause crashes with complex automations.
  this->defer([this, topic, payload]() {
#endif
#ifdef USE_MQTT_CRASH_LOG
    this->handle_crash_log_request_(topic, payload);
#endif
    // 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.
    // Restored rather than cleared, so only the outermost dispatch erases what was unsubscribed meanwhile.
    const bool was_dispatching = this->dispatching_;
    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 (!subscription.removed && topic_matches(topic.c_str(), subscription.topic))
        subscription.callback.call(topic, payload);
    }
    this->dispatching_ = was_dispatching;
    if (!was_dispatching && this->removed_while_dispatching_) {
      this->removed_while_dispatching_ = false;
      this->erase_removed_subscriptions_();
    }
#ifdef USE_ESP8266
  });
#endif
}
 
// Setters
void MQTTClientComponent::disable_log_message() { this->log_message_.topic = ""; }
bool MQTTClientComponent::is_log_message_enabled() const { return !this->log_message_.topic.empty(); }
void MQTTClientComponent::register_mqtt_component(MQTTComponent *component) { this->children_.push_back(component); }
void MQTTClientComponent::set_keep_alive(uint16_t keep_alive_s) { this->mqtt_backend_.set_keep_alive(keep_alive_s); }
void MQTTClientComponent::set_log_message_template(MQTTMessage &&message) { this->log_message_ = std::move(message); }
const MQTTDiscoveryInfo &MQTTClientComponent::get_discovery_info() const { return this->discovery_info_; }
void MQTTClientComponent::set_topic_prefix(std::string topic_prefix) { this->topic_prefix_ = std::move(topic_prefix); }
const std::string &MQTTClientComponent::get_topic_prefix() const { return this->topic_prefix_; }
void MQTTClientComponent::disable_birth_message() {
  this->birth_message_.topic = "";
  this->recalculate_availability_();
}
void MQTTClientComponent::disable_shutdown_message() {
  this->shutdown_message_.topic = "";
  this->recalculate_availability_();
}
bool MQTTClientComponent::is_discovery_enabled() const { return !this->discovery_info_.prefix.empty(); }
bool MQTTClientComponent::is_discovery_ip_enabled() const { return this->discovery_info_.discover_ip; }
const Availability &MQTTClientComponent::get_availability() { return this->availability_; }
void MQTTClientComponent::recalculate_availability_() {
  if (this->birth_message_.topic.empty() || this->birth_message_.topic != this->last_will_.topic) {
    this->availability_.topic = "";
    return;
  }
  this->availability_.topic = this->birth_message_.topic;
  this->availability_.payload_available = this->birth_message_.payload;
  this->availability_.payload_not_available = this->last_will_.payload;
}
 
void MQTTClientComponent::set_last_will(MQTTMessage &&message) {
  this->last_will_ = std::move(message);
  this->recalculate_availability_();
}
 
void MQTTClientComponent::set_birth_message(MQTTMessage &&message) {
  this->birth_message_ = std::move(message);
  this->recalculate_availability_();
}
 
void MQTTClientComponent::set_shutdown_message(MQTTMessage &&message) { this->shutdown_message_ = std::move(message); }
 
void MQTTClientComponent::set_discovery_info(std::string &&prefix, MQTTDiscoveryUniqueIdGenerator unique_id_generator,
                                             MQTTDiscoveryObjectIdGenerator object_id_generator, bool retain,
                                             bool discover_ip, bool clean) {
  this->discovery_info_.prefix = std::move(prefix);
  this->discovery_info_.discover_ip = discover_ip;
  this->discovery_info_.unique_id_generator = unique_id_generator;
  this->discovery_info_.object_id_generator = object_id_generator;
  this->discovery_info_.retain = retain;
  this->discovery_info_.clean = clean;
}
 
void MQTTClientComponent::disable_last_will() { this->last_will_.topic = ""; }
 
void MQTTClientComponent::disable_discovery() {
  this->discovery_info_ = MQTTDiscoveryInfo{
      .prefix = "",
      .retain = false,
      .discover_ip = false,
      .clean = false,
      .unique_id_generator = MQTT_LEGACY_UNIQUE_ID_GENERATOR,
      .object_id_generator = MQTT_NONE_OBJECT_ID_GENERATOR,
  };
}
void MQTTClientComponent::on_shutdown() {
  if (!this->shutdown_message_.topic.empty()) {
    yield();
    this->publish(this->shutdown_message_);
    yield();
  }
  this->mqtt_backend_.disconnect();
}
 
void MQTTClientComponent::set_on_connect(mqtt_on_connect_callback_t &&callback) {
  this->mqtt_backend_.set_on_connect(std::forward<mqtt_on_connect_callback_t>(callback));
}
 
void MQTTClientComponent::set_on_disconnect(mqtt_on_disconnect_callback_t &&callback) {
  auto callback_copy = callback;
  this->mqtt_backend_.set_on_disconnect(std::forward<mqtt_on_disconnect_callback_t>(callback));
  this->on_disconnect_.add(std::move(callback_copy));
}
 
MQTTClientComponent *global_mqtt_client = nullptr;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
 
}  // namespace esphome::mqtt
 
#endif  // USE_MQTT