jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-25 14:17:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Migrate mqtt discovery subscribes to use internal helper (#118279)

Browse Source
This commit is contained in:
J. Nick Koston 2024-05-27 18:14:58 -10:00 committed by GitHub
parent 63227f14ed
commit 69a177e864
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 45 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
homeassistant/components/mqtt/discovery.py
View File

@ -329,54 +329,55 @@ async def async_start( # noqa: C901
mqtt_data.last_discovery = time.monotonic() mqtt_data.last_discovery = time.monotonic()
mqtt_integrations = await async_get_mqtt(hass) mqtt_integrations = await async_get_mqtt(hass)
integration_unsubscribe = mqtt_data.integration_unsubscribe
for integration, topics in mqtt_integrations.items(): async def async_integration_message_received(
integration: str, msg: ReceiveMessage
) -> None:
"""Process the received message."""
if TYPE_CHECKING:
assert mqtt_data.data_config_flow_lock
key = f"{integration}_{msg.subscribed_topic}"
async def async_integration_message_received( # Lock to prevent initiating many parallel config flows.
integration: str, msg: ReceiveMessage # Note: The lock is not intended to prevent a race, only for performance
) -> None: async with mqtt_data.data_config_flow_lock:
"""Process the received message.""" # Already unsubscribed
if TYPE_CHECKING: if key not in integration_unsubscribe:
assert mqtt_data.data_config_flow_lock return
key = f"{integration}_{msg.subscribed_topic}"
# Lock to prevent initiating many parallel config flows. data = MqttServiceInfo(
# Note: The lock is not intended to prevent a race, only for performance topic=msg.topic,
async with mqtt_data.data_config_flow_lock: payload=msg.payload,
# Already unsubscribed qos=msg.qos,
if key not in mqtt_data.integration_unsubscribe: retain=msg.retain,
return subscribed_topic=msg.subscribed_topic,
timestamp=msg.timestamp,
)
result = await hass.config_entries.flow.async_init(
integration, context={"source": DOMAIN}, data=data
)
if (
result
and result["type"] == FlowResultType.ABORT
and result["reason"]
in ("already_configured", "single_instance_allowed")
):
integration_unsubscribe.pop(key)()
data = MqttServiceInfo( integration_unsubscribe.update(
topic=msg.topic, {
payload=msg.payload, f"{integration}_{topic}": mqtt.async_subscribe_internal(
qos=msg.qos, hass,
retain=msg.retain, topic,
subscribed_topic=msg.subscribed_topic, functools.partial(async_integration_message_received, integration),
timestamp=msg.timestamp, 0,
) job_type=HassJobType.Coroutinefunction,
result = await hass.config_entries.flow.async_init( )
integration, context={"source": DOMAIN}, data=data for integration, topics in mqtt_integrations.items()
) for topic in topics
if ( }
result )
and result["type"] == FlowResultType.ABORT
and result["reason"]
in ("already_configured", "single_instance_allowed")
):
mqtt_data.integration_unsubscribe.pop(key)()
mqtt_data.integration_unsubscribe.update(
{
f"{integration}_{topic}": await mqtt.async_subscribe(
hass,
topic,
functools.partial(async_integration_message_received, integration),
0,
)
for topic in topics
}
)
async def async_stop(hass: HomeAssistant) -> None: async def async_stop(hass: HomeAssistant) -> None: