cswimr/SeaCogs
cswimr/SeaCogs
1
1
Fork 1
Code Issues 11 Pull requests 9 Projects 1 Wiki Activity Actions

feat(hotreload): first pass at debounce support

Browse source
This commit is contained in:
cswimr 2025-02-18 14:57:31 -06:00
parent 346963fd4f
commit 5d92abb079
Signed by: cswimr
GPG key ID: 0EC431A8DA8F8087
1 changed files with 11 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

12
hotreload/hotreload.py
View file

@ -122,11 +122,18 @@ class HotReloadHandler(RegexMatchingEventHandler):
self.cog: HotReload = cog self.cog: HotReload = cog
self.path: Path = path self.path: Path = path
self.logger: RedTraceLogger = getLogger(name="red.SeaCogs.HotReload.Observer") self.logger: RedTraceLogger = getLogger(name="red.SeaCogs.HotReload.Observer")
self.debounce: bool = False
def on_any_event(self, event: FileSystemEvent) -> None: def on_any_event(self, event: FileSystemEvent) -> None:
"""Handle filesystem events.""" """Handle filesystem events."""
if event.is_directory: if event.is_directory:
return return
if self.debounce:
self.logger.trace("Debouncing enabled, ignoring event '%s'", event.event_type)
return
self.debounce = True
self.logger.verbose("Starting cog reload process, enabling debouncing during event '%s'", event.event_type)
allowed_events = ("moved", "deleted", "created", "modified") allowed_events = ("moved", "deleted", "created", "modified")
if event.event_type not in allowed_events: if event.event_type not in allowed_events:
@ -147,13 +154,16 @@ class HotReloadHandler(RegexMatchingEventHandler):
self.logger.info("File %s has been %s%s.", event.src_path, event.event_type, dest) self.logger.info("File %s has been %s%s.", event.src_path, event.event_type, dest)
run_coroutine_threadsafe( future = run_coroutine_threadsafe(
coro=self.reload_cogs( coro=self.reload_cogs(
cog_names=cogs_to_reload, cog_names=cogs_to_reload,
paths=[Path(str(p)) for p in (event.src_path, getattr(event, "dest_path", None)) if p], paths=[Path(str(p)) for p in (event.src_path, getattr(event, "dest_path", None)) if p],
), ),
loop=self.cog.bot.loop, loop=self.cog.bot.loop,
) )
if future.done():
self.debounce = False
self.logger.verbose("Debouncing disabled, event '%s' has been handled", event.event_type)
async def reload_cogs(self, cog_names: Sequence[str], paths: Sequence[Path]) -> None: async def reload_cogs(self, cog_names: Sequence[str], paths: Sequence[Path]) -> None:
"""Reload modified cogs.""" """Reload modified cogs."""