Function run_proxy_server

impit-python/test/setup_proxy.py:10–27

(stop_event: threading.Event)

def start_proxy_server(port: int = 3002) -> typing.Callable[[], None]:
    def run_proxy_server(stop_event: threading.Event) -> None:
        server = pproxy.Server(f'http://0.0.0.0:{port}')
        args = {'rserver': [], 'verbose': print}

        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        handler = loop.run_until_complete(server.start_server(args))

        try:
            while not stop_event.is_set():
                loop.run_until_complete(asyncio.sleep(0.1))
        except KeyboardInterrupt:
            print('exit!')

        handler.close()
        loop.run_until_complete(handler.wait_closed())
        loop.run_until_complete(loop.shutdown_asyncgens())
        loop.close()

    # Event to signal the proxy server to stop
    stop_event = threading.Event()