Add two new events on the reloader process (#2413)

This commit is contained in:
Adam Hopkins
2022-03-22 23:29:39 +02:00
committed by GitHub
parent 0c9df02e66
commit 2a8e91052f
4 changed files with 97 additions and 7 deletions

View File

@@ -17,6 +17,8 @@ class ListenerEvent(str, Enum):
AFTER_SERVER_STOP = "server.shutdown.after"
MAIN_PROCESS_START = auto()
MAIN_PROCESS_STOP = auto()
RELOAD_PROCESS_START = auto()
RELOAD_PROCESS_STOP = auto()
class ListenerMixin(metaclass=SanicMeta):
@@ -73,6 +75,16 @@ class ListenerMixin(metaclass=SanicMeta):
) -> ListenerType[Sanic]:
return self.listener(listener, "main_process_stop")
def reload_process_start(
self, listener: ListenerType[Sanic]
) -> ListenerType[Sanic]:
return self.listener(listener, "reload_process_start")
def reload_process_stop(
self, listener: ListenerType[Sanic]
) -> ListenerType[Sanic]:
return self.listener(listener, "reload_process_stop")
def before_server_start(
self, listener: ListenerType[Sanic]
) -> ListenerType[Sanic]:

View File

@@ -11,6 +11,7 @@ from asyncio import (
all_tasks,
get_event_loop,
get_running_loop,
new_event_loop,
)
from contextlib import suppress
from functools import partial
@@ -32,6 +33,7 @@ from sanic.models.handler_types import ListenerType
from sanic.server import Signal as ServerSignal
from sanic.server import try_use_uvloop
from sanic.server.async_server import AsyncioServer
from sanic.server.events import trigger_events
from sanic.server.protocols.http_protocol import HttpProtocol
from sanic.server.protocols.websocket_protocol import WebSocketProtocol
from sanic.server.runners import serve, serve_multiple, serve_single
@@ -538,15 +540,21 @@ class RunnerMixin(metaclass=SanicMeta):
except IndexError:
raise RuntimeError("Did not find any applications.")
reloader_start = primary.listeners.get("reload_process_start")
reloader_stop = primary.listeners.get("reload_process_stop")
# We want to run auto_reload if ANY of the applications have it enabled
if (
cls.should_auto_reload()
and os.environ.get("SANIC_SERVER_RUNNING") != "true"
):
): # no cov
loop = new_event_loop()
trigger_events(reloader_start, loop)
reload_dirs: Set[Path] = primary.state.reload_dirs.union(
*(app.state.reload_dirs for app in apps)
)
return reloader_helpers.watchdog(1.0, reload_dirs)
reloader_helpers.watchdog(1.0, reload_dirs)
trigger_events(reloader_stop, loop)
return
# This exists primarily for unit testing
if not primary.state.server_info: # no cov