[mqtt] Add wake_loop_threadsafe() for low-latency event processing on ESP32 (#11695)

esphome/components/mqtt/__init__.py

@@ -3,7 +3,7 @@ import re
 from esphome import automation
 from esphome.automation import Condition
 import esphome.codegen as cg
-from esphome.components import logger
+from esphome.components import logger, socket
 from esphome.components.esp32 import add_idf_sdkconfig_option
 from esphome.config_helpers import filter_source_files_from_platform
 import esphome.config_validation as cv
@@ -66,6 +66,9 @@ DEPENDENCIES = ["network"]
 def AUTO_LOAD():
     if CORE.is_esp8266 or CORE.is_libretiny:
         return ["async_tcp", "json"]
+    # ESP32 needs socket for wake_loop_threadsafe()
+    if CORE.is_esp32:
+        return ["json", "socket"]
     return ["json"]
 
 
@@ -213,8 +216,6 @@ def validate_fingerprint(value):
 
 def _consume_mqtt_sockets(config: ConfigType) -> ConfigType:
     """Register socket needs for MQTT component."""
-    from esphome.components import socket
-
     # MQTT needs 1 socket for the broker connection
     socket.consume_sockets(1, "mqtt")(config)
     return config
@@ -341,6 +342,11 @@ async def to_code(config):
         # https://github.com/heman/async-mqtt-client/blob/master/library.json
         cg.add_library("heman/AsyncMqttClient-esphome", "2.0.0")
 
+    # MQTT on ESP32 uses wake_loop_threadsafe() to wake the main loop from the MQTT event handler
+    # This enables low-latency MQTT event processing instead of waiting for select() timeout
+    if CORE.is_esp32:
+        socket.require_wake_loop_threadsafe()
+
     cg.add_define("USE_MQTT")
     cg.add_global(mqtt_ns.using)
 

esphome/components/mqtt/mqtt_backend_esp32.cpp

@@ -190,6 +190,11 @@ void MQTTBackendESP32::mqtt_event_handler(void *handler_args, esp_event_base_t b
   if (instance) {
     auto event = *static_cast<esp_mqtt_event_t *>(event_data);
     instance->mqtt_events_.emplace(event);
+
+    // Wake main loop immediately to process MQTT event instead of waiting for select() timeout
+#if defined(USE_SOCKET_SELECT_SUPPORT) && defined(USE_WAKE_LOOP_THREADSAFE)
+    App.wake_loop_threadsafe();
+#endif
   }
 }