Make connection handling more explicit and remove outdated examples

README.md

@@ -27,6 +27,7 @@ to start a simple redis websocket api on `ws://localhost:8765`.
 This does [roughly](./redis_websocket_api/__main__.py) the equivalant of:
 
 ```python
+import asyncio
 from aioredis import from_url
 from redis_websocket_api import WebsocketServer, WebsocketHandler
 
@@ -37,20 +38,20 @@ class PublishEverythingHandler(WebsocketHandler):
         return True
 
 
-WebsocketServer(
+ws_server = WebsocketServer(
     redis=from_url("redis:///", encoding="utf-8", decode_responses=True),
-    read_timeout=30,
     keep_alive_timeout=120,
     handler_class=PublishEverythingHandler,
-).listen(
-    host='localhost',
-    port=8000,
-    channel_patterns=["[a-z]*"],
 )
-```
 
-Have a look at `examples/demo.py` for an example with the `GeoCommandsMixin`
-added.
+asyncio.run(
+    ws_server.serve(
+        host='localhost',
+        port=8000,
+        channel_patterns=["[a-z]*"],
+    )
+)
+```
 
 
 Client-Side Usage

redis_websocket_api/__main__.py

@@ -5,6 +5,7 @@
 Intended for debugging and development only.
 """
 
+import asyncio
 from os import getenv
 from logging import basicConfig, INFO
 import aioredis
@@ -22,10 +23,13 @@ def main():
     redis = aioredis.from_url(
         getenv("REDIS_DSN", "redis:///"), encoding="utf-8", decode_responses=True
     )
-    server = WebsocketServer(redis=redis, handler_class=PublishEverythingHandler,)
+    server = WebsocketServer(
+        redis=redis,
+        handler_class=PublishEverythingHandler,
+    )
     host = getenv("HOST", "localhost")
     port = int(getenv("PORT", 8765))
-    server.listen(host, port, channel_patterns=["[a-z]*"])
+    asyncio.run(server.serve(host, port, channel_patterns=["[a-z]*"]))
 
 
 if __name__ == "__main__":

redis_websocket_api/handler.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import asyncio
 from collections import OrderedDict
 from logging import getLogger
@@ -15,18 +17,26 @@
 class WebsocketHandlerBase:
     """Define protocol for communication between web client and server."""
 
-    read_timeout = NotImplemented
     allowed_commands = NotImplemented
 
+    consumer_task: asyncio.Task  # for backwrds-compatibility, also in:
+    tasks: dict[str, asyncio.Task]  # Self-made TaskGroup for old Python
+
     def __init__(self, redis, websocket, read_timeout=None):
+        if read_timeout:
+            logger.warning(
+                "read_timeout is not used anymore because cleanup is triggered "
+                "immediately on connection loss"
+            )
+
         self.websocket = websocket
         self.redis = redis
-        self.read_timeout = read_timeout or self.read_timeout
 
         self.queue = asyncio.Queue()
 
         self.filters = OrderedDict()
         self.subscriptions = set()
+        self.tasks = {}
 
     async def _websocket_reader(self):
         try:
@@ -62,7 +72,7 @@ def _apply_filters(self, message, exclude=()):
         return passes, data
 
     async def _queue_reader(self):
-        source, message = await asyncio.wait_for(self.queue.get(), self.read_timeout)
+        source, message = await self.queue.get()
         if source == "websocket":
             await self._handle_remote_message(message)
         else:
@@ -131,35 +141,41 @@ def channel_is_allowed(self, channel_name):
     @classmethod
     async def create(cls, redis, websocket, read_timeout=None):
         """Create a handler instance setting up tasks and queues."""
-        self = cls(redis, websocket, read_timeout=read_timeout)
-        self.consumer_task = asyncio.ensure_future(self._websocket_reader())
+
+        if read_timeout:
+            logger.warning(
+                "read_timeout is not used anymore because cleanup is triggered "
+                "immediately on connection loss"
+            )
+
+        self = cls(redis, websocket)
+        self.consumer_task = asyncio.create_task(self._websocket_reader())
+        self.tasks["consumer_task"] = self.consumer_task
         return self
 
     async def listen(self):
-        """Read and handle messages from internal message queue.
-
-        This coroutine blocks for up to self.read_timeout seconds.
-        """
+        """Read and handle messages from internal message queue"""
         await self._send("websocket", {"status": "open"})
-        while not self.consumer_task.done():
-            try:
-                await self._queue_reader()
-            except asyncio.TimeoutError:
-                if not self.websocket.open:
-                    raise
+        while self.websocket.open:
+            queue_reader_task = asyncio.create_task(self._queue_reader())
+            self.tasks["queue_reader"] = queue_reader_task
+            await queue_reader_task
 
     async def close(self):
         """Close all connections and cancel all tasks."""
-        asyncio.wait_for(  # attempt to say goodbye to the client
-            self.websocket.close(), self.read_timeout
-        )
-        self.consumer_task.cancel()
+        if self.websocket.open:
+            await self.websocket.close()
+        for task in self.tasks.values():
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
 
 
 class WebsocketHandler(WebsocketHandlerBase, CommandsMixin):
     """Provides a Redis proxy to predefined channels"""
 
-    read_timeout = 30
     allowed_commands = "SUB", "DEL", "PING", "GET"
     channel_names = set()
 

redis_websocket_api/protocol.py

@@ -16,7 +16,7 @@ class CommandsMixin:
 
     The commend name is translated to a method name like this:
 
-        "_handle_{name}_command".format(name=a_tralis_protocol_command.lower())
+        "_handle_{name}_command".format(name=a_custom_protocol_command.lower())
     """
 
     async def _handle_del_command(self, channel_name):