Threaded dispatcher loop

src/django_idom/websocket_consumer.py

@@ -3,6 +3,7 @@
 import json
 import logging
 from dataclasses import dataclass
+from threading import Thread
 from typing import Any, Awaitable, Callable, Optional
 from urllib.parse import parse_qsl
 
@@ -18,6 +19,11 @@
 _logger = logging.getLogger(__name__)
 
 
+def start_background_loop(loop: asyncio.AbstractEventLoop) -> None:
+    asyncio.set_event_loop(loop)
+    loop.run_forever()
+
+
 @dataclass
 class WebsocketConnection:
     scope: dict
@@ -42,17 +48,26 @@ async def connect(self) -> None:
         elif user is None:
             _logger.warning("IDOM websocket is missing AuthMiddlewareStack!")
 
-        self._idom_dispatcher_future = asyncio.ensure_future(self._run_dispatch_loop())
+        self._recv_queue_loop = asyncio.new_event_loop()
+        t = Thread(
+            target=start_background_loop, args=[self._recv_queue_loop], daemon=True
+        )
+        t.start()
+        asyncio.run_coroutine_threadsafe(
+            self._run_dispatch_loop(), self._recv_queue_loop
+        )
 
     async def disconnect(self, code: int) -> None:
-        if self._idom_dispatcher_future.done():
-            await self._idom_dispatcher_future
-        else:
-            self._idom_dispatcher_future.cancel()
+        self._recv_queue_loop.stop()
         await super().disconnect(code)
 
     async def receive_json(self, content: Any, **kwargs: Any) -> None:
-        await self._idom_recv_queue.put(LayoutEvent(**content))
+        asyncio.run_coroutine_threadsafe(
+            self._recv_queue_put(content, **kwargs), self._recv_queue_loop
+        )
+
+    async def _recv_queue_put(self, content: Any, **kwargs: Any):
+        await self._recv_queue.put(LayoutEvent(**content))
 
     async def _run_dispatch_loop(self):
         view_id = self.scope["url_route"]["kwargs"]["view_id"]
@@ -78,12 +93,12 @@ async def _run_dispatch_loop(self):
             )
             return
 
-        self._idom_recv_queue = recv_queue = asyncio.Queue()
+        self._recv_queue = asyncio.Queue()
         try:
             await dispatch_single_view(
                 Layout(component_instance),
                 self.send_json,
-                recv_queue.get,
+                self._recv_queue.get,
             )
         except Exception:
             await self.close()