Merge pull request #24 from briangu/codex/fix-web-server-not-launching-in-repl

briangu · web-flow · commit de0dab3b9aba · 2025-05-20T07:30:22.000-07:00
Fix web server startup from REPL
diff --git a/klongpy/repl.py b/klongpy/repl.py
@@ -0,0 +1,73 @@
+import asyncio
+import threading
+import time
+import os
+import importlib.resources
+
+from . import KlongInterpreter
+from .utils import CallbackEvent
+
+
+def start_loop(loop: asyncio.AbstractEventLoop, stop_event: asyncio.Event) -> None:
+    asyncio.set_event_loop(loop)
+    loop.run_until_complete(stop_event.wait())
+
+
+def setup_async_loop(debug: bool = False, slow_callback_duration: float = 86400.0):
+    loop = asyncio.new_event_loop()
+    loop.slow_callback_duration = slow_callback_duration
+    if debug:
+        loop.set_debug(True)
+    stop_event = asyncio.Event()
+    thread = threading.Thread(target=start_loop, args=(loop, stop_event), daemon=True)
+    thread.start()
+    return loop, thread, stop_event
+
+
+def cleanup_async_loop(loop: asyncio.AbstractEventLoop, loop_thread: threading.Thread, stop_event: asyncio.Event, debug: bool = False, name: str | None = None) -> None:
+    if loop.is_closed():
+        return
+
+    loop.call_soon_threadsafe(stop_event.set)
+    loop_thread.join()
+
+    pending_tasks = asyncio.all_tasks(loop=loop)
+    if len(pending_tasks) > 0:
+        if name:
+            print(f"WARNING: pending tasks in {name} loop")
+        for task in pending_tasks:
+            loop.call_soon_threadsafe(task.cancel)
+        while len(asyncio.all_tasks(loop=loop)) > 0:
+            time.sleep(0)
+
+    loop.stop()
+
+    if not loop.is_closed():
+        loop.close()
+
+
+def append_pkg_resource_path_KLONGPATH() -> None:
+    with importlib.resources.as_file(importlib.resources.files('klongpy')) as pkg_path:
+        pkg_lib_path = os.path.join(pkg_path, 'lib')
+        klongpath = os.environ.get('KLONGPATH', '.:lib')
+        klongpath = f"{klongpath}:{pkg_lib_path}" if klongpath else str(pkg_lib_path)
+        os.environ['KLONGPATH'] = klongpath
+
+
+def create_repl(debug: bool = False):
+    io_loop, io_thread, io_stop = setup_async_loop(debug=debug)
+    klong_loop, klong_thread, klong_stop = setup_async_loop(debug=debug)
+
+    append_pkg_resource_path_KLONGPATH()
+
+    klong = KlongInterpreter()
+    shutdown_event = CallbackEvent()
+    klong['.system'] = {'ioloop': io_loop, 'klongloop': klong_loop, 'closeEvent': shutdown_event}
+
+    return klong, (io_loop, io_thread, io_stop, klong_loop, klong_thread, klong_stop)
+
+
+def cleanup_repl(loops, debug: bool = False) -> None:
+    io_loop, io_thread, io_stop, klong_loop, klong_thread, klong_stop = loops
+    cleanup_async_loop(io_loop, io_thread, io_stop, debug=debug, name='io_loop')
+    cleanup_async_loop(klong_loop, klong_thread, klong_stop, debug=debug, name='klong_loop')
diff --git a/klongpy/web/sys_fn_web.py b/klongpy/web/sys_fn_web.py
@@ -1,5 +1,7 @@
 import logging
 import sys
+import asyncio
+import concurrent.futures
 
 from aiohttp import web
 
@@ -113,7 +115,17 @@ async def start_server():
         site = web.TCPSite(runner, bind, port)
         await site.start()
 
-    server_task = klong['.system']['ioloop'].create_task(start_server())
+    # create the server task in the ioloop thread and capture the task handle
+    server_loop = klong['.system']['ioloop']
+    task_future = concurrent.futures.Future()
+
+    def _start():
+        task = asyncio.create_task(start_server())
+        task_future.set_result(task)
+
+    server_loop.call_soon_threadsafe(_start)
+    server_task = task_future.result()
+
     return WebServerHandle(bind, port, runner, server_task)
 
 
@@ -129,7 +141,7 @@ def eval_sys_fn_shutdown_web_server(klong, x):
         x = x.a.fn
         if isinstance(x, WebServerHandle) and x.runner is not None:
             print("shutting down web server")
-            klong['.system']['ioloop'].run_until_complete(x.shutdown())
+            asyncio.run_coroutine_threadsafe(x.shutdown(), klong['.system']['ioloop']).result()
             return 1
     return 0
 
diff --git a/tests/test_sys_fn_web.py b/tests/test_sys_fn_web.py
@@ -0,0 +1,50 @@
+import asyncio
+import socket
+import unittest
+
+import aiohttp
+
+from klongpy.repl import create_repl, cleanup_repl
+
+
+class TestSysFnWeb(unittest.TestCase):
+    def setUp(self):
+        self.klong, self.loops = create_repl()
+        (self.ioloop, self.ioloop_thread, self.io_stop,
+         self.klongloop, self.klongloop_thread, self.klong_stop) = self.loops
+        self.handle = None
+
+    def tearDown(self):
+        if self.handle is not None and self.handle.task is not None:
+            asyncio.run_coroutine_threadsafe(self.handle.shutdown(), self.ioloop).result()
+        cleanup_repl(self.loops)
+
+    def _free_port(self):
+        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s.bind(("", 0))
+        port = s.getsockname()[1]
+        s.close()
+        return port
+
+    def test_web_server_start_and_stop(self):
+        klong = self.klong
+        port = self._free_port()
+
+        klong('.py("klongpy.web")')
+        klong('index::{x;"hello"}')
+        klong('get:::{}')
+        klong('get,"/",index')
+        klong('post:::{}')
+        handle = klong(f'h::.web({port};get;post)')
+        self.handle = handle
+
+        async def fetch():
+            async with aiohttp.ClientSession() as session:
+                async with session.get(f"http://localhost:{port}/") as resp:
+                    return await resp.text()
+
+        response = asyncio.run_coroutine_threadsafe(fetch(), self.ioloop).result()
+        self.assertEqual(response, "hello")
+
+        asyncio.run_coroutine_threadsafe(handle.shutdown(), self.ioloop).result()
+
