Function run_server_or_router

tests/sync/server.py:40–72

(
    serve_or_route,
    handler_or_url_map,
    host="localhost",
    port=0,
    **kwargs,
)

@contextlib.contextmanager
def run_server_or_router(
    serve_or_route,
    handler_or_url_map,
    host="localhost",
    port=0,
    **kwargs,
):
    with serve_or_route(handler_or_url_map, host, port, **kwargs) as server:
        thread = threading.Thread(target=server.serve_forever)
        thread.start()

        # HACK: since the sync server doesn't track connections (yet), we record
        # a reference to the thread handling the most recent connection, then we
        # can wait for that thread to terminate when exiting the context.
        handler_thread = None
        original_handler = server.handler

        def handler(sock, addr):
            nonlocal handler_thread
            handler_thread = threading.current_thread()
            original_handler(sock, addr)

        server.handler = handler

        try:
            yield server
        finally:
            server.shutdown()
            thread.join()

            # HACK: wait for the thread handling the most recent connection.
            if handler_thread is not None:
                handler_thread.join()


def run_server(handler=handler, **kwargs):