Add ability to shutdown a Debouncer (#91439)

* Add ability to shutdown a Debouncer

* Use async_create_task
This commit is contained in:
epenet 2023-04-15 03:01:21 +02:00 committed by GitHub
parent 698345e88b
commit 19a6530c3c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 52 additions and 4 deletions

View file

@ -44,6 +44,7 @@ class Debouncer(Generic[_R_co]):
function, f"debouncer cooldown={cooldown}, immediate={immediate}"
)
)
self._shutdown_requested = False
@property
def function(self) -> Callable[[], _R_co] | None:
@ -62,6 +63,8 @@ class Debouncer(Generic[_R_co]):
async def async_call(self) -> None:
"""Call the function."""
if self._shutdown_requested:
raise RuntimeError("Debouncer has been shutdown.")
assert self._job is not None
if self._timer_task:
@ -115,6 +118,12 @@ class Debouncer(Generic[_R_co]):
# Schedule a new timer to prevent new runs during cooldown
self._schedule_timer()
@callback
def async_shutdown(self) -> None:
"""Cancel any scheduled call, and prevent new runs."""
self._shutdown_requested = True
self.async_cancel()
@callback
def async_cancel(self) -> None:
"""Cancel any scheduled call."""
@ -137,4 +146,7 @@ class Debouncer(Generic[_R_co]):
@callback
def _schedule_timer(self) -> None:
"""Schedule a timer."""
self._timer_task = self.hass.loop.call_later(self.cooldown, self._on_debounce)
if not self._shutdown_requested:
self._timer_task = self.hass.loop.call_later(
self.cooldown, self._on_debounce
)