fix(ws): per-instance event loop instead of module-level global (fixes #119, fixes #133)

lark_oapi/channel/tests/test_ws_client_loop.py

@@ -0,0 +1,55 @@
+"""ws.Client event-loop isolation (issues #119 / #133).
+
+Regression tests: each ws.Client must own a dedicated event loop (created
+lazily, per instance) instead of sharing a module-level global loop, so
+multi-bot setups work and a client constructed inside an already-running
+loop is not bound to it.
+"""
+
+import asyncio
+
+from lark_oapi.ws.client import Client
+
+
+def make_client(app_id="cli_0000000000000001"):
+    return Client(app_id=app_id, app_secret="secret")
+
+
+def test_client_owns_a_dedicated_loop():
+    c1 = make_client()
+    c2 = make_client(app_id="cli_0000000000000002")
+    l1 = c1._get_loop()
+    l2 = c2._get_loop()
+    assert l1 is not None
+    # distinct instances never share a loop (the multi-bot race in #119)
+    assert l1 is not l2
+    # stable per instance
+    assert c1._get_loop() is l1
+    assert c2._get_loop() is l2
+    l1.close()
+    l2.close()
+
+
+def test_no_module_level_loop_capture():
+    import lark_oapi.ws.client as ws_mod
+
+    assert not hasattr(ws_mod, "loop")
+
+
+def test_loop_created_inside_running_loop_is_dedicated():
+    async def inner():
+        c = make_client()
+        loop = c._get_loop()
+        # never the caller's running loop (#133)
+        assert loop is not asyncio.get_running_loop()
+        loop.close()
+
+    asyncio.run(inner())
+
+
+def test_lock_created_lazily():
+    c = make_client()
+    assert c._lock is None
+    lock = c._get_lock()
+    assert isinstance(lock, asyncio.Lock)
+    assert c._lock is lock

lark_oapi/ws/client.py

@@ -28,12 +28,6 @@
 from lark_oapi.ws.pb.google.protobuf.internal.containers import RepeatedCompositeFieldContainer
 from lark_oapi.ws.pb.pbbp2_pb2 import Frame
 
-try:
-    loop = asyncio.get_event_loop()
-except RuntimeError:
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(loop)
-
 
 def _get_by_key(headers: RepeatedCompositeFieldContainer, key: str) -> str:
     for header in headers:
@@ -151,7 +145,13 @@ def __init__(self,
         self._reconnect_interval: int = 120
         self._ping_interval: int = 120
         self._cache: ExpiringCache = ExpiringCache(clear_interval=30)
-        self._lock = asyncio.Lock()
+        # Event loop owned by this client instance (created lazily on first
+        # use). Previously a module-level global was shared by every client,
+        # which broke multi-bot setups (issues #119 / #133).
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+        # asyncio primitives bind to an event loop; created lazily on the
+        # client's own loop so it never binds to a foreign loop.
+        self._lock: Optional[asyncio.Lock] = None
         # Observer hooks for higher-level wrappers (e.g. FeishuChannel) to
         # react to reconnect lifecycle. ``on_reconnecting`` fires when the
         # client decides a connection was lost and starts retrying;
@@ -161,7 +161,28 @@ def __init__(self,
         self.on_reconnected: Callable[[], None] = lambda: None
         logger.setLevel(log_level.value)
 
+    def _get_loop(self) -> asyncio.AbstractEventLoop:
+        """Return the event loop this client runs on.
+
+        Every client owns a dedicated loop, created lazily on first use, so
+        multiple clients (multi-bot, one thread per bot) never share a loop,
+        and a client constructed inside an already-running loop (e.g. inside
+        ``asyncio.run``) is not tied to it. See issues #119 / #133.
+        """
+        if self._loop is None or self._loop.is_closed():
+            self._loop = asyncio.new_event_loop()
+        return self._loop
+
+    def _get_lock(self) -> asyncio.Lock:
+        # asyncio primitives bind to the running loop when created; create the
+        # lock lazily (from a coroutine running on the client's own loop) so
+        # it never binds to a foreign loop.
+        if self._lock is None:
+            self._lock = asyncio.Lock()
+        return self._lock
+
     def start(self) -> None:
+        loop = self._get_loop()
         try:
             loop.run_until_complete(self._connect())
         except ClientException as e:
@@ -191,7 +212,8 @@ async def _ping_loop(self):
                 await asyncio.sleep(self._ping_interval)
 
     async def _connect(self) -> None:
-        await self._lock.acquire()
+        lock = self._get_lock()
+        await lock.acquire()
         if self._conn is not None:
             return
         try:
@@ -208,19 +230,19 @@ async def _connect(self) -> None:
             self._service_id = service_id
 
             logger.info(self._fmt_log("connected to {}", conn_url))
-            loop.create_task(self._receive_message_loop())
+            asyncio.get_running_loop().create_task(self._receive_message_loop())
         except InvalidHandshake as e:
             _parse_ws_conn_exception(e)
         finally:
-            self._lock.release()
+            lock.release()
 
     async def _receive_message_loop(self):
         try:
             while True:
                 if self._conn is None:
                     raise ConnectionClosedException("connection is closed")
                 msg = await self._conn.recv()
-                loop.create_task(self._handle_message(msg))
+                asyncio.get_running_loop().create_task(self._handle_message(msg))
         except Exception as e:
             logger.error(self._fmt_log("receive message loop exit, err: {}", e))
             await self._disconnect()