Switch dispatcher to use async_run_hass_job (#74514)

* Switch dispatcher to use async_run_hass_job

- Since we already wrap all the callbacks in catch_log_exception
  we can use async_run_hass_job here

- The overhead of wrapping the call in a call_soon, queuing it
  and running it later usually exceeds the overhead of running
  the job itself

* fix size change during iteration

* fix out of order send

* fix missing mocking in unifi test

* Fix Legrand Home+ Control updating entities before the coordinator update had finished

* stray debug

homeassistant/components/hassio/websocket_api.py

@@ -86,8 +86,8 @@ async def websocket_supervisor_event(
     hass: HomeAssistant, connection: ActiveConnection, msg: dict
 ):
     """Publish events from the Supervisor."""
-    async_dispatcher_send(hass, EVENT_SUPERVISOR_EVENT, msg[ATTR_DATA])
     connection.send_result(msg[WS_ID])
+    async_dispatcher_send(hass, EVENT_SUPERVISOR_EVENT, msg[ATTR_DATA])
 
 
 @websocket_api.websocket_command(

homeassistant/components/home_plus_control/__init__.py

@@ -9,7 +9,7 @@ import voluptuous as vol
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET, Platform
-from homeassistant.core import HomeAssistant
+from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers import (
     config_entry_oauth2_flow,
     config_validation as cv,
@@ -102,12 +102,29 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
             # Note: asyncio.TimeoutError and aiohttp.ClientError are already
             # handled by the data update coordinator.
             async with async_timeout.timeout(10):
-                module_data = await api.async_get_modules()
+                return await api.async_get_modules()
         except HomePlusControlApiError as err:
             raise UpdateFailed(
                 f"Error communicating with API: {err} [{type(err)}]"
             ) from err
 
+    coordinator = DataUpdateCoordinator(
+        hass,
+        _LOGGER,
+        # Name of the data. For logging purposes.
+        name="home_plus_control_module",
+        update_method=async_update_data,
+        # Polling interval. Will only be polled if there are subscribers.
+        update_interval=timedelta(seconds=300),
+    )
+    hass_entry_data[DATA_COORDINATOR] = coordinator
+
+    @callback
+    def _async_update_entities():
+        """Process entities and add or remove them based after an update."""
+        if not (module_data := coordinator.data):
+            return
+
         # Remove obsolete entities from Home Assistant
         entity_uids_to_remove = uids - set(module_data)
         for uid in entity_uids_to_remove:
@@ -126,18 +143,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
                 coordinator,
             )
 
-        return module_data
-
-    coordinator = DataUpdateCoordinator(
-        hass,
-        _LOGGER,
-        # Name of the data. For logging purposes.
-        name="home_plus_control_module",
-        update_method=async_update_data,
-        # Polling interval. Will only be polled if there are subscribers.
-        update_interval=timedelta(seconds=300),
-    )
-    hass_entry_data[DATA_COORDINATOR] = coordinator
+    entry.async_on_unload(coordinator.async_add_listener(_async_update_entities))
 
     async def start_platforms():
         """Continue setting up the platforms."""

homeassistant/helpers/dispatcher.py

@@ -85,8 +85,18 @@ def async_dispatcher_send(hass: HomeAssistant, signal: str, *args: Any) -> None:
     This method must be run in the event loop.
     """
     target_list = hass.data.get(DATA_DISPATCHER, {}).get(signal, {})
+
+    run: list[HassJob] = []
     for target, job in target_list.items():
         if job is None:
             job = _generate_job(signal, target)
             target_list[target] = job
-        hass.async_add_hass_job(job, *args)
+
+        # Run the jobs all at the end
+        # to ensure no jobs add more disptachers
+        # which can result in the target_list
+        # changing size during iteration
+        run.append(job)
+
+    for job in run:
+        hass.async_run_hass_job(job, *args)

tests/components/unifi/test_controller.py

@@ -379,7 +379,21 @@ async def test_wireless_client_event_calls_update_wireless_devices(
     hass, aioclient_mock, mock_unifi_websocket
 ):
     """Call update_wireless_devices method when receiving wireless client event."""
-    await setup_unifi_integration(hass, aioclient_mock)
+    client_1_dict = {
+        "essid": "ssid",
+        "disabled": False,
+        "hostname": "client_1",
+        "ip": "10.0.0.4",
+        "is_wired": False,
+        "last_seen": dt_util.as_timestamp(dt_util.utcnow()),
+        "mac": "00:00:00:00:00:01",
+    }
+    await setup_unifi_integration(
+        hass,
+        aioclient_mock,
+        clients_response=[client_1_dict],
+        known_wireless_clients=(client_1_dict["mac"],),
+    )
 
     with patch(
         "homeassistant.components.unifi.controller.UniFiController.update_wireless_clients",
@@ -391,6 +405,7 @@ async def test_wireless_client_event_calls_update_wireless_devices(
                 "data": [
                     {
                         "datetime": "2020-01-20T19:37:04Z",
+                        "user": "00:00:00:00:00:01",
                         "key": aiounifi.events.WIRELESS_CLIENT_CONNECTED,
                         "msg": "User[11:22:33:44:55:66] has connected to WLAN",
                         "time": 1579549024893,