reorganize Server base class and primary subclasses

This commit is contained in:
Sam G. 2024-05-31 05:30:08 -07:00
parent d60250eb8b
commit 18c7bf1adf
3 changed files with 141 additions and 109 deletions

View File

@ -32,5 +32,6 @@ reference/documentation/index
``` ```
```{include} ../README.md ```{include} ../README.md
:relative-docs: docs/
:relative-images:
``` ```

View File

@ -990,6 +990,21 @@ class ChainRouter[E: Event](Router[E]):
return self.running_events.pop(event_idx, None) return self.running_events.pop(event_idx, None)
def get_listener(self, listener_cls=None): def get_listener(self, listener_cls=None):
'''
Gets the listener for the entire router chain.
Builds a single listener from the registered endpoints of each router, allowing
a global iNotify instance to serve all routes. This provides a clean target for
outer server context managers.
Note that having multiple iNotify instances, even those watching the same
directories, in general shouldn't pose an issue. iNotify instances are
independent, and the kernel will send events to each listening instance when one
occurs. This is why we consolidate the listener for chained routers: we want to
respond to events sequentially according to router order, rather than have each
router expose a listener that can independently hear and respond to its own
events.
'''
if listener_cls is None: if listener_cls is None:
for router in self.ordered_routers: for router in self.ordered_routers:
if router.listener_cls is not None: if router.listener_cls is not None:

View File

@ -36,59 +36,58 @@ class Server:
''' '''
Wraps up a development static file server and live reloader. Wraps up a development static file server and live reloader.
''' '''
def __init__(self):
self.server = None
# MT/MP server implementations can check this variable for graceful shutdowns
self.should_shutdown = False
self.started = False
# used to isolate server creation logic, when applicable
self._init_server()
def _init_server(self):
pass
def start(self):
self.started = True
def shutdown(self):
self.should_shutdown = True
logger.debug("Server shutdown request received")
class ProcessServer(Server):
'''
general subprocess start, shutdown, output catching logic
'''
pass
class HTTPServer(Server):
def __init__( def __init__(
self, self,
host, host,
port, port,
root, root,
static : bool = False, loop = None,
livereload : bool = False,
managed_listeners : list | None = None,
): ):
''' '''
Parameters: Parameters:
host: host server address (either 0.0.0.0, 127.0.0.1, localhost) host: host server address (either 0.0.0.0, 127.0.0.1, localhost)
port: port at which to start the server port: port at which to start the server
root: base path for static files _and_ where router bases are attached (i.e.,
when files at this path change, a reload event will be
propagated to a corresponding client page)
static: whether or not to start a static file server
livereload: whether or not to start a livereload server
managed_listeners: auxiliary listeners to "attach" to the server process, and to
propagate the shutdown signal to when the server receives an
interrupt.
''' '''
self.host = host self.host = host
self.port = port self.port = port
self.root = root self.loop = loop
self.static = static
self.livereload = livereload
if managed_listeners is None:
managed_listeners = []
self.managed_listeners = managed_listeners
self.listener = None
self.userver = None self.userver = None
self.server = None
self.server_text = ''
self.server_args = {}
self.started = False super().__init__()
self.loop = None def _init_server(self):
self._server_setup()
def _wrap_static(self):
self.server.mount("/", StaticFiles(directory=self.root), name="static")
def _wrap_livereload(self):
self.server.websocket_route('/livereload')(LREndpoint)
#self.server.add_api_websocket_route('/livereload', LREndpoint)
def _server_setup(self):
''' '''
Set up the FastAPI server. Only a single server instance is used here, optionally Set up the FastAPI server and Uvicorn hook.
Only a single server instance is used here, optionally
mounting the static route (if static serving enabled) and providing a websocket mounting the static route (if static serving enabled) and providing a websocket
endpoint (if livereload enabled). endpoint (if livereload enabled).
@ -97,6 +96,10 @@ class Server:
behave appropriately, even when remounting the root if serving static files behave appropriately, even when remounting the root if serving static files
(which, if done in the opposite order, would "eat up" the ``/livereload`` endpoint). (which, if done in the opposite order, would "eat up" the ``/livereload`` endpoint).
''' '''
if self.loop is None:
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
# enable propagation and clear handlers for uvicorn internal loggers; # enable propagation and clear handlers for uvicorn internal loggers;
# allows logging messages to propagate to my root logger # allows logging messages to propagate to my root logger
log_config = uvicorn.config.LOGGING_CONFIG log_config = uvicorn.config.LOGGING_CONFIG
@ -107,47 +110,20 @@ class Server:
log_config['loggers']['uvicorn.error']['propagate'] = False log_config['loggers']['uvicorn.error']['propagate'] = False
log_config['loggers']['uvicorn.error']['handlers'] = [] log_config['loggers']['uvicorn.error']['handlers'] = []
self.server_args['log_config'] = log_config server_args = {}
self.server_args['host'] = self.host server_args['log_config'] = log_config
self.server_args['port'] = self.port server_args['host'] = self.host
server_args['port'] = self.port
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI): async def lifespan(app: FastAPI):
yield yield
self.shutdown() self.shutdown()
if self.static or self.livereload:
self.server = FastAPI(lifespan=lifespan) self.server = FastAPI(lifespan=lifespan)
#self.server.on_event('shutdown')(self.shutdown)
if self.livereload: uconfig = uvicorn.Config(app=self.server, loop=self.loop, **self.server_args)
self._wrap_livereload() self.userver = uvicorn.Server(config=uconfig)
self._listener_setup()
self.server_text += '+reload'
if self.static:
self._wrap_static()
self.server_text += '+static'
def _listener_setup(self):
'''
flags.MODIFY okay since we don't need to reload non-existent pages
'''
if self.loop is None:
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
#self.listener = listener.WatchFS(loop=self.loop)
self.router = PathRouter(loop=self.loop)
self.router.register(
path=str(self.root),
func=LREndpoint.reload_clients,
delay=100,
flags=flags.MODIFY,
)
self.listener = self.router.get_listener()
self.managed_listeners.append(self.listener)
def start(self): def start(self):
''' '''
@ -220,22 +196,10 @@ class Server:
and shut down their thread pools, gracefully close up the Uvicorn server and and shut down their thread pools, gracefully close up the Uvicorn server and
allow the serve coroutine to complete, and finally close down the event loop. allow the serve coroutine to complete, and finally close down the event loop.
''' '''
if self.loop is None: super().start()
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
for listener in self.managed_listeners: logger.info(f'Starting HTTP server @ http://{self.host}:{self.port}')
#loop.run_in_executor(None, partial(self.listener.start, loop=loop))
if not listener.started:
listener.start()
self.started = False
if self.server:
logger.info(f'Server{self.server_text} @ http://{self.host}:{self.port}')
uconfig = uvicorn.Config(app=self.server, loop=self.loop, **self.server_args)
self.userver = uvicorn.Server(config=uconfig)
self.loop.run_until_complete(self.userver.serve()) self.loop.run_until_complete(self.userver.serve())
self.loop.close() self.loop.close()
@ -279,15 +243,6 @@ class Server:
task should be considered "completed," at which point the event loop can be closed task should be considered "completed," at which point the event loop can be closed
successfully. successfully.
''' '''
logger.info("Shutting down server...")
# stop attached auxiliary listeners, both internal & external
if self.managed_listeners:
logger.info(f"Stopping {len(self.managed_listeners)} listeners...")
for listener in self.managed_listeners:
listener.stop()
# stop FastAPI server if started # stop FastAPI server if started
if self.userver is not None: if self.userver is not None:
def set_should_exit(): def set_should_exit():
@ -295,7 +250,39 @@ class Server:
self.loop.call_soon_threadsafe(set_should_exit) self.loop.call_soon_threadsafe(set_should_exit)
class ListenerServer: class StaticHTTPServer(Server):
def __init__(
self,
root,
*args,
**kwargs
):
'''
Parameters:
root: base path for static files _and_ where router bases are attached (i.e.,
when files at this path change, a reload event will be propagated to a
corresponding client page)
'''
self.root = root
super().__init__(*args, **kwargs)
def _init_server(self):
super()._init_server()
self.server.mount(
"/",
StaticFiles(directory=self.root),
name="static"
)
class LiveReloadHTTPServer(Server):
def _init_server(self):
super()._init_server()
self.server.websocket_route('/livereload')(LREndpoint)
class ListenerServer(Server):
''' '''
Server abstraction to handle disparate listeners. Server abstraction to handle disparate listeners.
''' '''
@ -303,24 +290,53 @@ class ListenerServer:
self, self,
managed_listeners : list | None = None, managed_listeners : list | None = None,
): ):
'''
Parameters:
managed_listeners: auxiliary listeners to "attach" to the server process, and to
propagate the shutdown signal to when the server receives an
interrupt.
'''
super().__init__()
if managed_listeners is None: if managed_listeners is None:
managed_listeners = [] managed_listeners = []
self.managed_listeners = managed_listeners self.managed_listeners = managed_listeners
self.started = False
def _listener_setup(self):
'''
flags.MODIFY okay since we don't need to reload non-existent pages
'''
if self.loop is None:
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
#self.listener = listener.WatchFS(loop=self.loop)
self.router = PathRouter(loop=self.loop)
self.router.register(
path=str(self.root),
func=LREndpoint.reload_clients,
delay=100,
flags=flags.MODIFY,
)
self.listener = self.router.get_listener()
self.managed_listeners.append(self.listener)
def start(self): def start(self):
super().start()
for listener in self.managed_listeners: for listener in self.managed_listeners:
#loop.run_in_executor(None, partial(self.listener.start, loop=loop)) #loop.run_in_executor(None, partial(self.listener.start, loop=loop))
if not listener.started: if not listener.started:
listener.start() listener.start()
self.started = True
for listener in self.managed_listeners: for listener in self.managed_listeners:
listener.join() listener.join()
def shutdown(self): def shutdown(self):
super().shutdown()
# stop attached auxiliary listeners, both internal & external # stop attached auxiliary listeners, both internal & external
if self.managed_listeners: if self.managed_listeners:
logger.info(f"Stopping {len(self.managed_listeners)} listeners...") logger.info(f"Stopping {len(self.managed_listeners)} listeners...")