Avoid expensive inspection of callbacks to setup mqtt subscriptions (#118161)

homeassistant/components/mqtt/alarm_control_panel.py

@@ -25,7 +25,7 @@ from homeassistant.const import (
     STATE_ALARM_PENDING,
     STATE_ALARM_TRIGGERED,
 )
-from homeassistant.core import HomeAssistant
+from homeassistant.core import HassJobType, HomeAssistant
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.typing import ConfigType
@@ -220,6 +220,7 @@ class MqttAlarm(MqttEntity, alarm.AlarmControlPanelEntity):
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
                     "encoding": self._config[CONF_ENCODING] or None,
+                    "job_type": HassJobType.Callback,
                 }
             },
         )

homeassistant/components/mqtt/binary_sensor.py

@@ -26,7 +26,7 @@ from homeassistant.const import (
     STATE_UNAVAILABLE,
     STATE_UNKNOWN,
 )
-from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
+from homeassistant.core import CALLBACK_TYPE, HassJobType, HomeAssistant, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 import homeassistant.helpers.event as evt
@@ -248,6 +248,7 @@ class MqttBinarySensor(MqttEntity, BinarySensorEntity, RestoreEntity):
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
                     "encoding": self._config[CONF_ENCODING] or None,
+                    "job_type": HassJobType.Callback,
                 }
             },
         )

homeassistant/components/mqtt/camera.py

@@ -13,7 +13,7 @@ from homeassistant.components import camera
 from homeassistant.components.camera import Camera
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_NAME
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
@@ -124,6 +124,7 @@ class MqttCamera(MqttEntity, Camera):
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
                     "encoding": None,
+                    "job_type": HassJobType.Callback,
                 }
             },
         )

homeassistant/components/mqtt/client.py

@@ -201,6 +201,7 @@ def async_subscribe_internal(
     msg_callback: Callable[[ReceiveMessage], Coroutine[Any, Any, None] | None],
     qos: int = DEFAULT_QOS,
     encoding: str | None = DEFAULT_ENCODING,
+    job_type: HassJobType | None = None,
 ) -> CALLBACK_TYPE:
     """Subscribe to an MQTT topic.
 
@@ -228,7 +229,7 @@ def async_subscribe_internal(
             translation_domain=DOMAIN,
             translation_placeholders={"topic": topic},
         )
-    return client.async_subscribe(topic, msg_callback, qos, encoding)
+    return client.async_subscribe(topic, msg_callback, qos, encoding, job_type)
 
 
 @bind_hass
@@ -867,12 +868,14 @@ class MQTT:
         msg_callback: Callable[[ReceiveMessage], Coroutine[Any, Any, None] | None],
         qos: int,
         encoding: str | None = None,
+        job_type: HassJobType | None = None,
     ) -> Callable[[], None]:
         """Set up a subscription to a topic with the provided qos."""
         if not isinstance(topic, str):
             raise HomeAssistantError("Topic needs to be a string!")
 
-        job_type = get_hassjob_callable_job_type(msg_callback)
+        if job_type is None:
+            job_type = get_hassjob_callable_job_type(msg_callback)
         if job_type is not HassJobType.Callback:
             # Only wrap the callback with catch_log_exception
             # if it is not a simple callback since we catch

homeassistant/components/mqtt/climate.py

@@ -43,7 +43,7 @@ from homeassistant.const import (
     PRECISION_WHOLE,
     UnitOfTemperature,
 )
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.template import Template
@@ -429,6 +429,7 @@ class MqttTemperatureControlEntity(MqttEntity, ABC):
                 "entity_id": self.entity_id,
                 "qos": qos,
                 "encoding": self._config[CONF_ENCODING] or None,
+                "job_type": HassJobType.Callback,
             }
 
     def render_template(

homeassistant/components/mqtt/cover.py

@@ -28,7 +28,7 @@ from homeassistant.const import (
     STATE_OPEN,
     STATE_OPENING,
 )
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.service_info.mqtt import ReceivePayloadType
@@ -478,6 +478,7 @@ class MqttCover(MqttEntity, CoverEntity):
                 "entity_id": self.entity_id,
                 "qos": self._config[CONF_QOS],
                 "encoding": self._config[CONF_ENCODING] or None,
+                "job_type": HassJobType.Callback,
             }
 
         if self._config.get(CONF_STATE_TOPIC):
@@ -491,6 +492,7 @@ class MqttCover(MqttEntity, CoverEntity):
                 "entity_id": self.entity_id,
                 "qos": self._config[CONF_QOS],
                 "encoding": self._config[CONF_ENCODING] or None,
+                "job_type": HassJobType.Callback,
             }
 
         if self._config.get(CONF_TILT_STATUS_TOPIC) is not None:
@@ -504,6 +506,7 @@ class MqttCover(MqttEntity, CoverEntity):
                 "entity_id": self.entity_id,
                 "qos": self._config[CONF_QOS],
                 "encoding": self._config[CONF_ENCODING] or None,
+                "job_type": HassJobType.Callback,
             }
 
         self._sub_state = subscription.async_prepare_subscribe_topics(

homeassistant/components/mqtt/device_tracker.py

@@ -25,7 +25,7 @@ from homeassistant.const import (
     STATE_HOME,
     STATE_NOT_HOME,
 )
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.typing import ConfigType
@@ -155,6 +155,7 @@ class MqttDeviceTracker(MqttEntity, TrackerEntity):
                     ),
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
+                    "job_type": HassJobType.Callback,
                 }
             },
         )

homeassistant/components/mqtt/discovery.py

@@ -12,7 +12,7 @@ from typing import TYPE_CHECKING, Any
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_DEVICE, CONF_PLATFORM
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 from homeassistant.data_entry_flow import FlowResultType
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.dispatcher import (
@@ -319,10 +319,14 @@ async def async_start(  # noqa: C901
                 hass, MQTT_DISCOVERY_DONE.format(*discovery_hash), None
             )
 
-    # async_subscribe will never suspend so there is no need to create a task
-    # here and its faster to await them in sequence
     mqtt_data.discovery_unsubscribe = [
-        await mqtt.async_subscribe(hass, topic, async_discovery_message_received, 0)
+        mqtt.async_subscribe_internal(
+            hass,
+            topic,
+            async_discovery_message_received,
+            0,
+            job_type=HassJobType.Callback,
+        )
         for topic in (
             f"{discovery_topic}/+/+/config",
             f"{discovery_topic}/+/+/+/config",

homeassistant/components/mqtt/event.py

@@ -17,7 +17,7 @@ from homeassistant.components.event import (
 )
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_DEVICE_CLASS, CONF_NAME, CONF_VALUE_TEMPLATE
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.typing import ConfigType
@@ -200,6 +200,7 @@ class MqttEvent(MqttEntity, EventEntity):
             "entity_id": self.entity_id,
             "qos": self._config[CONF_QOS],
             "encoding": self._config[CONF_ENCODING] or None,
+            "job_type": HassJobType.Callback,
         }
 
         self._sub_state = subscription.async_prepare_subscribe_topics(

homeassistant/components/mqtt/fan.py

@@ -27,7 +27,7 @@ from homeassistant.const import (
     CONF_PAYLOAD_ON,
     CONF_STATE,
 )
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.template import Template
@@ -447,6 +447,7 @@ class MqttFan(MqttEntity, FanEntity):
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
                     "encoding": self._config[CONF_ENCODING] or None,
+                    "job_type": HassJobType.Callback,
                 }
             return has_topic
 

homeassistant/components/mqtt/humidifier.py

@@ -30,7 +30,7 @@ from homeassistant.const import (
     CONF_PAYLOAD_ON,
     CONF_STATE,
 )
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.template import Template
@@ -293,6 +293,7 @@ class MqttHumidifier(MqttEntity, HumidifierEntity):
                 "entity_id": self.entity_id,
                 "qos": qos,
                 "encoding": self._config[CONF_ENCODING] or None,
+                "job_type": HassJobType.Callback,
             }
 
     @callback

homeassistant/components/mqtt/image.py

@@ -16,7 +16,7 @@ from homeassistant.components import image
 from homeassistant.components.image import DEFAULT_CONTENT_TYPE, ImageEntity
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_NAME
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.httpx_client import get_async_client
@@ -202,6 +202,7 @@ class MqttImage(MqttEntity, ImageEntity):
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
                     "encoding": encoding,
+                    "job_type": HassJobType.Callback,
                 }
             return has_topic
 

homeassistant/components/mqtt/lawn_mower.py

@@ -17,7 +17,7 @@ from homeassistant.components.lawn_mower import (
 )
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_NAME, CONF_OPTIMISTIC
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.restore_state import RestoreEntity
@@ -192,6 +192,7 @@ class MqttLawnMower(MqttEntity, LawnMowerEntity, RestoreEntity):
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
                     "encoding": self._config[CONF_ENCODING] or None,
+                    "job_type": HassJobType.Callback,
                 }
             },
         )

homeassistant/components/mqtt/light/schema_basic.py

@@ -37,7 +37,7 @@ from homeassistant.const import (
     CONF_PAYLOAD_ON,
     STATE_ON,
 )
-from homeassistant.core import callback
+from homeassistant.core import HassJobType, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.restore_state import RestoreEntity
 from homeassistant.helpers.typing import ConfigType
@@ -580,6 +580,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
                     "encoding": self._config[CONF_ENCODING] or None,
+                    "job_type": HassJobType.Callback,
                 }
 
         add_topic(CONF_STATE_TOPIC, self._state_received, {"_attr_is_on"})

homeassistant/components/mqtt/light/schema_json.py

@@ -47,7 +47,7 @@ from homeassistant.const import (
     CONF_XY,
     STATE_ON,
 )
-from homeassistant.core import async_get_hass, callback
+from homeassistant.core import HassJobType, async_get_hass, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
 from homeassistant.helpers.json import json_dumps
@@ -522,6 +522,7 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
                     "encoding": self._config[CONF_ENCODING] or None,
+                    "job_type": HassJobType.Callback,
                 }
             },
         )

homeassistant/components/mqtt/light/schema_template.py

@@ -29,7 +29,7 @@ from homeassistant.const import (
     STATE_OFF,
     STATE_ON,
 )
-from homeassistant.core import callback
+from homeassistant.core import HassJobType, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.restore_state import RestoreEntity
 from homeassistant.helpers.typing import ConfigType, TemplateVarsType
@@ -282,6 +282,7 @@ class MqttLightTemplate(MqttEntity, LightEntity, RestoreEntity):
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
                     "encoding": self._config[CONF_ENCODING] or None,
+                    "job_type": HassJobType.Callback,
                 }
             },
         )

homeassistant/components/mqtt/lock.py

@@ -19,7 +19,7 @@ from homeassistant.const import (
     CONF_OPTIMISTIC,
     CONF_VALUE_TEMPLATE,
 )
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.typing import ConfigType, TemplateVarsType
@@ -232,6 +232,7 @@ class MqttLock(MqttEntity, LockEntity):
                 "entity_id": self.entity_id,
                 CONF_QOS: qos,
                 CONF_ENCODING: encoding,
+                "job_type": HassJobType.Callback,
             }
         }
 

homeassistant/components/mqtt/mixins.py

@@ -30,7 +30,7 @@ from homeassistant.const import (
     CONF_UNIQUE_ID,
     CONF_VALUE_TEMPLATE,
 )
-from homeassistant.core import Event, HomeAssistant, callback
+from homeassistant.core import Event, HassJobType, HomeAssistant, callback
 from homeassistant.helpers import device_registry as dr, entity_registry as er
 from homeassistant.helpers.device_registry import (
     DeviceEntry,
@@ -405,6 +405,7 @@ class MqttAttributesMixin(Entity):
                     "entity_id": self.entity_id,
                     "qos": self._attributes_config.get(CONF_QOS),
                     "encoding": self._attributes_config[CONF_ENCODING] or None,
+                    "job_type": HassJobType.Callback,
                 }
             },
         )
@@ -519,6 +520,7 @@ class MqttAvailabilityMixin(Entity):
                 "entity_id": self.entity_id,
                 "qos": self._avail_config[CONF_QOS],
                 "encoding": self._avail_config[CONF_ENCODING] or None,
+                "job_type": HassJobType.Callback,
             }
             for topic in self._avail_topics
         }

homeassistant/components/mqtt/number.py

@@ -26,7 +26,7 @@ from homeassistant.const import (
     CONF_UNIT_OF_MEASUREMENT,
     CONF_VALUE_TEMPLATE,
 )
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.typing import ConfigType
@@ -214,6 +214,7 @@ class MqttNumber(MqttEntity, RestoreNumber):
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
                     "encoding": self._config[CONF_ENCODING] or None,
+                    "job_type": HassJobType.Callback,
                 }
             },
         )

homeassistant/components/mqtt/select.py

@@ -12,7 +12,7 @@ from homeassistant.components import select
 from homeassistant.components.select import SelectEntity
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_NAME, CONF_OPTIMISTIC, CONF_VALUE_TEMPLATE
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.restore_state import RestoreEntity
@@ -154,6 +154,7 @@ class MqttSelect(MqttEntity, SelectEntity, RestoreEntity):
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
                     "encoding": self._config[CONF_ENCODING] or None,
+                    "job_type": HassJobType.Callback,
                 }
             },
         )

homeassistant/components/mqtt/sensor.py

@@ -31,7 +31,13 @@ from homeassistant.const import (
     STATE_UNAVAILABLE,
     STATE_UNKNOWN,
 )
-from homeassistant.core import CALLBACK_TYPE, HomeAssistant, State, callback
+from homeassistant.core import (
+    CALLBACK_TYPE,
+    HassJobType,
+    HomeAssistant,
+    State,
+    callback,
+)
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.event import async_call_later
@@ -297,6 +303,7 @@ class MqttSensor(MqttEntity, RestoreSensor):
             "entity_id": self.entity_id,
             "qos": self._config[CONF_QOS],
             "encoding": self._config[CONF_ENCODING] or None,
+            "job_type": HassJobType.Callback,
         }
 
         self._sub_state = subscription.async_prepare_subscribe_topics(

homeassistant/components/mqtt/siren.py

@@ -28,7 +28,7 @@ from homeassistant.const import (
     CONF_PAYLOAD_OFF,
     CONF_PAYLOAD_ON,
 )
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.json import json_dumps
@@ -282,6 +282,7 @@ class MqttSiren(MqttEntity, SirenEntity):
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
                     "encoding": self._config[CONF_ENCODING] or None,
+                    "job_type": HassJobType.Callback,
                 }
             },
         )

homeassistant/components/mqtt/subscription.py

@@ -7,7 +7,7 @@ from dataclasses import dataclass
 from functools import partial
 from typing import TYPE_CHECKING, Any
 
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 
 from . import debug_info
 from .client import async_subscribe_internal
@@ -27,6 +27,7 @@ class EntitySubscription:
     qos: int = 0
     encoding: str = "utf-8"
     entity_id: str | None = None
+    job_type: HassJobType | None = None
 
     def resubscribe_if_necessary(
         self, hass: HomeAssistant, other: EntitySubscription | None
@@ -62,7 +63,12 @@ class EntitySubscription:
         if not self.should_subscribe or not self.topic:
             return
         self.unsubscribe_callback = async_subscribe_internal(
-            self.hass, self.topic, self.message_callback, self.qos, self.encoding
+            self.hass,
+            self.topic,
+            self.message_callback,
+            self.qos,
+            self.encoding,
+            self.job_type,
         )
 
     def _should_resubscribe(self, other: EntitySubscription | None) -> bool:
@@ -112,6 +118,7 @@ def async_prepare_subscribe_topics(
             hass=hass,
             should_subscribe=None,
             entity_id=value.get("entity_id", None),
+            job_type=value.get("job_type", None),
         )
         # Get the current subscription state
         current = current_subscriptions.pop(key, None)

homeassistant/components/mqtt/switch.py

@@ -20,7 +20,7 @@ from homeassistant.const import (
     CONF_VALUE_TEMPLATE,
     STATE_ON,
 )
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.restore_state import RestoreEntity
@@ -145,6 +145,7 @@ class MqttSwitch(MqttEntity, SwitchEntity, RestoreEntity):
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
                     "encoding": self._config[CONF_ENCODING] or None,
+                    "job_type": HassJobType.Callback,
                 }
             },
         )

homeassistant/components/mqtt/tag.py

@@ -11,7 +11,7 @@ import voluptuous as vol
 from homeassistant.components import tag
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_DEVICE, CONF_VALUE_TEMPLATE
-from homeassistant.core import HomeAssistant
+from homeassistant.core import HassJobType, HomeAssistant, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
@@ -142,28 +142,32 @@ class MQTTTagScanner(MqttDiscoveryDeviceUpdateMixin):
         update_device(self.hass, self._config_entry, config)
         await self.subscribe_topics()
 
+    @callback
+    def _async_tag_scanned(self, msg: ReceiveMessage) -> None:
+        """Handle new tag scanned."""
+        try:
+            tag_id = str(self._value_template(msg.payload, "")).strip()
+        except MqttValueTemplateException as exc:
+            _LOGGER.warning(exc)
+            return
+        if not tag_id:  # No output from template, ignore
+            return
+
+        self.hass.async_create_task(
+            tag.async_scan_tag(self.hass, tag_id, self.device_id)
+        )
+
     async def subscribe_topics(self) -> None:
         """Subscribe to MQTT topics."""
-
-        async def tag_scanned(msg: ReceiveMessage) -> None:
-            try:
-                tag_id = str(self._value_template(msg.payload, "")).strip()
-            except MqttValueTemplateException as exc:
-                _LOGGER.warning(exc)
-                return
-            if not tag_id:  # No output from template, ignore
-                return
-
-            await tag.async_scan_tag(self.hass, tag_id, self.device_id)
-
         self._sub_state = subscription.async_prepare_subscribe_topics(
             self.hass,
             self._sub_state,
             {
                 "state_topic": {
                     "topic": self._config[CONF_TOPIC],
-                    "msg_callback": tag_scanned,
+                    "msg_callback": self._async_tag_scanned,
                     "qos": self._config[CONF_QOS],
+                    "job_type": HassJobType.Callback,
                 }
             },
         )

homeassistant/components/mqtt/text.py

@@ -20,7 +20,7 @@ from homeassistant.const import (
     CONF_VALUE_TEMPLATE,
     MAX_LENGTH_STATE_STATE,
 )
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.typing import ConfigType
@@ -183,6 +183,7 @@ class MqttTextEntity(MqttEntity, TextEntity):
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
                     "encoding": self._config[CONF_ENCODING] or None,
+                    "job_type": HassJobType.Callback,
                 }
 
         add_subscription(

homeassistant/components/mqtt/trigger.py

@@ -10,7 +10,13 @@ from typing import Any
 import voluptuous as vol
 
 from homeassistant.const import CONF_PAYLOAD, CONF_PLATFORM, CONF_VALUE_TEMPLATE
-from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
+from homeassistant.core import (
+    CALLBACK_TYPE,
+    HassJob,
+    HassJobType,
+    HomeAssistant,
+    callback,
+)
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.template import Template
 from homeassistant.helpers.trigger import TriggerActionType, TriggerData, TriggerInfo
@@ -99,6 +105,11 @@ async def async_attach_trigger(
         "Attaching MQTT trigger for topic: '%s', payload: '%s'", topic, wanted_payload
     )
 
-    return await mqtt.async_subscribe(
-        hass, topic, mqtt_automation_listener, encoding=encoding, qos=qos
+    return mqtt.async_subscribe_internal(
+        hass,
+        topic,
+        mqtt_automation_listener,
+        encoding=encoding,
+        qos=qos,
+        job_type=HassJobType.Callback,
     )

homeassistant/components/mqtt/update.py

@@ -16,7 +16,7 @@ from homeassistant.components.update import (
 )
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_DEVICE_CLASS, CONF_NAME, CONF_VALUE_TEMPLATE
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.restore_state import RestoreEntity
@@ -229,6 +229,7 @@ class MqttUpdate(MqttEntity, UpdateEntity, RestoreEntity):
                     "entity_id": self.entity_id,
                     "qos": self._config[CONF_QOS],
                     "encoding": self._config[CONF_ENCODING] or None,
+                    "job_type": HassJobType.Callback,
                 }
 
         add_subscription(

homeassistant/components/mqtt/vacuum.py

@@ -31,7 +31,7 @@ from homeassistant.const import (
     STATE_IDLE,
     STATE_PAUSED,
 )
-from homeassistant.core import HomeAssistant, async_get_hass, callback
+from homeassistant.core import HassJobType, HomeAssistant, async_get_hass, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
@@ -346,6 +346,7 @@ class MqttStateVacuum(MqttEntity, StateVacuumEntity):
                 "entity_id": self.entity_id,
                 "qos": self._config[CONF_QOS],
                 "encoding": self._config[CONF_ENCODING] or None,
+                "job_type": HassJobType.Callback,
             }
         self._sub_state = subscription.async_prepare_subscribe_topics(
             self.hass, self._sub_state, topics

homeassistant/components/mqtt/valve.py

@@ -26,7 +26,7 @@ from homeassistant.const import (
     STATE_OPEN,
     STATE_OPENING,
 )
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HassJobType, HomeAssistant, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.typing import ConfigType
@@ -357,6 +357,7 @@ class MqttValve(MqttEntity, ValveEntity):
                 "entity_id": self.entity_id,
                 "qos": self._config[CONF_QOS],
                 "encoding": self._config[CONF_ENCODING] or None,
+                "job_type": HassJobType.Callback,
             }
 
         self._sub_state = subscription.async_prepare_subscribe_topics(

tests/components/axis/test_hub.py

@@ -2,7 +2,7 @@
 
 from ipaddress import ip_address
 from unittest import mock
-from unittest.mock import Mock, call, patch
+from unittest.mock import ANY, Mock, call, patch
 
 import axis as axislib
 import pytest
@@ -90,7 +90,7 @@ async def test_device_support_mqtt(
     hass: HomeAssistant, mqtt_mock: MqttMockHAClient, setup_config_entry
 ) -> None:
     """Successful setup."""
-    mqtt_call = call(f"axis/{MAC}/#", mock.ANY, 0, "utf-8")
+    mqtt_call = call(f"axis/{MAC}/#", mock.ANY, 0, "utf-8", ANY)
     assert mqtt_call in mqtt_mock.async_subscribe.call_args_list
 
     topic = f"axis/{MAC}/event/tns:onvif/Device/tns:axis/Sensor/PIR/$source/sensor/0"

tests/components/mqtt/test_common.py

@@ -27,7 +27,7 @@ from homeassistant.const import (
     STATE_UNAVAILABLE,
     EntityCategory,
 )
-from homeassistant.core import HomeAssistant
+from homeassistant.core import HassJobType, HomeAssistant
 from homeassistant.generated.mqtt import MQTT
 from homeassistant.helpers import device_registry as dr, entity_registry as er
 from homeassistant.helpers.dispatcher import async_dispatcher_send
@@ -1189,7 +1189,9 @@ async def help_test_entity_id_update_subscriptions(
     assert state is not None
     assert mqtt_mock.async_subscribe.call_count == len(topics) + 2 + DISCOVERY_COUNT
     for topic in topics:
-        mqtt_mock.async_subscribe.assert_any_call(topic, ANY, ANY, ANY)
+        mqtt_mock.async_subscribe.assert_any_call(
+            topic, ANY, ANY, ANY, HassJobType.Callback
+        )
     mqtt_mock.async_subscribe.reset_mock()
 
     entity_registry.async_update_entity(
@@ -1203,7 +1205,9 @@ async def help_test_entity_id_update_subscriptions(
     state = hass.states.get(f"{domain}.milk")
     assert state is not None
     for topic in topics:
-        mqtt_mock.async_subscribe.assert_any_call(topic, ANY, ANY, ANY)
+        mqtt_mock.async_subscribe.assert_any_call(
+            topic, ANY, ANY, ANY, HassJobType.Callback
+        )
 
 
 async def help_test_entity_id_update_discovery_update(

tests/components/mqtt/test_subscription.py

@@ -154,7 +154,7 @@ async def test_qos_encoding_default(
         {"test_topic1": {"topic": "test-topic1", "msg_callback": msg_callback}},
     )
     await async_subscribe_topics(hass, sub_state)
-    mqtt_mock.async_subscribe.assert_called_with("test-topic1", ANY, 0, "utf-8")
+    mqtt_mock.async_subscribe.assert_called_with("test-topic1", ANY, 0, "utf-8", None)
 
 
 async def test_qos_encoding_custom(
@@ -183,7 +183,7 @@ async def test_qos_encoding_custom(
         },
     )
     await async_subscribe_topics(hass, sub_state)
-    mqtt_mock.async_subscribe.assert_called_with("test-topic1", ANY, 1, "utf-16")
+    mqtt_mock.async_subscribe.assert_called_with("test-topic1", ANY, 1, "utf-16", None)
 
 
 async def test_no_change(

tests/components/mqtt/test_trigger.py

@@ -6,7 +6,7 @@ import pytest
 
 from homeassistant.components import automation
 from homeassistant.const import ATTR_ENTITY_ID, ENTITY_MATCH_ALL, SERVICE_TURN_OFF
-from homeassistant.core import HomeAssistant
+from homeassistant.core import HassJobType, HomeAssistant
 from homeassistant.setup import async_setup_component
 
 from tests.common import async_fire_mqtt_message, async_mock_service, mock_component
@@ -239,7 +239,9 @@ async def test_encoding_default(hass: HomeAssistant, calls, setup_comp) -> None:
         },
     )
 
-    setup_comp.async_subscribe.assert_called_with("test-topic", ANY, 0, "utf-8")
+    setup_comp.async_subscribe.assert_called_with(
+        "test-topic", ANY, 0, "utf-8", HassJobType.Callback
+    )
 
 
 async def test_encoding_custom(hass: HomeAssistant, calls, setup_comp) -> None:
@@ -255,4 +257,6 @@ async def test_encoding_custom(hass: HomeAssistant, calls, setup_comp) -> None:
         },
     )
 
-    setup_comp.async_subscribe.assert_called_with("test-topic", ANY, 0, None)
+    setup_comp.async_subscribe.assert_called_with(
+        "test-topic", ANY, 0, None, HassJobType.Callback
+    )

tests/components/mqtt_eventstream/test_init.py

@@ -66,7 +66,7 @@ async def test_subscribe(hass: HomeAssistant, mqtt_mock: MqttMockHAClient) -> No
     await hass.async_block_till_done()
 
     # Verify that the this entity was subscribed to the topic
-    mqtt_mock.async_subscribe.assert_called_with(sub_topic, ANY, 0, ANY)
+    mqtt_mock.async_subscribe.assert_called_with(sub_topic, ANY, 0, ANY, ANY)
 
 
 async def test_state_changed_event_sends_message(

tests/components/tasmota/test_common.py

@@ -693,7 +693,7 @@ async def help_test_entity_id_update_subscriptions(
     assert state is not None
     assert mqtt_mock.async_subscribe.call_count == len(topics)
     for topic in topics:
-        mqtt_mock.async_subscribe.assert_any_call(topic, ANY, ANY, ANY)
+        mqtt_mock.async_subscribe.assert_any_call(topic, ANY, ANY, ANY, ANY)
     mqtt_mock.async_subscribe.reset_mock()
 
     entity_reg.async_update_entity(
@@ -707,7 +707,7 @@ async def help_test_entity_id_update_subscriptions(
     state = hass.states.get(f"{domain}.milk")
     assert state is not None
     for topic in topics:
-        mqtt_mock.async_subscribe.assert_any_call(topic, ANY, ANY, ANY)
+        mqtt_mock.async_subscribe.assert_any_call(topic, ANY, ANY, ANY, ANY)
 
 
 async def help_test_entity_id_update_discovery_update(

tests/components/tasmota/test_discovery.py

@@ -30,7 +30,9 @@ async def test_subscribing_config_topic(
     discovery_topic = DEFAULT_PREFIX
 
     assert mqtt_mock.async_subscribe.called
-    mqtt_mock.async_subscribe.assert_any_call(discovery_topic + "/#", ANY, 0, "utf-8")
+    mqtt_mock.async_subscribe.assert_any_call(
+        discovery_topic + "/#", ANY, 0, "utf-8", ANY
+    )
 
 
 async def test_future_discovery_message(