Sync with fork

.github/FUNDING.yml

@@ -0,0 +1 @@
+github: encode

.github/workflows/publish.yml

@@ -0,0 +1,29 @@
+name: Publish
+
+on:
+  push:
+    tags:
+      - "*"
+
+jobs:
+  publish:
+    name: "Publish release"
+    runs-on: "ubuntu-latest"
+
+    steps:
+      - uses: "actions/checkout@v3"
+      - uses: "actions/setup-python@v4"
+        with:
+          python-version: "3.10"
+
+      - name: "Install dependencies"
+        run: "scripts/install"
+
+      - name: "Build package & docs"
+        run: "scripts/build"
+
+      - name: "Publish to PyPI & deploy docs"
+        run: "scripts/publish"
+        env:
+          TWINE_USERNAME: __token__
+          TWINE_PASSWORD: ${{ secrets.PYPI_TOKEN }}

.github/workflows/test-suite.yml

@@ -14,7 +14,7 @@ jobs:
 
     strategy:
       matrix:
-        python-version: ["3.7", "3.8", "3.9"]
+        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
 
     services:
       zookeeper:
@@ -66,4 +66,4 @@ jobs:
       - name: "Run tests"
         run: "scripts/test"
       - name: "Enforce coverage"
-        run: "scripts/coverage"
+        run: "scripts/coverage"

.gitignore

@@ -3,5 +3,7 @@ test.db
 .coverage
 .pytest_cache/
 .mypy_cache/
-starlette.egg-info/
+*.egg-info/
 venv/
+build/
+dist/

README.md

@@ -3,7 +3,7 @@
 Broadcaster helps you develop realtime streaming functionality by providing
 a simple broadcast API onto a number of different backend services.
 
-It currently supports [Redis PUB/SUB](https://redis.io/topics/pubsub), [Apache Kafka](https://kafka.apache.org/), and [Postgres LISTEN/NOTIFY](https://www.postgresql.org/docs/current/sql-notify.html), plus a simple in-memory backend, that you can use for local development or during testing.
+It currently supports [Redis PUB/SUB](https://redis.io/topics/pubsub), [Redis Streams](https://redis.io/docs/latest/develop/data-types/streams/), [Apache Kafka](https://kafka.apache.org/), and [Postgres LISTEN/NOTIFY](https://www.postgresql.org/docs/current/sql-notify.html), plus a simple in-memory backend, that you can use for local development or during testing.
 
 <img src="https://raw.githubusercontent.com/encode/broadcaster/master/docs/demo.gif" alt='WebSockets Demo'>
 
@@ -14,9 +14,9 @@ Here's a complete example of the backend code for a simple websocket chat app:
 ```python
 # Requires: `starlette`, `uvicorn`, `jinja2`
 # Run with `uvicorn example:app`
+import anyio
 from broadcaster import Broadcast
 from starlette.applications import Starlette
-from starlette.concurrency import run_until_first_complete
 from starlette.routing import Route, WebSocketRoute
 from starlette.templating import Jinja2Templates
 
@@ -33,10 +33,15 @@ async def homepage(request):
 
 async def chatroom_ws(websocket):
     await websocket.accept()
-    await run_until_first_complete(
-        (chatroom_ws_receiver, {"websocket": websocket}),
-        (chatroom_ws_sender, {"websocket": websocket}),
-    )
+
+    async with anyio.create_task_group() as task_group:
+        # run until first is complete
+        async def run_chatroom_ws_receiver() -> None:
+            await chatroom_ws_receiver(websocket=websocket)
+            task_group.cancel_scope.cancel()
+
+        task_group.start_soon(run_chatroom_ws_receiver)
+        await chatroom_ws_sender(websocket)
 
 
 async def chatroom_ws_receiver(websocket):
@@ -65,7 +70,7 @@ The HTML template for the front end [is available here](https://github.com/encod
 
 ## Requirements
 
-Python 3.7+
+Python 3.8+
 
 ## Installation
 
@@ -78,19 +83,42 @@ Python 3.7+
 
 * `Broadcast('memory://')`
 * `Broadcast("redis://localhost:6379")`
+* `Broadcast("redis-stream://localhost:6379")`
 * `Broadcast("postgres://localhost:5432/broadcaster")`
 * `Broadcast("kafka://localhost:9092")`
 
+
+### Using custom backends
+
+You can create your own backend and use it with `broadcaster`.
+To do that you need to create a class which extends from `BroadcastBackend`
+and pass it to the `broadcaster` via `backend` argument.
+
+```python
+from broadcaster import Broadcaster, BroadcastBackend
+
+class MyBackend(BroadcastBackend):
+
+broadcaster = Broadcaster(backend=MyBackend())
+```
+
 ## Where next?
 
 At the moment `broadcaster` is in Alpha, and should be considered a working design document.
 
 The API should be considered subject to change. If you *do* want to use Broadcaster in its current
-state, make sure to strictly pin your requirements to `broadcaster==0.2.0`.
+state, make sure to strictly pin your requirements to `broadcaster==0.3.0`.
 
 To be more capable we'd really want to add some additional backends, provide API support for reading recent event history from persistent stores, and provide a serialization/deserialization API...
 
 * Serialization / deserialization to support broadcasting structured data.
-* Backends for Redis Streams, Apache Kafka, and RabbitMQ.
+* A backend for RabbitMQ.
 * Add support for `subscribe('chatroom', history=100)` for backends which provide persistence. (Redis Streams, Apache Kafka) This will allow applications to subscribe to channel updates, while also being given an initial window onto the most recent events. We *might* also want to support some basic paging operations, to allow applications to scan back in the event history.
 * Support for pattern subscribes in backends that support it.
+
+## Third Party Packages
+
+### MQTT backend
+[Gist](https://gist.github.com/alex-oleshkevich/68411a0e7ad24d53afd28c3fa5da468c)
+
+Integrates MQTT with Broadcaster

broadcaster/__init__.py

@@ -1,4 +1,5 @@
 from ._base import Broadcast, Event
+from .backends.base import BroadcastBackend
 
-__version__ = "0.2.0"
-__all__ = ["Broadcast", "Event"]
+__version__ = "0.3.1"
+__all__ = ["Broadcast", "Event", "BroadcastBackend"]

broadcaster/_base.py

@@ -1,20 +1,21 @@
+from __future__ import annotations
+
 import asyncio
 from contextlib import asynccontextmanager
-from typing import Any, AsyncGenerator, AsyncIterator, Dict, Optional
+from typing import TYPE_CHECKING, Any, AsyncGenerator, AsyncIterator, cast
 from urllib.parse import urlparse
 
+if TYPE_CHECKING:  # pragma: no cover
+    from broadcaster.backends.base import BroadcastBackend
+
 
 class Event:
     def __init__(self, channel: str, message: str) -> None:
         self.channel = channel
         self.message = message
 
     def __eq__(self, other: object) -> bool:
-        return (
-            isinstance(other, Event)
-            and self.channel == other.channel
-            and self.message == other.message
-        )
+        return isinstance(other, Event) and self.channel == other.channel and self.message == other.message
 
     def __repr__(self) -> str:
         return f"Event(channel={self.channel!r}, message={self.message!r})"
@@ -25,33 +26,40 @@ class Unsubscribed(Exception):
 
 
 class Broadcast:
-    def __init__(self, url: str):
-        from broadcaster._backends.base import BroadcastBackend
+    def __init__(self, url: str | None = None, *, backend: BroadcastBackend | None = None) -> None:
+        assert url or backend, "Either `url` or `backend` must be provided."
+        self._backend = backend or self._create_backend(cast(str, url))
+        self._subscribers: dict[str, set[asyncio.Queue[Event | None]]] = {}
 
+    def _create_backend(self, url: str) -> BroadcastBackend:
         parsed_url = urlparse(url)
-        self._backend: BroadcastBackend
-        self._subscribers: Dict[str, Any] = {}
-        if parsed_url.scheme == "redis":
-            from broadcaster._backends.redis import RedisBackend
+        if parsed_url.scheme in ("redis", "rediss"):
+            from broadcaster.backends.redis import RedisBackend
+
+            return RedisBackend(url)
+
+        elif parsed_url.scheme == "redis-stream":
+            from broadcaster.backends.redis import RedisStreamBackend
 
-            self._backend = RedisBackend(url)
+            return RedisStreamBackend(url)
 
         elif parsed_url.scheme in ("postgres", "postgresql"):
-            from broadcaster._backends.postgres import PostgresBackend
+            from broadcaster.backends.postgres import PostgresBackend
 
-            self._backend = PostgresBackend(url)
+            return PostgresBackend(url)
 
         if parsed_url.scheme == "kafka":
-            from broadcaster._backends.kafka import KafkaBackend
+            from broadcaster.backends.kafka import KafkaBackend
 
-            self._backend = KafkaBackend(url)
+            return KafkaBackend(url)
 
         elif parsed_url.scheme == "memory":
-            from broadcaster._backends.memory import MemoryBackend
+            from broadcaster.backends.memory import MemoryBackend
 
-            self._backend = MemoryBackend(url)
+            return MemoryBackend(url)
+        raise ValueError(f"Unsupported backend: {parsed_url.scheme}")
 
-    async def __aenter__(self) -> "Broadcast":
+    async def __aenter__(self) -> Broadcast:
         await self.connect()
         return self
 
@@ -79,31 +87,30 @@ async def publish(self, channel: str, message: Any) -> None:
         await self._backend.publish(channel, message)
 
     @asynccontextmanager
-    async def subscribe(self, channel: str) -> AsyncIterator["Subscriber"]:
-        queue: asyncio.Queue = asyncio.Queue()
+    async def subscribe(self, channel: str) -> AsyncIterator[Subscriber]:
+        queue: asyncio.Queue[Event | None] = asyncio.Queue()
 
         try:
             if not self._subscribers.get(channel):
                 await self._backend.subscribe(channel)
-                self._subscribers[channel] = set([queue])
+                self._subscribers[channel] = {queue}
             else:
                 self._subscribers[channel].add(queue)
 
             yield Subscriber(queue)
-
+        finally:
             self._subscribers[channel].remove(queue)
             if not self._subscribers.get(channel):
                 del self._subscribers[channel]
                 await self._backend.unsubscribe(channel)
-        finally:
             await queue.put(None)
 
 
 class Subscriber:
-    def __init__(self, queue: asyncio.Queue) -> None:
+    def __init__(self, queue: asyncio.Queue[Event | None]) -> None:
         self._queue = queue
 
-    async def __aiter__(self) -> Optional[AsyncGenerator]:
+    async def __aiter__(self) -> AsyncGenerator[Event | None, None]:
         try:
             while True:
                 yield await self.get()

broadcaster/_backends/base.py → broadcaster/backends/base.py

@@ -13,10 +13,10 @@ async def connect(self) -> None:
     async def disconnect(self) -> None:
         raise NotImplementedError()
 
-    async def subscribe(self, group: str) -> None:
+    async def subscribe(self, channel: str) -> None:
         raise NotImplementedError()
 
-    async def unsubscribe(self, group: str) -> None:
+    async def unsubscribe(self, channel: str) -> None:
         raise NotImplementedError()
 
     async def publish(self, channel: str, message: Any) -> None: