Data Coordinator to return unsub func
diff --git a/homeassistant/components/hue/hue_event.py b/homeassistant/components/hue/hue_event.py
index ed1bc1c8f7d9be..a588f68ea4e672 100644
--- a/homeassistant/components/hue/hue_event.py
+++ b/homeassistant/components/hue/hue_event.py
@@ -35,19 +35,13 @@ def __init__(self, sensor, name, bridge, primary_sensor=None):
         self._last_updated = self.sensor.lastupdated
 
         # Register callback in coordinator and add job to remove it on bridge reset.
-        self.bridge.sensor_manager.coordinator.async_add_listener(
-            self.async_update_callback
+        self.bridge.reset_jobs.append(
+            self.bridge.sensor_manager.coordinator.async_add_listener(
+                self.async_update_callback
+            )
         )
-        self.bridge.reset_jobs.append(self.async_will_remove_from_hass)
         _LOGGER.debug("Hue event created: %s", self.event_id)
 
-    @callback
-    def async_will_remove_from_hass(self):
-        """Remove listener on bridge reset."""
-        self.bridge.sensor_manager.coordinator.async_remove_listener(
-            self.async_update_callback
-        )
-
     @callback
     def async_update_callback(self):
         """Fire the event if reason is that state is updated."""
diff --git a/homeassistant/components/hue/light.py b/homeassistant/components/hue/light.py
index e468c516676588..649be55e94e980 100644
--- a/homeassistant/components/hue/light.py
+++ b/homeassistant/components/hue/light.py
@@ -118,13 +118,9 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
     )
 
     # We add a listener after fetching the data, so manually trigger listener
-    light_coordinator.async_add_listener(update_lights)
+    bridge.reset_jobs.append(light_coordinator.async_add_listener(update_lights))
     update_lights()
 
-    bridge.reset_jobs.append(
-        lambda: light_coordinator.async_remove_listener(update_lights)
-    )
-
     api_version = tuple(int(v) for v in bridge.api.config.apiversion.split("."))
 
     allow_groups = bridge.allow_groups
@@ -155,13 +151,9 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
         partial(create_light, HueLight, group_coordinator, bridge, True),
     )
 
-    group_coordinator.async_add_listener(update_groups)
+    bridge.reset_jobs.append(group_coordinator.async_add_listener(update_groups))
     await group_coordinator.async_refresh()
 
-    bridge.reset_jobs.append(
-        lambda: group_coordinator.async_remove_listener(update_groups)
-    )
-
 
 async def async_safe_fetch(bridge, fetch_method):
     """Safely fetch data."""
@@ -339,11 +331,9 @@ def device_info(self):
 
     async def async_added_to_hass(self):
         """When entity is added to hass."""
-        self.coordinator.async_add_listener(self.async_write_ha_state)
-
-    async def async_will_remove_from_hass(self):
-        """When entity will be removed from hass."""
-        self.coordinator.async_remove_listener(self.async_write_ha_state)
+        self.async_on_remove(
+            self.coordinator.async_add_listener(self.async_write_ha_state)
+        )
 
     async def async_turn_on(self, **kwargs):
         """Turn the specified or all lights on."""
diff --git a/homeassistant/components/hue/sensor_base.py b/homeassistant/components/hue/sensor_base.py
index 113957d140ef86..93b98a7c9ce8d3 100644
--- a/homeassistant/components/hue/sensor_base.py
+++ b/homeassistant/components/hue/sensor_base.py
@@ -76,9 +76,8 @@ async def async_register_component(self, platform, async_add_entities):
             return
 
         # We have all components available, start the updating.
-        self.coordinator.async_add_listener(self.async_update_items)
         self.bridge.reset_jobs.append(
-            lambda: self.coordinator.async_remove_listener(self.async_update_items)
+            self.coordinator.async_add_listener(self.async_update_items)
         )
         await self.coordinator.async_refresh()
 
@@ -178,14 +177,10 @@ def available(self):
 
     async def async_added_to_hass(self):
         """When entity is added to hass."""
-        self.bridge.sensor_manager.coordinator.async_add_listener(
-            self.async_write_ha_state
-        )
-
-    async def async_will_remove_from_hass(self):
-        """When entity will be removed from hass."""
-        self.bridge.sensor_manager.coordinator.async_remove_listener(
-            self.async_write_ha_state
+        self.async_on_remove(
+            self.bridge.sensor_manager.coordinator.async_add_listener(
+                self.async_write_ha_state
+            )
         )
 
     async def async_update(self):
diff --git a/homeassistant/helpers/update_coordinator.py b/homeassistant/helpers/update_coordinator.py
index b2b048166168a3..d9a79d6555c1cf 100644
--- a/homeassistant/helpers/update_coordinator.py
+++ b/homeassistant/helpers/update_coordinator.py
@@ -62,7 +62,7 @@ def __init__(
         self._debounced_refresh = request_refresh_debouncer
 
     @callback
-    def async_add_listener(self, update_callback: CALLBACK_TYPE) -> None:
+    def async_add_listener(self, update_callback: CALLBACK_TYPE) -> Callable[[], None]:
         """Listen for data updates."""
         schedule_refresh = not self._listeners
 
@@ -72,6 +72,13 @@ def async_add_listener(self, update_callback: CALLBACK_TYPE) -> None:
         if schedule_refresh:
             self._schedule_refresh()
 
+        @callback
+        def remove_listener() -> None:
+            """Remove update listener."""
+            self.async_remove_listener(update_callback)
+
+        return remove_listener
+
     @callback
     def async_remove_listener(self, update_callback: CALLBACK_TYPE) -> None:
         """Remove data update."""
diff --git a/tests/helpers/test_update_coordinator.py b/tests/helpers/test_update_coordinator.py
index c17c79ccbc8acc..17caa1f747863f 100644
--- a/tests/helpers/test_update_coordinator.py
+++ b/tests/helpers/test_update_coordinator.py
@@ -18,11 +18,12 @@
 @pytest.fixture
 def crd(hass):
     """Coordinator mock."""
-    calls = []
+    calls = 0
 
     async def refresh():
-        calls.append(None)
-        return len(calls)
+        nonlocal calls
+        calls += 1
+        return calls
 
     crd = update_coordinator.DataUpdateCoordinator(
         hass,
@@ -46,16 +47,19 @@ async def test_async_refresh(crd):
     def update_callback():
         updates.append(crd.data)
 
-    crd.async_add_listener(update_callback)
-
+    unsub = crd.async_add_listener(update_callback)
     await crd.async_refresh()
+    assert updates == [2]
 
+    # Test unsubscribing through function
+    unsub()
+    await crd.async_refresh()
     assert updates == [2]
 
+    # Test unsubscribing through method
+    crd.async_add_listener(update_callback)
     crd.async_remove_listener(update_callback)
-
     await crd.async_refresh()
-
     assert updates == [2]
 
 

<!--
  You are amazing! Thanks for contributing to our project!
  Please, DO NOT DELETE ANY TEXT from this template! (unless instructed).
-->
## Breaking change
<!--
  If your PR contains a breaking change for existing users, it is important
  to tell them what breaks, how to make it work again and why we did this.
  This piece of text is published with the release notes, so it helps if you
  write it towards our users, not us.
  Note: Remove this section if this PR is NOT a breaking change.
-->


## Proposed change
<!-- 
  Describe the big picture of your changes here to communicate to the
  maintainers why we should accept this pull request. If it fixes a bug
  or resolves a feature request, be sure to link to that issue in the
  additional information section.
-->
In #33510 I made sure that we use `Entity.async_on_remove` for cleaning up dispatcher connections. I realized that we don't support this for DataUpdateCoordinator so added this functionality and applied the clean up to the Hue integration.

## Type of change
<!--
  What type of change does your PR introduce to Home Assistant?
  NOTE: Please, check only 1! box! 
  If your PR requires multiple boxes to be checked, you'll most likely need to
  split it into multiple PRs. This makes things easier and faster to code review.
-->

- [ ] Dependency upgrade
- [ ] Bugfix (non-breaking change which fixes an issue)
- [ ] New integration (thank you!)
- [ ] New feature (which adds functionality to an existing integration)
- [ ] Breaking change (fix/feature causing existing functionality to break)
- [x] Code quality improvements to existing code or addition of tests

## Example entry for `configuration.yaml`:
<!--
  Supplying a configuration snippet, makes it easier for a maintainer to test
  your PR. Furthermore, for new integrations, it gives an impression of how
  the configuration would look like.
  Note: Remove this section if this PR does not have an example entry.
-->

```yaml
# Example configuration.yaml

```

## Additional information
<!--
  Details are important, and help maintainers processing your PR.
  Please be sure to fill out additional details, if applicable.
-->

- This PR fixes or closes issue: fixes #
- This PR is related to issue: 
- Link to documentation pull request: 

## Checklist
<!--
  Put an `x` in the boxes that apply. You can also fill these out after
  creating the PR. If you're unsure about any of them, don't hesitate to ask.
  We're here to help! This is simply a reminder of what we are going to look
  for before merging your code.
-->

- [ ] The code change is tested and works locally.
- [ ] Local tests pass. **Your PR cannot be merged unless tests pass**
- [ ] There is no commented out code in this PR.
- [ ] I have followed the [development checklist][dev-checklist]
- [ ] The code has been formatted using Black (`black --fast homeassistant tests`)
- [ ] Tests have been added to verify that the new code works.

If user exposed functionality or configuration variables are added/changed:

- [ ] Documentation added/updated for [www.home-assistant.io][docs-repository]

If the code communicates with devices, web services, or third-party tools:

- [ ] The [manifest file][manifest-docs] has all fields filled out correctly.  
      Updated and included derived files by running: `python3 -m script.hassfest`.
- [ ] New or updated dependencies have been added to `requirements_all.txt`.  
      Updated by running `python3 -m script.gen_requirements_all`.
- [ ] Untested files have been added to `.coveragerc`.

The integration reached or maintains the following [Integration Quality Scale][quality-scale]:
<!--
  The Integration Quality Scale scores an integration on the code quality
  and user experience. Each level of the quality scale consists of a list
  of requirements. We highly recommend getting your integration scored!
-->

- [ ] No score or internal
- [ ] 🥈 Silver
- [ ] 🥇 Gold
- [ ] 🏆 Platinum

<!--
  Thank you for contributing <3

  Below, some useful links you could explore:
-->
[dev-checklist]: https://developers.home-assistant.io/docs/en/development_checklist.html
[manifest-docs]: https://developers.home-assistant.io/docs/en/creating_integration_manifest.html
[quality-scale]: https://developers.home-assistant.io/docs/en/next/integration_quality_scale_index.html
[docs-repository]: https://github.com/home-assistant/home-assistant.io
