From e66e5ee9ee727b6c3541251f96e943da2decc98e Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Fri, 19 Aug 2022 21:59:27 +0200
Subject: [PATCH 01/33] Ignore *.egg.info (#71)

---
 .gitignore | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.gitignore b/.gitignore
index 7b5d431..fe79e5d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -3,5 +3,5 @@ test.db
 .coverage
 .pytest_cache/
 .mypy_cache/
-starlette.egg-info/
+*.egg-info/
 venv/

From 40af624b5cd1a5aab3889f4994bc8c3711756e6e Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Fri, 19 Aug 2022 22:54:15 +0200
Subject: [PATCH 02/33] Remove await from Kafka `unsubscribe` (#72)

---
 broadcaster/_backends/kafka.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/broadcaster/_backends/kafka.py b/broadcaster/_backends/kafka.py
index a3df086..a244763 100644
--- a/broadcaster/_backends/kafka.py
+++ b/broadcaster/_backends/kafka.py
@@ -29,7 +29,7 @@ async def subscribe(self, channel: str) -> None:
         self._consumer.subscribe(topics=self._consumer_channels)
 
     async def unsubscribe(self, channel: str) -> None:
-        await self._consumer.unsubscribe()
+        self._consumer.unsubscribe()
 
     async def publish(self, channel: str, message: typing.Any) -> None:
         await self._producer.send_and_wait(channel, message.encode("utf8"))

From 8c0382d2a33f67a5bf877b2cff3fd797c34e8a5d Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Sat, 20 Aug 2022 07:45:42 +0200
Subject: [PATCH 03/33] Replace references to httpx by broadcaster (#73)

Co-authored-by: Pablo Woolvett <17148684+pwoolvett@users.noreply.github.com>
---
 .gitignore    | 2 ++
 scripts/check | 4 ++--
 scripts/lint  | 6 +++---
 setup.cfg     | 2 +-
 4 files changed, 8 insertions(+), 6 deletions(-)

diff --git a/.gitignore b/.gitignore
index fe79e5d..013870b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,3 +5,5 @@ test.db
 .mypy_cache/
 *.egg-info/
 venv/
+build/
+dist/
diff --git a/scripts/check b/scripts/check
index 77acf65..5ee0190 100755
--- a/scripts/check
+++ b/scripts/check
@@ -1,7 +1,7 @@
 #!/bin/sh -e
 
 export PREFIX=""
-if [ -d 'venv' ] ; then
+if [ -d 'venv' ]; then
     export PREFIX="venv/bin/"
 fi
 export SOURCE_FILES="broadcaster tests"
@@ -11,4 +11,4 @@ set -x
 ${PREFIX}black --check --diff --target-version=py37 $SOURCE_FILES
 ${PREFIX}flake8 $SOURCE_FILES
 ${PREFIX}mypy $SOURCE_FILES
-${PREFIX}isort --check --diff --project=httpx $SOURCE_FILES
+${PREFIX}isort --check --diff --project=broadcaster $SOURCE_FILES
diff --git a/scripts/lint b/scripts/lint
index 81851c6..db86237 100755
--- a/scripts/lint
+++ b/scripts/lint
@@ -1,13 +1,13 @@
 #!/bin/sh -e
 
 export PREFIX=""
-if [ -d 'venv' ] ; then
+if [ -d 'venv' ]; then
     export PREFIX="venv/bin/"
 fi
-export SOURCE_FILES="httpx tests"
+export SOURCE_FILES="broadcaster tests"
 
 set -x
 
 ${PREFIX}autoflake --in-place --recursive $SOURCE_FILES
-${PREFIX}isort --project=httpx $SOURCE_FILES
+${PREFIX}isort --project=broadcaster $SOURCE_FILES
 ${PREFIX}black --target-version=py37 $SOURCE_FILES
diff --git a/setup.cfg b/setup.cfg
index c860d81..37b5d85 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -21,4 +21,4 @@ markers =
 
 [coverage:run]
 omit = venv/*
-include = httpx/*, tests/*
+include = broadcaster/*, tests/*

From 25b364d039976fe70993c8bbf562d982ca49373b Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Sat, 20 Aug 2022 08:10:25 +0200
Subject: [PATCH 04/33] Pin dependencies (#74)

---
 requirements.txt        | 33 +++++++++++++++------------------
 scripts/coverage        |  2 +-
 tests/test_broadcast.py |  1 +
 3 files changed, 17 insertions(+), 19 deletions(-)

diff --git a/requirements.txt b/requirements.txt
index 811910c..02846f0 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,25 +1,22 @@
 -e .[redis,postgres,kafka]
 
 # Documentation
-mkdocs
-mkautodoc
-mkdocs-material
+mkdocs==1.3.1
+mkautodoc==0.2.0
+mkdocs-material==8.4.0
 
 # Packaging
-twine
-wheel
+twine==4.0.1
+wheel==0.37.1
 
 # Tests & Linting
-autoflake
-black==20.8b1
-coverage==5.3
-flake8
-flake8-bugbear
-flake8-pie==0.5.*
-isort==5.*
-mypy
-pytest==5.*
-pytest-asyncio
-pytest-trio
-trio
-trio-typing
+autoflake==1.4
+black==22.6.0
+coverage==6.4.4
+flake8==3.9.2
+flake8-bugbear==22.7.1
+flake8-pie==0.16.0
+isort==5.10.1
+mypy==0.971
+pytest==7.1.2
+pytest-asyncio==0.19.0
diff --git a/scripts/coverage b/scripts/coverage
index 73a2198..ccab2e3 100755
--- a/scripts/coverage
+++ b/scripts/coverage
@@ -8,4 +8,4 @@ export SOURCE_FILES="broadcaster tests"
 
 set -x
 
-${PREFIX}coverage report --show-missing --skip-covered --fail-under=100
+${PREFIX}coverage report --show-missing --skip-covered --fail-under=88
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index 61e7295..e3313bc 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -35,6 +35,7 @@ async def test_postgres():
             assert event.message == "hello"
 
 
+@pytest.mark.skip("Deadlock on `next_published`")
 @pytest.mark.asyncio
 async def test_kafka():
     async with Broadcast("kafka://localhost:9092") as broadcast:

From e9fdd029816baac37c47cdeef43f7c60c5fc6c25 Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Sat, 20 Aug 2022 08:38:18 +0200
Subject: [PATCH 05/33] Comply with PEP 561 (#75)

Co-authored-by: Diogo Dutra <9977774+dutradda@users.noreply.github.com>
---
 broadcaster/py.typed | 0
 setup.py             | 9 +++------
 2 files changed, 3 insertions(+), 6 deletions(-)
 create mode 100644 broadcaster/py.typed

diff --git a/broadcaster/py.typed b/broadcaster/py.typed
new file mode 100644
index 0000000..e69de29
diff --git a/setup.py b/setup.py
index 668b715..dcde310 100644
--- a/setup.py
+++ b/setup.py
@@ -1,6 +1,3 @@
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
 import os
 import re
 
@@ -46,12 +43,13 @@ def get_packages(package):
     author="Tom Christie",
     author_email="tom@tomchristie.com",
     packages=get_packages("broadcaster"),
-    data_files=[("", ["LICENSE.md"])],
+    package_data={"broadcaster": ["py.typed"]},
+    include_package_data=True,
     extras_require={
         "redis": ["asyncio-redis"],
         "postgres": ["asyncpg"],
         "kafka": ["aiokafka"],
-        "test": ["pytest", "pytest-asyncio"]
+        "test": ["pytest", "pytest-asyncio"],
     },
     classifiers=[
         "Development Status :: 3 - Alpha",
@@ -64,5 +62,4 @@ def get_packages(package):
         "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
     ],
-    # zip_safe=False,
 )

From 956571d030d33d6cb820758ec5ed8fe79c3288c6 Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Sat, 20 Aug 2022 08:42:54 +0200
Subject: [PATCH 06/33] Remove deprecated loop parameter from AIOKafka
 constructors (#76)

---
 broadcaster/_backends/kafka.py | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/broadcaster/_backends/kafka.py b/broadcaster/_backends/kafka.py
index a244763..18b88d2 100644
--- a/broadcaster/_backends/kafka.py
+++ b/broadcaster/_backends/kafka.py
@@ -1,4 +1,3 @@
-import asyncio
 import typing
 from urllib.parse import urlparse
 
@@ -14,9 +13,8 @@ def __init__(self, url: str):
         self._consumer_channels: typing.Set = set()
 
     async def connect(self) -> None:
-        loop = asyncio.get_event_loop()
-        self._producer = AIOKafkaProducer(loop=loop, bootstrap_servers=self._servers)
-        self._consumer = AIOKafkaConsumer(loop=loop, bootstrap_servers=self._servers)
+        self._producer = AIOKafkaProducer(bootstrap_servers=self._servers)
+        self._consumer = AIOKafkaConsumer(bootstrap_servers=self._servers)
         await self._producer.start()
         await self._consumer.start()
 

From ed15621858f88c2597c3d8ad06d3c2e9b2b55dd9 Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Sat, 20 Aug 2022 08:44:32 +0200
Subject: [PATCH 07/33] Add GitHub funding (#77)

---
 .github/FUNDING.yml | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 .github/FUNDING.yml

diff --git a/.github/FUNDING.yml b/.github/FUNDING.yml
new file mode 100644
index 0000000..2f87d94
--- /dev/null
+++ b/.github/FUNDING.yml
@@ -0,0 +1 @@
+github: encode

From f11d94d85caa98ae9fb3cfd7aaf75fdcf3ee7468 Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Sat, 20 Aug 2022 08:52:00 +0200
Subject: [PATCH 08/33] Support Python 3.10 (#78)

---
 .github/workflows/test-suite.yml | 4 ++--
 setup.py                         | 3 +++
 2 files changed, 5 insertions(+), 2 deletions(-)

diff --git a/.github/workflows/test-suite.yml b/.github/workflows/test-suite.yml
index 458bcad..e256667 100644
--- a/.github/workflows/test-suite.yml
+++ b/.github/workflows/test-suite.yml
@@ -14,7 +14,7 @@ jobs:
 
     strategy:
       matrix:
-        python-version: ["3.7", "3.8", "3.9"]
+        python-version: ["3.7", "3.8", "3.9", "3.10", "3.11-dev"]
 
     services:
       zookeeper:
@@ -66,4 +66,4 @@ jobs:
       - name: "Run tests"
         run: "scripts/test"
       - name: "Enforce coverage"
-        run: "scripts/coverage"
\ No newline at end of file
+        run: "scripts/coverage"
diff --git a/setup.py b/setup.py
index dcde310..9cdf906 100644
--- a/setup.py
+++ b/setup.py
@@ -61,5 +61,8 @@ def get_packages(package):
         "Programming Language :: Python :: 3",
         "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
+        "Programming Language :: Python :: 3.9",
+        "Programming Language :: Python :: 3.10",
+        "Programming Language :: Python :: 3.11",
     ],
 )

From e1f297796ae1b7f16522b013454cae546fc91fcd Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Sat, 20 Aug 2022 13:37:36 +0200
Subject: [PATCH 09/33] Add `password` support for Redis backend (#79)

Co-authored-by: k1dave6412 <14539608+k1dave6412@users.noreply.github.com>
---
 broadcaster/_backends/redis.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/broadcaster/_backends/redis.py b/broadcaster/_backends/redis.py
index b545bad..2c4aeba 100644
--- a/broadcaster/_backends/redis.py
+++ b/broadcaster/_backends/redis.py
@@ -12,10 +12,12 @@ def __init__(self, url: str):
         parsed_url = urlparse(url)
         self._host = parsed_url.hostname or "localhost"
         self._port = parsed_url.port or 6379
+        self._password = parsed_url.password or None
 
     async def connect(self) -> None:
-        self._pub_conn = await asyncio_redis.Connection.create(self._host, self._port)
-        self._sub_conn = await asyncio_redis.Connection.create(self._host, self._port)
+        kwargs = {"host": self._host, "port": self._port, "password": self._password}
+        self._pub_conn = await asyncio_redis.Connection.create(**kwargs)
+        self._sub_conn = await asyncio_redis.Connection.create(**kwargs)
         self._subscriber = await self._sub_conn.start_subscribe()
 
     async def disconnect(self) -> None:

From 312fde9b3e7f2228bf14952dc3aad67b3c2afea2 Mon Sep 17 00:00:00 2001
From: Pravesh Chapagain <spravesh1818@live.com>
Date: Sat, 20 Aug 2022 17:30:13 +0545
Subject: [PATCH 10/33] Support `rediss://` scheme for Redis backend (#52)

* Multiple url schemes for redis

* Update broadcaster/_base.py

Co-authored-by: Pravesh Chapagain <spravesh1818@gmail.com>
Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>
---
 broadcaster/_base.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/broadcaster/_base.py b/broadcaster/_base.py
index c58cb1d..4de1417 100644
--- a/broadcaster/_base.py
+++ b/broadcaster/_base.py
@@ -31,7 +31,7 @@ def __init__(self, url: str):
         parsed_url = urlparse(url)
         self._backend: BroadcastBackend
         self._subscribers: Dict[str, Any] = {}
-        if parsed_url.scheme == "redis":
+        if parsed_url.scheme in ("redis", "rediss"):
             from broadcaster._backends.redis import RedisBackend
 
             self._backend = RedisBackend(url)

From 4d993ad31cc34975fa9b28763415678dfd559d20 Mon Sep 17 00:00:00 2001
From: Gabriel Abud <Buuntu@users.noreply.github.com>
Date: Sat, 20 Aug 2022 04:51:24 -0700
Subject: [PATCH 11/33] Add `websockets` requirement to example (#43)

---
 example/requirements.txt | 1 +
 1 file changed, 1 insertion(+)

diff --git a/example/requirements.txt b/example/requirements.txt
index 2b7b4ad..44835e5 100644
--- a/example/requirements.txt
+++ b/example/requirements.txt
@@ -1,4 +1,5 @@
 uvicorn
+websockets
 starlette
 jinja2
 broadcaster[redis,postgres,kafka]

From ccd476e1c03aa54541f966183f839278c34b0d3d Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Sat, 20 Aug 2022 14:14:59 +0200
Subject: [PATCH 12/33] Add `publish` GitHub workflow (#80)

---
 .github/workflows/publish.yml | 29 +++++++++++++++++++++++++++++
 1 file changed, 29 insertions(+)
 create mode 100644 .github/workflows/publish.yml

diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml
new file mode 100644
index 0000000..3ea8eef
--- /dev/null
+++ b/.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 }}

From 9aa890c61cf500d7a50f1607822ee70de07e9a91 Mon Sep 17 00:00:00 2001
From: anabasalo <58712297+anabasalo@users.noreply.github.com>
Date: Sun, 18 Jun 2023 19:05:47 -0300
Subject: [PATCH 13/33] Drop Python 3.7 support (#95)

---
 .github/workflows/test-suite.yml | 2 +-
 README.md                        | 2 +-
 setup.py                         | 3 +--
 3 files changed, 3 insertions(+), 4 deletions(-)

diff --git a/.github/workflows/test-suite.yml b/.github/workflows/test-suite.yml
index e256667..900834d 100644
--- a/.github/workflows/test-suite.yml
+++ b/.github/workflows/test-suite.yml
@@ -14,7 +14,7 @@ jobs:
 
     strategy:
       matrix:
-        python-version: ["3.7", "3.8", "3.9", "3.10", "3.11-dev"]
+        python-version: ["3.8", "3.9", "3.10", "3.11-dev"]
 
     services:
       zookeeper:
diff --git a/README.md b/README.md
index 61709ac..f2226e3 100644
--- a/README.md
+++ b/README.md
@@ -65,7 +65,7 @@ The HTML template for the front end [is available here](https://github.com/encod
 
 ## Requirements
 
-Python 3.7+
+Python 3.8+
 
 ## Installation
 
diff --git a/setup.py b/setup.py
index 9cdf906..ccafacc 100644
--- a/setup.py
+++ b/setup.py
@@ -33,7 +33,7 @@ def get_packages(package):
 
 setup(
     name="broadcaster",
-    python_requires=">=3.7",
+    python_requires=">=3.8",
     version=get_version("broadcaster"),
     url="https://github.com/encode/broadcaster",
     license="BSD",
@@ -59,7 +59,6 @@ def get_packages(package):
         "Operating System :: OS Independent",
         "Topic :: Internet :: WWW/HTTP",
         "Programming Language :: Python :: 3",
-        "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
         "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",

From 2499fbd28397b065894706a09d24d0d636b7858a Mon Sep 17 00:00:00 2001
From: Sumit Kashyap <mr.sumitkrr@gmail.com>
Date: Mon, 19 Jun 2023 22:51:26 +0530
Subject: [PATCH 14/33] Use `pyproject.toml` with hatch instead of `setup.py`
 (#96)

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>
---
 pyproject.toml   | 48 ++++++++++++++++++++++++++++++++++
 requirements.txt |  2 +-
 scripts/build    |  2 +-
 setup.py         | 67 ------------------------------------------------
 4 files changed, 50 insertions(+), 69 deletions(-)
 create mode 100644 pyproject.toml
 delete mode 100644 setup.py

diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..5447356
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,48 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "broadcaster"
+dynamic = ["version"]
+description = "Simple broadcast channels."
+readme = "README.md"
+license = "BSD-3-Clause"
+requires-python = ">=3.8"
+authors = [
+    { name = "Tom Christie", email = "tom@tomchristie.com" },
+]
+classifiers = [
+    "Development Status :: 3 - Alpha",
+    "Environment :: Web Environment",
+    "Intended Audience :: Developers",
+    "License :: OSI Approved :: BSD License",
+    "Operating System :: OS Independent",
+    "Topic :: Internet :: WWW/HTTP",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3.8",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+]
+dependencies = [
+    "anyio>=3.4.0,<5",
+    "typing_extensions>=3.10.0; python_version < '3.10'",
+]
+
+[project.optional-dependencies]
+redis = ["asyncio-redis"]
+postgres = ["asyncpg"]
+kafka = ["aiokafka"]
+test =  ["pytest", "pytest-asyncio"]
+
+[project.urls]
+Homepage = "https://github.com/encode/broadcaster"
+
+[tool.hatch.version]
+path = "broadcaster/__init__.py"
+
+[tool.hatch.build.targets.sdist]
+include = [
+    "/broadcaster",
+]
diff --git a/requirements.txt b/requirements.txt
index 02846f0..688c4c0 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -6,8 +6,8 @@ mkautodoc==0.2.0
 mkdocs-material==8.4.0
 
 # Packaging
+build==0.10.0
 twine==4.0.1
-wheel==0.37.1
 
 # Tests & Linting
 autoflake==1.4
diff --git a/scripts/build b/scripts/build
index 7d327b5..5d5e71a 100755
--- a/scripts/build
+++ b/scripts/build
@@ -8,6 +8,6 @@ fi
 
 set -x
 
-${PREFIX}python setup.py sdist bdist_wheel
+${PREFIX}python -m build
 ${PREFIX}twine check dist/*
 # ${PREFIX}mkdocs build
diff --git a/setup.py b/setup.py
deleted file mode 100644
index ccafacc..0000000
--- a/setup.py
+++ /dev/null
@@ -1,67 +0,0 @@
-import os
-import re
-
-from setuptools import setup
-
-
-def get_version(package):
-    """
-    Return package version as listed in `__version__` in `init.py`.
-    """
-    with open(os.path.join(package, "__init__.py")) as f:
-        return re.search("__version__ = ['\"]([^'\"]+)['\"]", f.read()).group(1)
-
-
-def get_long_description():
-    """
-    Return the README.
-    """
-    with open("README.md", encoding="utf8") as f:
-        return f.read()
-
-
-def get_packages(package):
-    """
-    Return root package and all sub-packages.
-    """
-    return [
-        dirpath
-        for dirpath, dirnames, filenames in os.walk(package)
-        if os.path.exists(os.path.join(dirpath, "__init__.py"))
-    ]
-
-
-setup(
-    name="broadcaster",
-    python_requires=">=3.8",
-    version=get_version("broadcaster"),
-    url="https://github.com/encode/broadcaster",
-    license="BSD",
-    description="Simple broadcast channels.",
-    long_description=get_long_description(),
-    long_description_content_type="text/markdown",
-    author="Tom Christie",
-    author_email="tom@tomchristie.com",
-    packages=get_packages("broadcaster"),
-    package_data={"broadcaster": ["py.typed"]},
-    include_package_data=True,
-    extras_require={
-        "redis": ["asyncio-redis"],
-        "postgres": ["asyncpg"],
-        "kafka": ["aiokafka"],
-        "test": ["pytest", "pytest-asyncio"],
-    },
-    classifiers=[
-        "Development Status :: 3 - Alpha",
-        "Environment :: Web Environment",
-        "Intended Audience :: Developers",
-        "License :: OSI Approved :: BSD License",
-        "Operating System :: OS Independent",
-        "Topic :: Internet :: WWW/HTTP",
-        "Programming Language :: Python :: 3",
-        "Programming Language :: Python :: 3.8",
-        "Programming Language :: Python :: 3.9",
-        "Programming Language :: Python :: 3.10",
-        "Programming Language :: Python :: 3.11",
-    ],
-)

From 94bf88721df859fe6e4b5dbaa4608c06fff24412 Mon Sep 17 00:00:00 2001
From: immanelg <dmitryborovicky@gmail.com>
Date: Tue, 27 Jun 2023 14:49:23 +0000
Subject: [PATCH 15/33] Replace `run_until_first_complete` with task group
 (#101)

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>
---
 README.md      | 15 ++++++++++-----
 example/app.py | 16 ++++++++++------
 2 files changed, 20 insertions(+), 11 deletions(-)

diff --git a/README.md b/README.md
index f2226e3..f3b7d93 100644
--- a/README.md
+++ b/README.md
@@ -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):
diff --git a/example/app.py b/example/app.py
index 9b57e44..ca61d03 100644
--- a/example/app.py
+++ b/example/app.py
@@ -1,8 +1,7 @@
 import os
 
-import uvicorn
+import anyio
 from starlette.applications import Starlette
-from starlette.concurrency import run_until_first_complete
 from starlette.routing import Route, WebSocketRoute
 from starlette.templating import Jinja2Templates
 
@@ -22,10 +21,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):

From 377b4043814ac88a21255af1cc12d5650c785082 Mon Sep 17 00:00:00 2001
From: Sumit Kashyap <mr.sumitkrr@gmail.com>
Date: Wed, 5 Jul 2023 18:33:25 +0530
Subject: [PATCH 16/33] Move `setup.cfg` to `pyproject.toml` (#98)

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>
---
 example/app.py   |  1 +
 pyproject.toml   | 29 +++++++++++++++++++++++++++++
 requirements.txt |  8 ++------
 scripts/check    |  3 +--
 scripts/coverage |  3 +--
 scripts/lint     |  3 +--
 setup.cfg        | 24 ------------------------
 7 files changed, 35 insertions(+), 36 deletions(-)
 delete mode 100644 setup.cfg

diff --git a/example/app.py b/example/app.py
index ca61d03..a201221 100644
--- a/example/app.py
+++ b/example/app.py
@@ -1,5 +1,6 @@
 import os
 
+
 import anyio
 from starlette.applications import Starlette
 from starlette.routing import Route, WebSocketRoute
diff --git a/pyproject.toml b/pyproject.toml
index 5447356..ef59da4 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -46,3 +46,32 @@ path = "broadcaster/__init__.py"
 include = [
     "/broadcaster",
 ]
+
+[tool.ruff]
+ignore = []
+line-length = 120
+select = ["E","F","W"]
+
+[tool.ruff.isort]
+combine-as-imports = true
+
+[tool.mypy]
+disallow_untyped_defs = true
+ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+module = "tests.*"
+disallow_untyped_defs = false
+check_untyped_defs = true
+
+[tool.pytest.ini_options]
+addopts = "-rxXs"
+markers = "copied_from(source, changes=None): mark test as copied from somewhere else, along with a description of changes made to accodomate e.g. our test setup"
+
+[tool.coverage.run]
+source_pkgs = ["broadcaster", "tests"]
+
+[tool.coverage.report]
+fail_under = 78
+show_missing = true
+skip_covered = true
diff --git a/requirements.txt b/requirements.txt
index 688c4c0..dcf0fcf 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -10,13 +10,9 @@ build==0.10.0
 twine==4.0.1
 
 # Tests & Linting
-autoflake==1.4
+ruff==0.0.277
 black==22.6.0
 coverage==6.4.4
-flake8==3.9.2
-flake8-bugbear==22.7.1
-flake8-pie==0.16.0
-isort==5.10.1
 mypy==0.971
 pytest==7.1.2
-pytest-asyncio==0.19.0
+pytest-asyncio==0.19.0
\ No newline at end of file
diff --git a/scripts/check b/scripts/check
index 5ee0190..e514548 100755
--- a/scripts/check
+++ b/scripts/check
@@ -9,6 +9,5 @@ export SOURCE_FILES="broadcaster tests"
 set -x
 
 ${PREFIX}black --check --diff --target-version=py37 $SOURCE_FILES
-${PREFIX}flake8 $SOURCE_FILES
+${PREFIX}ruff check $SOURCE_FILES
 ${PREFIX}mypy $SOURCE_FILES
-${PREFIX}isort --check --diff --project=broadcaster $SOURCE_FILES
diff --git a/scripts/coverage b/scripts/coverage
index ccab2e3..0015d39 100755
--- a/scripts/coverage
+++ b/scripts/coverage
@@ -4,8 +4,7 @@ export PREFIX=""
 if [ -d 'venv' ] ; then
     export PREFIX="venv/bin/"
 fi
-export SOURCE_FILES="broadcaster tests"
 
 set -x
 
-${PREFIX}coverage report --show-missing --skip-covered --fail-under=88
+${PREFIX}coverage report
diff --git a/scripts/lint b/scripts/lint
index db86237..4751285 100755
--- a/scripts/lint
+++ b/scripts/lint
@@ -8,6 +8,5 @@ export SOURCE_FILES="broadcaster tests"
 
 set -x
 
-${PREFIX}autoflake --in-place --recursive $SOURCE_FILES
-${PREFIX}isort --project=broadcaster $SOURCE_FILES
+${PREFIX}ruff --fix $SOURCE_FILES
 ${PREFIX}black --target-version=py37 $SOURCE_FILES
diff --git a/setup.cfg b/setup.cfg
deleted file mode 100644
index 37b5d85..0000000
--- a/setup.cfg
+++ /dev/null
@@ -1,24 +0,0 @@
-[flake8]
-ignore = W503, E203, B305
-max-line-length = 120
-
-[mypy]
-disallow_untyped_defs = True
-ignore_missing_imports = True
-
-[mypy-tests.*]
-disallow_untyped_defs = False
-check_untyped_defs = True
-
-[tool:isort]
-profile = black
-combine_as_imports = True
-
-[tool:pytest]
-addopts = -rxXs
-markers =
-  copied_from(source, changes=None): mark test as copied from somewhere else, along with a description of changes made to accodomate e.g. our test setup
-
-[coverage:run]
-omit = venv/*
-include = broadcaster/*, tests/*

From de6ef400220fc8942e4e22fec77532bb387ce77d Mon Sep 17 00:00:00 2001
From: Alex Oleshkevich <alex.oleshkevich@gmail.com>
Date: Thu, 28 Mar 2024 08:54:55 +0100
Subject: [PATCH 17/33] Switch to `redis-py` library (#111)

* use redis-py library

* update dependencies
---
 broadcaster/_backends/redis.py | 45 +++++++++++++++++++++-------------
 pyproject.toml                 |  2 +-
 2 files changed, 29 insertions(+), 18 deletions(-)

diff --git a/broadcaster/_backends/redis.py b/broadcaster/_backends/redis.py
index 2c4aeba..78ed7eb 100644
--- a/broadcaster/_backends/redis.py
+++ b/broadcaster/_backends/redis.py
@@ -1,7 +1,7 @@
+import asyncio
 import typing
-from urllib.parse import urlparse
 
-import asyncio_redis
+from redis import asyncio as redis
 
 from .._base import Event
 from .base import BroadcastBackend
@@ -9,30 +9,41 @@
 
 class RedisBackend(BroadcastBackend):
     def __init__(self, url: str):
-        parsed_url = urlparse(url)
-        self._host = parsed_url.hostname or "localhost"
-        self._port = parsed_url.port or 6379
-        self._password = parsed_url.password or None
+        self._conn = redis.Redis.from_url(url)
+        self._pubsub = self._conn.pubsub()
+        self._ready = asyncio.Event()
+        self._queue: asyncio.Queue[Event] = asyncio.Queue()
+        self._listener = asyncio.create_task(self._pubsub_listener())
 
     async def connect(self) -> None:
-        kwargs = {"host": self._host, "port": self._port, "password": self._password}
-        self._pub_conn = await asyncio_redis.Connection.create(**kwargs)
-        self._sub_conn = await asyncio_redis.Connection.create(**kwargs)
-        self._subscriber = await self._sub_conn.start_subscribe()
+        await self._pubsub.connect()
 
     async def disconnect(self) -> None:
-        self._pub_conn.close()
-        self._sub_conn.close()
+        await self._pubsub.aclose()
+        await self._conn.aclose()
+        self._listener.cancel()
 
     async def subscribe(self, channel: str) -> None:
-        await self._subscriber.subscribe([channel])
+        self._ready.set()
+        await self._pubsub.subscribe(channel)
 
     async def unsubscribe(self, channel: str) -> None:
-        await self._subscriber.unsubscribe([channel])
+        await self._pubsub.unsubscribe(channel)
 
     async def publish(self, channel: str, message: typing.Any) -> None:
-        await self._pub_conn.publish(channel, message)
+        await self._conn.publish(channel, message)
 
     async def next_published(self) -> Event:
-        message = await self._subscriber.next_published()
-        return Event(channel=message.channel, message=message.value)
+        return await self._queue.get()
+
+    async def _pubsub_listener(self) -> None:
+        # redis-py does not listen to the pubsub connection if there are no channels subscribed
+        # so we need to wait until the first channel is subscribed to start listening
+        await self._ready.wait()
+        async for message in self._pubsub.listen():
+            if message["type"] == "message":
+                event = Event(
+                    channel=message["channel"].decode(),
+                    message=message["data"].decode(),
+                )
+                await self._queue.put(event)
diff --git a/pyproject.toml b/pyproject.toml
index ef59da4..10f7cc5 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -31,7 +31,7 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
-redis = ["asyncio-redis"]
+redis = ["redis"]
 postgres = ["asyncpg"]
 kafka = ["aiokafka"]
 test =  ["pytest", "pytest-asyncio"]

From eeba6a796d5ab523d21234f5870cef384c156553 Mon Sep 17 00:00:00 2001
From: jolorke <165886222+jolorke@users.noreply.github.com>
Date: Wed, 3 Apr 2024 21:22:40 +0200
Subject: [PATCH 18/33] Fix subscriber not properly unsubscribing when an
 exception is raised inside the context manager (#112)

* add testcases for unsubsribe

* fix the context manager of the function "subscribe" not removing the associated queue from the channel in case of a raised exception inside the context manager (e.g. being used inside a generator that gets closed raising GeneratorExit)

* activate broadcaster via context manager

---------

Co-authored-by: alex.oleshkevich <alex.oleshkevich@gmail.com>
---
 broadcaster/_base.py      |  3 +--
 tests/test_unsubscribe.py | 25 +++++++++++++++++++++++++
 2 files changed, 26 insertions(+), 2 deletions(-)
 create mode 100644 tests/test_unsubscribe.py

diff --git a/broadcaster/_base.py b/broadcaster/_base.py
index 4de1417..c8dc221 100644
--- a/broadcaster/_base.py
+++ b/broadcaster/_base.py
@@ -90,12 +90,11 @@ async def subscribe(self, channel: str) -> AsyncIterator["Subscriber"]:
                 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)
 
 
diff --git a/tests/test_unsubscribe.py b/tests/test_unsubscribe.py
new file mode 100644
index 0000000..30f928b
--- /dev/null
+++ b/tests/test_unsubscribe.py
@@ -0,0 +1,25 @@
+import pytest
+from broadcaster import Broadcast
+
+
+@pytest.mark.asyncio
+async def test_unsubscribe():
+    """The queue should be removed when the context manager is left."""
+    async with Broadcast("memory://") as broadcast:
+        async with broadcast.subscribe("chatroom"):
+            pass
+
+        assert "chatroom" not in broadcast._subscribers
+
+
+@pytest.mark.asyncio
+async def test_unsubscribe_w_exception():
+    """In case an exception is raised inside the context manager, the queue should be removed."""
+    async with Broadcast("memory://") as broadcast:
+        try:
+            async with broadcast.subscribe("chatroom"):
+                raise RuntimeError("MyException")
+        except RuntimeError:
+            pass
+
+        assert "chatroom" not in broadcast._subscribers

From eeb09ded25ed47485b7aa2cd69d53bd6c625c88f Mon Sep 17 00:00:00 2001
From: Alex Oleshkevich <alex.oleshkevich@gmail.com>
Date: Thu, 4 Apr 2024 11:07:00 +0200
Subject: [PATCH 19/33] Allow user backends (#110)

* allow user backends

* update docs

* Update README.md

Co-authored-by: Tom Christie <tom@tomchristie.com>

* use custom backend in tests

---------

Co-authored-by: Tom Christie <tom@tomchristie.com>
---
 README.md               | 15 +++++++++++
 broadcaster/__init__.py |  3 ++-
 broadcaster/_base.py    | 33 +++++++++++++++++-------
 tests/test_broadcast.py | 57 ++++++++++++++++++++++++++++++++++++++++-
 4 files changed, 97 insertions(+), 11 deletions(-)

diff --git a/README.md b/README.md
index f3b7d93..d11a9ed 100644
--- a/README.md
+++ b/README.md
@@ -86,6 +86,21 @@ Python 3.8+
 * `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.
diff --git a/broadcaster/__init__.py b/broadcaster/__init__.py
index edc56d6..b5dd0bf 100644
--- a/broadcaster/__init__.py
+++ b/broadcaster/__init__.py
@@ -1,4 +1,5 @@
 from ._base import Broadcast, Event
+from ._backends.base import BroadcastBackend
 
 __version__ = "0.2.0"
-__all__ = ["Broadcast", "Event"]
+__all__ = ["Broadcast", "Event", "BroadcastBackend"]
diff --git a/broadcaster/_base.py b/broadcaster/_base.py
index c8dc221..997b82a 100644
--- a/broadcaster/_base.py
+++ b/broadcaster/_base.py
@@ -1,8 +1,19 @@
 import asyncio
 from contextlib import asynccontextmanager
-from typing import Any, AsyncGenerator, AsyncIterator, Dict, Optional
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    AsyncGenerator,
+    AsyncIterator,
+    Dict,
+    Optional,
+    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:
@@ -25,31 +36,35 @@ class Unsubscribed(Exception):
 
 
 class Broadcast:
-    def __init__(self, url: str):
-        from broadcaster._backends.base import BroadcastBackend
+    def __init__(
+        self, url: Optional[str] = None, *, backend: Optional["BroadcastBackend"] = 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, Any] = {}
 
+    def _create_backend(self, url: str) -> "BroadcastBackend":
         parsed_url = urlparse(url)
-        self._backend: BroadcastBackend
-        self._subscribers: Dict[str, Any] = {}
         if parsed_url.scheme in ("redis", "rediss"):
             from broadcaster._backends.redis import RedisBackend
 
-            self._backend = RedisBackend(url)
+            return RedisBackend(url)
 
         elif parsed_url.scheme in ("postgres", "postgresql"):
             from broadcaster._backends.postgres import PostgresBackend
 
-            self._backend = PostgresBackend(url)
+            return PostgresBackend(url)
 
         if parsed_url.scheme == "kafka":
             from broadcaster._backends.kafka import KafkaBackend
 
-            self._backend = KafkaBackend(url)
+            return KafkaBackend(url)
 
         elif parsed_url.scheme == "memory":
             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":
         await self.connect()
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index e3313bc..4cf9e45 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -1,6 +1,35 @@
 import pytest
+import typing
+import asyncio
 
-from broadcaster import Broadcast
+from broadcaster import Broadcast, BroadcastBackend, Event
+
+
+class CustomBackend(BroadcastBackend):
+    def __init__(self, url: str):
+        self._subscribed: typing.Set = set()
+
+    async def connect(self) -> None:
+        self._published: asyncio.Queue = asyncio.Queue()
+
+    async def disconnect(self) -> None:
+        pass
+
+    async def subscribe(self, channel: str) -> None:
+        self._subscribed.add(channel)
+
+    async def unsubscribe(self, channel: str) -> None:
+        self._subscribed.remove(channel)
+
+    async def publish(self, channel: str, message: typing.Any) -> None:
+        event = Event(channel=channel, message=message)
+        await self._published.put(event)
+
+    async def next_published(self) -> Event:
+        while True:
+            event = await self._published.get()
+            if event.channel in self._subscribed:
+                return event
 
 
 @pytest.mark.asyncio
@@ -44,3 +73,29 @@ async def test_kafka():
             event = await subscriber.get()
             assert event.channel == "chatroom"
             assert event.message == "hello"
+
+
+@pytest.mark.asyncio
+async def test_custom():
+    backend = CustomBackend("")
+    async with Broadcast(backend=backend) as broadcast:
+        async with broadcast.subscribe("chatroom") as subscriber:
+            await broadcast.publish("chatroom", "hello")
+            event = await subscriber.get()
+            assert event.channel == "chatroom"
+            assert event.message == "hello"
+
+
+@pytest.mark.asyncio
+async def test_unknown_backend():
+    with pytest.raises(ValueError, match="Unsupported backend"):
+        async with Broadcast(url="unknown://"):
+            pass
+
+
+@pytest.mark.asyncio
+async def test_needs_url_or_backend():
+    with pytest.raises(
+        AssertionError, match="Either `url` or `backend` must be provided."
+    ):
+        Broadcast()

From 6dc07d6723cef7379ab057847387712afac05a0e Mon Sep 17 00:00:00 2001
From: Alex Oleshkevich <alex.oleshkevich@gmail.com>
Date: Fri, 5 Apr 2024 09:32:09 +0200
Subject: [PATCH 20/33] General project maintenance. (#114)

* update linting tools, sync lint configs with starlette, improve typings

* disable 3.13-dev as asyncpg cannot be build yet

* remove .python-version

* Update scripts/lint

Co-authored-by: Zanie Blue <contact@zanie.dev>

* update ruff

---------

Co-authored-by: Zanie Blue <contact@zanie.dev>
---
 .github/workflows/test-suite.yml  |  2 +-
 broadcaster/__init__.py           |  2 +-
 broadcaster/_backends/kafka.py    |  4 +++-
 broadcaster/_backends/memory.py   |  6 ++++--
 broadcaster/_backends/postgres.py |  2 +-
 broadcaster/_base.py              | 36 +++++++++++--------------------
 pyproject.toml                    | 10 ++++++---
 requirements.txt                  | 19 ++++++++--------
 scripts/check                     |  4 ++--
 scripts/lint                      |  6 +++---
 tests/test_broadcast.py           | 19 ++++++++--------
 tests/test_unsubscribe.py         |  1 +
 12 files changed, 53 insertions(+), 58 deletions(-)

diff --git a/.github/workflows/test-suite.yml b/.github/workflows/test-suite.yml
index 900834d..81db761 100644
--- a/.github/workflows/test-suite.yml
+++ b/.github/workflows/test-suite.yml
@@ -14,7 +14,7 @@ jobs:
 
     strategy:
       matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11-dev"]
+        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
 
     services:
       zookeeper:
diff --git a/broadcaster/__init__.py b/broadcaster/__init__.py
index b5dd0bf..a41bbeb 100644
--- a/broadcaster/__init__.py
+++ b/broadcaster/__init__.py
@@ -1,5 +1,5 @@
-from ._base import Broadcast, Event
 from ._backends.base import BroadcastBackend
+from ._base import Broadcast, Event
 
 __version__ = "0.2.0"
 __all__ = ["Broadcast", "Event", "BroadcastBackend"]
diff --git a/broadcaster/_backends/kafka.py b/broadcaster/_backends/kafka.py
index 18b88d2..e577769 100644
--- a/broadcaster/_backends/kafka.py
+++ b/broadcaster/_backends/kafka.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import typing
 from urllib.parse import urlparse
 
@@ -10,7 +12,7 @@
 class KafkaBackend(BroadcastBackend):
     def __init__(self, url: str):
         self._servers = [urlparse(url).netloc]
-        self._consumer_channels: typing.Set = set()
+        self._consumer_channels: set[str] = set()
 
     async def connect(self) -> None:
         self._producer = AIOKafkaProducer(bootstrap_servers=self._servers)
diff --git a/broadcaster/_backends/memory.py b/broadcaster/_backends/memory.py
index 5a9fa53..bfd0c44 100644
--- a/broadcaster/_backends/memory.py
+++ b/broadcaster/_backends/memory.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import asyncio
 import typing
 
@@ -7,10 +9,10 @@
 
 class MemoryBackend(BroadcastBackend):
     def __init__(self, url: str):
-        self._subscribed: typing.Set = set()
+        self._subscribed: set[str] = set()
 
     async def connect(self) -> None:
-        self._published: asyncio.Queue = asyncio.Queue()
+        self._published: asyncio.Queue[Event] = asyncio.Queue()
 
     async def disconnect(self) -> None:
         pass
diff --git a/broadcaster/_backends/postgres.py b/broadcaster/_backends/postgres.py
index 47ef4f6..7769962 100644
--- a/broadcaster/_backends/postgres.py
+++ b/broadcaster/_backends/postgres.py
@@ -13,7 +13,7 @@ def __init__(self, url: str):
 
     async def connect(self) -> None:
         self._conn = await asyncpg.connect(self._url)
-        self._listen_queue: asyncio.Queue = asyncio.Queue()
+        self._listen_queue: asyncio.Queue[Event] = asyncio.Queue()
 
     async def disconnect(self) -> None:
         await self._conn.close()
diff --git a/broadcaster/_base.py b/broadcaster/_base.py
index 997b82a..0166034 100644
--- a/broadcaster/_base.py
+++ b/broadcaster/_base.py
@@ -1,14 +1,8 @@
+from __future__ import annotations
+
 import asyncio
 from contextlib import asynccontextmanager
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    AsyncGenerator,
-    AsyncIterator,
-    Dict,
-    Optional,
-    cast,
-)
+from typing import TYPE_CHECKING, Any, AsyncGenerator, AsyncIterator, cast
 from urllib.parse import urlparse
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -21,11 +15,7 @@ def __init__(self, channel: str, message: str) -> None:
         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})"
@@ -36,14 +26,12 @@ class Unsubscribed(Exception):
 
 
 class Broadcast:
-    def __init__(
-        self, url: Optional[str] = None, *, backend: Optional["BroadcastBackend"] = None
-    ) -> None:
+    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, Any] = {}
+        self._subscribers: dict[str, set[asyncio.Queue[Event | None]]] = {}
 
-    def _create_backend(self, url: str) -> "BroadcastBackend":
+    def _create_backend(self, url: str) -> BroadcastBackend:
         parsed_url = urlparse(url)
         if parsed_url.scheme in ("redis", "rediss"):
             from broadcaster._backends.redis import RedisBackend
@@ -66,7 +54,7 @@ def _create_backend(self, url: str) -> "BroadcastBackend":
             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
 
@@ -94,8 +82,8 @@ 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):
@@ -114,10 +102,10 @@ async def subscribe(self, channel: str) -> AsyncIterator["Subscriber"]:
 
 
 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] | None:
         try:
             while True:
                 yield await self.get()
diff --git a/pyproject.toml b/pyproject.toml
index 10f7cc5..c4e8036 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -24,6 +24,7 @@ classifiers = [
     "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
 ]
 dependencies = [
     "anyio>=3.4.0,<5",
@@ -48,14 +49,17 @@ include = [
 ]
 
 [tool.ruff]
-ignore = []
 line-length = 120
-select = ["E","F","W"]
 
-[tool.ruff.isort]
+[tool.ruff.lint]
+select = ["E", "F", "I", "FA", "UP"]
+
+[tool.ruff.lint.isort]
 combine-as-imports = true
 
 [tool.mypy]
+strict = true
+python_version = "3.8"
 disallow_untyped_defs = true
 ignore_missing_imports = true
 
diff --git a/requirements.txt b/requirements.txt
index dcf0fcf..ed2926b 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,18 +1,17 @@
 -e .[redis,postgres,kafka]
 
 # Documentation
-mkdocs==1.3.1
+mkdocs==1.5.3
+mkdocs-material==9.5.12
 mkautodoc==0.2.0
-mkdocs-material==8.4.0
 
 # Packaging
-build==0.10.0
-twine==4.0.1
+build==1.1.1
+twine==5.0.0
 
 # Tests & Linting
-ruff==0.0.277
-black==22.6.0
-coverage==6.4.4
-mypy==0.971
-pytest==7.1.2
-pytest-asyncio==0.19.0
\ No newline at end of file
+ruff==0.3.5
+coverage==7.4.3
+mypy==1.8.0
+pytest==8.0.2
+pytest-asyncio==0.23.6
\ No newline at end of file
diff --git a/scripts/check b/scripts/check
index e514548..d8fb02b 100755
--- a/scripts/check
+++ b/scripts/check
@@ -8,6 +8,6 @@ export SOURCE_FILES="broadcaster tests"
 
 set -x
 
-${PREFIX}black --check --diff --target-version=py37 $SOURCE_FILES
-${PREFIX}ruff check $SOURCE_FILES
+${PREFIX}ruff format --check --diff $SOURCE_FILES
 ${PREFIX}mypy $SOURCE_FILES
+${PREFIX}ruff check $SOURCE_FILES
diff --git a/scripts/lint b/scripts/lint
index 4751285..cb718d0 100755
--- a/scripts/lint
+++ b/scripts/lint
@@ -1,12 +1,12 @@
 #!/bin/sh -e
 
 export PREFIX=""
-if [ -d 'venv' ]; then
+if [ -d 'venv' ] ; then
     export PREFIX="venv/bin/"
 fi
 export SOURCE_FILES="broadcaster tests"
 
 set -x
 
-${PREFIX}ruff --fix $SOURCE_FILES
-${PREFIX}black --target-version=py37 $SOURCE_FILES
+${PREFIX}ruff format $SOURCE_FILES
+${PREFIX}ruff check --fix $SOURCE_FILES
\ No newline at end of file
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index 4cf9e45..b516ee2 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -1,16 +1,19 @@
-import pytest
-import typing
+from __future__ import annotations
+
 import asyncio
+import typing
+
+import pytest
 
 from broadcaster import Broadcast, BroadcastBackend, Event
 
 
 class CustomBackend(BroadcastBackend):
     def __init__(self, url: str):
-        self._subscribed: typing.Set = set()
+        self._subscribed: set[str] = set()
 
     async def connect(self) -> None:
-        self._published: asyncio.Queue = asyncio.Queue()
+        self._published: asyncio.Queue[Event] = asyncio.Queue()
 
     async def disconnect(self) -> None:
         pass
@@ -54,9 +57,7 @@ async def test_redis():
 
 @pytest.mark.asyncio
 async def test_postgres():
-    async with Broadcast(
-        "postgres://postgres:postgres@localhost:5432/broadcaster"
-    ) as broadcast:
+    async with Broadcast("postgres://postgres:postgres@localhost:5432/broadcaster") as broadcast:
         async with broadcast.subscribe("chatroom") as subscriber:
             await broadcast.publish("chatroom", "hello")
             event = await subscriber.get()
@@ -95,7 +96,5 @@ async def test_unknown_backend():
 
 @pytest.mark.asyncio
 async def test_needs_url_or_backend():
-    with pytest.raises(
-        AssertionError, match="Either `url` or `backend` must be provided."
-    ):
+    with pytest.raises(AssertionError, match="Either `url` or `backend` must be provided."):
         Broadcast()
diff --git a/tests/test_unsubscribe.py b/tests/test_unsubscribe.py
index 30f928b..ae89401 100644
--- a/tests/test_unsubscribe.py
+++ b/tests/test_unsubscribe.py
@@ -1,4 +1,5 @@
 import pytest
+
 from broadcaster import Broadcast
 
 

From 672d10d91f49ba386ccfde8766094d4bf308d25d Mon Sep 17 00:00:00 2001
From: Tom Christie <tom@tomchristie.com>
Date: Mon, 22 Apr 2024 16:43:27 +0100
Subject: [PATCH 21/33] Update README.md (#116)

---
 README.md | 1 +
 1 file changed, 1 insertion(+)

diff --git a/README.md b/README.md
index d11a9ed..0bff43b 100644
--- a/README.md
+++ b/README.md
@@ -100,6 +100,7 @@ class MyBackend(BroadcastBackend):
     ...
 
 broadcaster = Broadcaster(backend=MyBackend())
+```
 
 ## Where next?
 

From c4b4d599225a389f9ae5894020389163cbaa4ac8 Mon Sep 17 00:00:00 2001
From: Alex Oleshkevich <alex.oleshkevich@gmail.com>
Date: Mon, 22 Apr 2024 17:48:55 +0200
Subject: [PATCH 22/33] Add redis-streams backend (#115)

* Fixed #55 - Not Close Subscribing

* Added Redis Stream backend

* Fixed Linting Test

* Added Test For Redis Stream Backend

* align with master

* update docs

---------

Co-authored-by: tsotne <tsotnesharvadze@gmail.com>
Co-authored-by: Tom Christie <tom@tomchristie.com>
---
 README.md                      |  6 ++--
 broadcaster/_backends/redis.py | 53 ++++++++++++++++++++++++++++++++++
 broadcaster/_base.py           |  7 ++++-
 tests/test_broadcast.py        | 15 ++++++++++
 4 files changed, 77 insertions(+), 4 deletions(-)

diff --git a/README.md b/README.md
index 0bff43b..eb547ab 100644
--- a/README.md
+++ b/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'>
 
@@ -83,6 +83,7 @@ Python 3.8+
 
 * `Broadcast('memory://')`
 * `Broadcast("redis://localhost:6379")`
+* `Broadcast("redis-stream://localhost:6379")`
 * `Broadcast("postgres://localhost:5432/broadcaster")`
 * `Broadcast("kafka://localhost:9092")`
 
@@ -97,7 +98,6 @@ and pass it to the `broadcaster` via `backend` argument.
 from broadcaster import Broadcaster, BroadcastBackend
 
 class MyBackend(BroadcastBackend):
-    ...
 
 broadcaster = Broadcaster(backend=MyBackend())
 ```
@@ -112,6 +112,6 @@ state, make sure to strictly pin your requirements to `broadcaster==0.2.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.
diff --git a/broadcaster/_backends/redis.py b/broadcaster/_backends/redis.py
index 78ed7eb..e1f279e 100644
--- a/broadcaster/_backends/redis.py
+++ b/broadcaster/_backends/redis.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import asyncio
 import typing
 
@@ -47,3 +49,54 @@ async def _pubsub_listener(self) -> None:
                     message=message["data"].decode(),
                 )
                 await self._queue.put(event)
+
+
+StreamMessageType = typing.Tuple[bytes, typing.Tuple[typing.Tuple[bytes, typing.Dict[bytes, bytes]]]]
+
+
+class RedisStreamBackend(BroadcastBackend):
+    def __init__(self, url: str):
+        url = url.replace("redis-stream", "redis", 1)
+        self.streams: dict[str, str] = {}
+        self._ready = asyncio.Event()
+        self._producer = redis.Redis.from_url(url)
+        self._consumer = redis.Redis.from_url(url)
+
+    async def connect(self) -> None:
+        pass
+
+    async def disconnect(self) -> None:
+        await self._producer.aclose()
+        await self._consumer.aclose()
+
+    async def subscribe(self, channel: str) -> None:
+        try:
+            info = await self._consumer.xinfo_stream(channel)
+            last_id = info["last-generated-id"]
+        except redis.ResponseError:
+            last_id = "0"
+        self.streams[channel] = last_id
+        self._ready.set()
+
+    async def unsubscribe(self, channel: str) -> None:
+        self.streams.pop(channel, None)
+
+    async def publish(self, channel: str, message: typing.Any) -> None:
+        await self._producer.xadd(channel, {"message": message})
+
+    async def wait_for_messages(self) -> list[StreamMessageType]:
+        await self._ready.wait()
+        messages = None
+        while not messages:
+            messages = await self._consumer.xread(self.streams, count=1, block=100)
+        return messages
+
+    async def next_published(self) -> Event:
+        messages = await self.wait_for_messages()
+        stream, events = messages[0]
+        _msg_id, message = events[0]
+        self.streams[stream.decode("utf-8")] = _msg_id.decode("utf-8")
+        return Event(
+            channel=stream.decode("utf-8"),
+            message=message.get(b"message", b"").decode("utf-8"),
+        )
diff --git a/broadcaster/_base.py b/broadcaster/_base.py
index 0166034..4650e0a 100644
--- a/broadcaster/_base.py
+++ b/broadcaster/_base.py
@@ -38,6 +38,11 @@ def _create_backend(self, url: str) -> BroadcastBackend:
 
             return RedisBackend(url)
 
+        elif parsed_url.scheme == "redis-stream":
+            from broadcaster._backends.redis import RedisStreamBackend
+
+            return RedisStreamBackend(url)
+
         elif parsed_url.scheme in ("postgres", "postgresql"):
             from broadcaster._backends.postgres import PostgresBackend
 
@@ -88,7 +93,7 @@ async def subscribe(self, channel: str) -> AsyncIterator[Subscriber]:
         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)
 
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index b516ee2..d418508 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -55,6 +55,21 @@ async def test_redis():
             assert event.message == "hello"
 
 
+@pytest.mark.asyncio
+async def test_redis_stream():
+    async with Broadcast("redis-stream://localhost:6379") as broadcast:
+        async with broadcast.subscribe("chatroom") as subscriber:
+            await broadcast.publish("chatroom", "hello")
+            event = await subscriber.get()
+            assert event.channel == "chatroom"
+            assert event.message == "hello"
+        async with broadcast.subscribe("chatroom1") as subscriber:
+            await broadcast.publish("chatroom1", "hello")
+            event = await subscriber.get()
+            assert event.channel == "chatroom1"
+            assert event.message == "hello"
+
+
 @pytest.mark.asyncio
 async def test_postgres():
     async with Broadcast("postgres://postgres:postgres@localhost:5432/broadcaster") as broadcast:

From 4ff8fa688e8c22881e0bd58a99a816fa276656b3 Mon Sep 17 00:00:00 2001
From: Alex Oleshkevich <alex.oleshkevich@gmail.com>
Date: Thu, 6 Jun 2024 11:08:39 +0200
Subject: [PATCH 23/33] Add mqtt link (#126)

---
 README.md | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/README.md b/README.md
index eb547ab..b7129d4 100644
--- a/README.md
+++ b/README.md
@@ -91,7 +91,7 @@ Python 3.8+
 ### 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` 
+To do that you need to create a class which extends from `BroadcastBackend`
 and pass it to the `broadcaster` via `backend` argument.
 
 ```python
@@ -115,3 +115,10 @@ To be more capable we'd really want to add some additional backends, provide API
 * 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

From 6daa0d246f8c0711f930d60bdefb8f278c1a777f Mon Sep 17 00:00:00 2001
From: Alex Oleshkevich <alex.oleshkevich@gmail.com>
Date: Tue, 11 Jun 2024 12:57:18 +0200
Subject: [PATCH 24/33] [redis] defer listener initialization (#128)

---
 broadcaster/_backends/redis.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/broadcaster/_backends/redis.py b/broadcaster/_backends/redis.py
index e1f279e..ca00bc2 100644
--- a/broadcaster/_backends/redis.py
+++ b/broadcaster/_backends/redis.py
@@ -15,15 +15,17 @@ def __init__(self, url: str):
         self._pubsub = self._conn.pubsub()
         self._ready = asyncio.Event()
         self._queue: asyncio.Queue[Event] = asyncio.Queue()
-        self._listener = asyncio.create_task(self._pubsub_listener())
+        self._listener: asyncio.Task[None] | None = None
 
     async def connect(self) -> None:
+        self._listener = asyncio.create_task(self._pubsub_listener())
         await self._pubsub.connect()
 
     async def disconnect(self) -> None:
         await self._pubsub.aclose()
         await self._conn.aclose()
-        self._listener.cancel()
+        if self._listener is not None:
+            self._listener.cancel()
 
     async def subscribe(self, channel: str) -> None:
         self._ready.set()

From 399442d0677127284ce9263fc06bf7c9c81311a9 Mon Sep 17 00:00:00 2001
From: Alex Oleshkevich <alex.oleshkevich@gmail.com>
Date: Tue, 11 Jun 2024 12:57:30 +0200
Subject: [PATCH 25/33] Improvements to Kafka backend (#125)

---
 broadcaster/_backends/kafka.py | 16 ++++++++++++++--
 tests/test_broadcast.py        | 12 +++++++++++-
 2 files changed, 25 insertions(+), 3 deletions(-)

diff --git a/broadcaster/_backends/kafka.py b/broadcaster/_backends/kafka.py
index e577769..46fba6b 100644
--- a/broadcaster/_backends/kafka.py
+++ b/broadcaster/_backends/kafka.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import typing
 from urllib.parse import urlparse
 
@@ -10,9 +11,11 @@
 
 
 class KafkaBackend(BroadcastBackend):
-    def __init__(self, url: str):
-        self._servers = [urlparse(url).netloc]
+    def __init__(self, urls: str | list[str]) -> None:
+        urls = [urls] if isinstance(urls, str) else urls
+        self._servers = [urlparse(url).netloc for url in urls]
         self._consumer_channels: set[str] = set()
+        self._ready = asyncio.Event()
 
     async def connect(self) -> None:
         self._producer = AIOKafkaProducer(bootstrap_servers=self._servers)
@@ -27,6 +30,7 @@ async def disconnect(self) -> None:
     async def subscribe(self, channel: str) -> None:
         self._consumer_channels.add(channel)
         self._consumer.subscribe(topics=self._consumer_channels)
+        await self._wait_for_assignment()
 
     async def unsubscribe(self, channel: str) -> None:
         self._consumer.unsubscribe()
@@ -35,5 +39,13 @@ async def publish(self, channel: str, message: typing.Any) -> None:
         await self._producer.send_and_wait(channel, message.encode("utf8"))
 
     async def next_published(self) -> Event:
+        await self._ready.wait()
         message = await self._consumer.getone()
         return Event(channel=message.topic, message=message.value.decode("utf8"))
+
+    async def _wait_for_assignment(self) -> None:
+        """Wait for the consumer to be assigned to the partition."""
+        while not self._consumer.assignment():
+            await asyncio.sleep(0.001)
+
+        self._ready.set()
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index d418508..e73b9eb 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -6,6 +6,7 @@
 import pytest
 
 from broadcaster import Broadcast, BroadcastBackend, Event
+from broadcaster._backends.kafka import KafkaBackend
 
 
 class CustomBackend(BroadcastBackend):
@@ -80,7 +81,6 @@ async def test_postgres():
             assert event.message == "hello"
 
 
-@pytest.mark.skip("Deadlock on `next_published`")
 @pytest.mark.asyncio
 async def test_kafka():
     async with Broadcast("kafka://localhost:9092") as broadcast:
@@ -91,6 +91,16 @@ async def test_kafka():
             assert event.message == "hello"
 
 
+@pytest.mark.asyncio
+async def test_kafka_multiple_urls():
+    async with Broadcast(backend=KafkaBackend(urls=["kafka://localhost:9092", "kafka://localhost:9092"])) as broadcast:
+        async with broadcast.subscribe("chatroom") as subscriber:
+            await broadcast.publish("chatroom", "hello")
+            event = await subscriber.get()
+            assert event.channel == "chatroom"
+            assert event.message == "hello"
+
+
 @pytest.mark.asyncio
 async def test_custom():
     backend = CustomBackend("")

From d6806ace2478daa5a4caaf78fa2487c0f55740ee Mon Sep 17 00:00:00 2001
From: Alex Oleshkevich <alex.oleshkevich@gmail.com>
Date: Thu, 27 Jun 2024 20:08:33 +0200
Subject: [PATCH 26/33] Bump version (#131)

---
 README.md               | 2 +-
 broadcaster/__init__.py | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/README.md b/README.md
index b7129d4..959b34d 100644
--- a/README.md
+++ b/README.md
@@ -107,7 +107,7 @@ broadcaster = Broadcaster(backend=MyBackend())
 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...
 
diff --git a/broadcaster/__init__.py b/broadcaster/__init__.py
index a41bbeb..8259f1d 100644
--- a/broadcaster/__init__.py
+++ b/broadcaster/__init__.py
@@ -1,5 +1,5 @@
 from ._backends.base import BroadcastBackend
 from ._base import Broadcast, Event
 
-__version__ = "0.2.0"
+__version__ = "0.3.0"
 __all__ = ["Broadcast", "Event", "BroadcastBackend"]

From 22e8b2afb131321dcd0a9a06be2d860849cfbe5a Mon Sep 17 00:00:00 2001
From: Alex Oleshkevich <alex.oleshkevich@gmail.com>
Date: Thu, 27 Jun 2024 21:40:43 +0200
Subject: [PATCH 27/33] sync publish script with starlette (#132)

---
 scripts/publish | 34 +++++++++++++---------------------
 1 file changed, 13 insertions(+), 21 deletions(-)

diff --git a/scripts/publish b/scripts/publish
index 238d8cb..400f43f 100755
--- a/scripts/publish
+++ b/scripts/publish
@@ -1,34 +1,26 @@
 #!/bin/sh -e
 
-export VERSION=`cat broadcaster/__init__.py | grep __version__ | sed "s/__version__ = //" | sed "s/'//g"`
-export PREFIX=""
+VERSION_FILE="broadcaster/__init__.py"
+
 if [ -d 'venv' ] ; then
-    export PREFIX="venv/bin/"
+    PREFIX="venv/bin/"
+else
+    PREFIX=""
 fi
 
-scripts/clean
+if [ ! -z "$GITHUB_ACTIONS" ]; then
+  git config --local user.email "41898282+github-actions[bot]@users.noreply.github.com"
+  git config --local user.name "GitHub Action"
 
-if ! command -v "${PREFIX}twine" &>/dev/null ; then
-    echo "Unable to find the 'twine' command."
-    echo "Install from PyPI, using '${PREFIX}pip install twine'."
-    exit 1
-fi
+  VERSION=`grep __version__ ${VERSION_FILE} | grep -o '[0-9][^"]*'`
 
-if ! command -v "${PREFIX}wheel" &>/dev/null ; then
-    echo "Unable to find the 'wheel' command."
-    echo "Install from PyPI, using '${PREFIX}pip install wheel'."
+  if [ "refs/tags/${VERSION}" != "${GITHUB_REF}" ] ; then
+    echo "GitHub Ref '${GITHUB_REF}' did not match package version '${VERSION}'"
     exit 1
+  fi
 fi
 
-find broadcaster -type f -name "*.py[co]" -delete
-find broadcaster -type d -name __pycache__ -delete
+set -x
 
-${PREFIX}python setup.py sdist bdist_wheel
 ${PREFIX}twine upload dist/*
-#${PREFIX}mkdocs gh-deploy
-
-echo "You probably want to also tag the version now:"
-echo "git tag -a ${VERSION} -m 'version ${VERSION}'"
-echo "git push --tags"
 
-scripts/clean

From 528cf42c809c172ae42036baea4bb8585820e4dc Mon Sep 17 00:00:00 2001
From: BUG-Fly <61271333+Fly-Playgroud@users.noreply.github.com>
Date: Fri, 2 Aug 2024 04:59:24 +0800
Subject: [PATCH 28/33] Fix redisBackend `_pubsub_listener`  just listen once
 (#134)

* fix: solve the redisBackend `_pubsub_listener` just listen once.

* chore: change the comment
---
 broadcaster/_backends/redis.py | 22 ++++++++++++++--------
 1 file changed, 14 insertions(+), 8 deletions(-)

diff --git a/broadcaster/_backends/redis.py b/broadcaster/_backends/redis.py
index ca00bc2..3122d4c 100644
--- a/broadcaster/_backends/redis.py
+++ b/broadcaster/_backends/redis.py
@@ -43,14 +43,20 @@ async def next_published(self) -> Event:
     async def _pubsub_listener(self) -> None:
         # redis-py does not listen to the pubsub connection if there are no channels subscribed
         # so we need to wait until the first channel is subscribed to start listening
-        await self._ready.wait()
-        async for message in self._pubsub.listen():
-            if message["type"] == "message":
-                event = Event(
-                    channel=message["channel"].decode(),
-                    message=message["data"].decode(),
-                )
-                await self._queue.put(event)
+        while True:
+            await self._ready.wait()
+            async for message in self._pubsub.listen():
+                if message["type"] == "message":
+                    event = Event(
+                        channel=message["channel"].decode(),
+                        message=message["data"].decode(),
+                    )
+                    await self._queue.put(event)
+
+            # when no channel subscribed, clear the event.
+            # And then in next loop, event will blocked again until
+            # the new channel subscribed.Now asyncio.Task will not exit again.
+            self._ready.clear()
 
 
 StreamMessageType = typing.Tuple[bytes, typing.Tuple[typing.Tuple[bytes, typing.Dict[bytes, bytes]]]]

From 23c9b40136231d0a78d1299588b62d7162067ff4 Mon Sep 17 00:00:00 2001
From: "alex.oleshkevich" <alex.oleshkevich@gmail.com>
Date: Thu, 1 Aug 2024 23:14:22 +0200
Subject: [PATCH 29/33] bump version

---
 broadcaster/__init__.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/broadcaster/__init__.py b/broadcaster/__init__.py
index 8259f1d..5ab9306 100644
--- a/broadcaster/__init__.py
+++ b/broadcaster/__init__.py
@@ -1,5 +1,5 @@
 from ._backends.base import BroadcastBackend
 from ._base import Broadcast, Event
 
-__version__ = "0.3.0"
+__version__ = "0.3.1"
 __all__ = ["Broadcast", "Event", "BroadcastBackend"]

From 69cf29a41066f53a45498f0dfa36288befd73dd7 Mon Sep 17 00:00:00 2001
From: Alex Oleshkevich <alex.oleshkevich@gmail.com>
Date: Fri, 23 Aug 2024 16:43:37 +0200
Subject: [PATCH 30/33] #136 improve typing (#139)

---
 broadcaster/_backends/base.py  |  4 ++--
 broadcaster/_backends/kafka.py | 12 +++++++++---
 broadcaster/_backends/redis.py |  2 +-
 broadcaster/_base.py           |  2 +-
 4 files changed, 13 insertions(+), 7 deletions(-)

diff --git a/broadcaster/_backends/base.py b/broadcaster/_backends/base.py
index 1c65b86..7017df3 100644
--- a/broadcaster/_backends/base.py
+++ b/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:
diff --git a/broadcaster/_backends/kafka.py b/broadcaster/_backends/kafka.py
index 46fba6b..f09dca1 100644
--- a/broadcaster/_backends/kafka.py
+++ b/broadcaster/_backends/kafka.py
@@ -18,8 +18,8 @@ def __init__(self, urls: str | list[str]) -> None:
         self._ready = asyncio.Event()
 
     async def connect(self) -> None:
-        self._producer = AIOKafkaProducer(bootstrap_servers=self._servers)
-        self._consumer = AIOKafkaConsumer(bootstrap_servers=self._servers)
+        self._producer = AIOKafkaProducer(bootstrap_servers=self._servers)  # pyright: ignore
+        self._consumer = AIOKafkaConsumer(bootstrap_servers=self._servers)  # pyright: ignore
         await self._producer.start()
         await self._consumer.start()
 
@@ -41,7 +41,13 @@ async def publish(self, channel: str, message: typing.Any) -> None:
     async def next_published(self) -> Event:
         await self._ready.wait()
         message = await self._consumer.getone()
-        return Event(channel=message.topic, message=message.value.decode("utf8"))
+        value = message.value
+
+        # for type compatibility:
+        # we declare Event.message as str, so convert None to empty string
+        if value is None:
+            value = b""
+        return Event(channel=message.topic, message=value.decode("utf8"))
 
     async def _wait_for_assignment(self) -> None:
         """Wait for the consumer to be assigned to the partition."""
diff --git a/broadcaster/_backends/redis.py b/broadcaster/_backends/redis.py
index 3122d4c..1be4195 100644
--- a/broadcaster/_backends/redis.py
+++ b/broadcaster/_backends/redis.py
@@ -65,7 +65,7 @@ async def _pubsub_listener(self) -> None:
 class RedisStreamBackend(BroadcastBackend):
     def __init__(self, url: str):
         url = url.replace("redis-stream", "redis", 1)
-        self.streams: dict[str, str] = {}
+        self.streams: dict[bytes | str | memoryview, int | bytes | str | memoryview] = {}
         self._ready = asyncio.Event()
         self._producer = redis.Redis.from_url(url)
         self._consumer = redis.Redis.from_url(url)
diff --git a/broadcaster/_base.py b/broadcaster/_base.py
index 4650e0a..e966a6f 100644
--- a/broadcaster/_base.py
+++ b/broadcaster/_base.py
@@ -110,7 +110,7 @@ class Subscriber:
     def __init__(self, queue: asyncio.Queue[Event | None]) -> None:
         self._queue = queue
 
-    async def __aiter__(self) -> AsyncGenerator[Event | None, None] | None:
+    async def __aiter__(self) -> AsyncGenerator[Event | None, None]:
         try:
             while True:
                 yield await self.get()

From a422d8a378cfcf07e938473754f91edf34b0e0c2 Mon Sep 17 00:00:00 2001
From: Alex Oleshkevich <alex.oleshkevich@gmail.com>
Date: Sat, 2 Nov 2024 16:33:15 +0100
Subject: [PATCH 31/33] export backends (#145)

---
 broadcaster/__init__.py                         |  2 +-
 broadcaster/_base.py                            | 12 ++++++------
 broadcaster/{_backends => backends}/__init__.py |  0
 broadcaster/{_backends => backends}/base.py     |  0
 broadcaster/{_backends => backends}/kafka.py    |  0
 broadcaster/{_backends => backends}/memory.py   |  0
 broadcaster/{_backends => backends}/postgres.py |  0
 broadcaster/{_backends => backends}/redis.py    |  0
 tests/test_broadcast.py                         |  2 +-
 9 files changed, 8 insertions(+), 8 deletions(-)
 rename broadcaster/{_backends => backends}/__init__.py (100%)
 rename broadcaster/{_backends => backends}/base.py (100%)
 rename broadcaster/{_backends => backends}/kafka.py (100%)
 rename broadcaster/{_backends => backends}/memory.py (100%)
 rename broadcaster/{_backends => backends}/postgres.py (100%)
 rename broadcaster/{_backends => backends}/redis.py (100%)

diff --git a/broadcaster/__init__.py b/broadcaster/__init__.py
index 5ab9306..0bcd9d2 100644
--- a/broadcaster/__init__.py
+++ b/broadcaster/__init__.py
@@ -1,5 +1,5 @@
-from ._backends.base import BroadcastBackend
 from ._base import Broadcast, Event
+from .backends.base import BroadcastBackend
 
 __version__ = "0.3.1"
 __all__ = ["Broadcast", "Event", "BroadcastBackend"]
diff --git a/broadcaster/_base.py b/broadcaster/_base.py
index e966a6f..a63b22b 100644
--- a/broadcaster/_base.py
+++ b/broadcaster/_base.py
@@ -6,7 +6,7 @@
 from urllib.parse import urlparse
 
 if TYPE_CHECKING:  # pragma: no cover
-    from broadcaster._backends.base import BroadcastBackend
+    from broadcaster.backends.base import BroadcastBackend
 
 
 class Event:
@@ -34,27 +34,27 @@ def __init__(self, url: str | None = None, *, backend: BroadcastBackend | None =
     def _create_backend(self, url: str) -> BroadcastBackend:
         parsed_url = urlparse(url)
         if parsed_url.scheme in ("redis", "rediss"):
-            from broadcaster._backends.redis import RedisBackend
+            from broadcaster.backends.redis import RedisBackend
 
             return RedisBackend(url)
 
         elif parsed_url.scheme == "redis-stream":
-            from broadcaster._backends.redis import RedisStreamBackend
+            from broadcaster.backends.redis import RedisStreamBackend
 
             return RedisStreamBackend(url)
 
         elif parsed_url.scheme in ("postgres", "postgresql"):
-            from broadcaster._backends.postgres import PostgresBackend
+            from broadcaster.backends.postgres import PostgresBackend
 
             return PostgresBackend(url)
 
         if parsed_url.scheme == "kafka":
-            from broadcaster._backends.kafka import KafkaBackend
+            from broadcaster.backends.kafka import KafkaBackend
 
             return KafkaBackend(url)
 
         elif parsed_url.scheme == "memory":
-            from broadcaster._backends.memory import MemoryBackend
+            from broadcaster.backends.memory import MemoryBackend
 
             return MemoryBackend(url)
         raise ValueError(f"Unsupported backend: {parsed_url.scheme}")
diff --git a/broadcaster/_backends/__init__.py b/broadcaster/backends/__init__.py
similarity index 100%
rename from broadcaster/_backends/__init__.py
rename to broadcaster/backends/__init__.py
diff --git a/broadcaster/_backends/base.py b/broadcaster/backends/base.py
similarity index 100%
rename from broadcaster/_backends/base.py
rename to broadcaster/backends/base.py
diff --git a/broadcaster/_backends/kafka.py b/broadcaster/backends/kafka.py
similarity index 100%
rename from broadcaster/_backends/kafka.py
rename to broadcaster/backends/kafka.py
diff --git a/broadcaster/_backends/memory.py b/broadcaster/backends/memory.py
similarity index 100%
rename from broadcaster/_backends/memory.py
rename to broadcaster/backends/memory.py
diff --git a/broadcaster/_backends/postgres.py b/broadcaster/backends/postgres.py
similarity index 100%
rename from broadcaster/_backends/postgres.py
rename to broadcaster/backends/postgres.py
diff --git a/broadcaster/_backends/redis.py b/broadcaster/backends/redis.py
similarity index 100%
rename from broadcaster/_backends/redis.py
rename to broadcaster/backends/redis.py
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index e73b9eb..a8bd3eb 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -6,7 +6,7 @@
 import pytest
 
 from broadcaster import Broadcast, BroadcastBackend, Event
-from broadcaster._backends.kafka import KafkaBackend
+from broadcaster.backends.kafka import KafkaBackend
 
 
 class CustomBackend(BroadcastBackend):

From c892e525d0794649065d541a707a1a4c459df574 Mon Sep 17 00:00:00 2001
From: Alex Oleshkevich <alex.oleshkevich@gmail.com>
Date: Sun, 16 Mar 2025 14:14:47 +0100
Subject: [PATCH 32/33] allow preconfigured redis clients (#146)

---
 broadcaster/backends/redis.py | 15 +++++++++++----
 tests/test_broadcast.py       | 19 +++++++++++++++++++
 2 files changed, 30 insertions(+), 4 deletions(-)

diff --git a/broadcaster/backends/redis.py b/broadcaster/backends/redis.py
index 1be4195..effb166 100644
--- a/broadcaster/backends/redis.py
+++ b/broadcaster/backends/redis.py
@@ -10,8 +10,15 @@
 
 
 class RedisBackend(BroadcastBackend):
-    def __init__(self, url: str):
-        self._conn = redis.Redis.from_url(url)
+    _conn: redis.Redis
+
+    def __init__(self, url: str | None = None, *, conn: redis.Redis | None = None):
+        if url is None:
+            assert conn is not None, "conn must be provided if url is not"
+            self._conn = conn
+        else:
+            self._conn = redis.Redis.from_url(url)
+
         self._pubsub = self._conn.pubsub()
         self._ready = asyncio.Event()
         self._queue: asyncio.Queue[Event] = asyncio.Queue()
@@ -19,10 +26,10 @@ def __init__(self, url: str):
 
     async def connect(self) -> None:
         self._listener = asyncio.create_task(self._pubsub_listener())
-        await self._pubsub.connect()
+        await self._pubsub.connect()  # type: ignore[no-untyped-call]
 
     async def disconnect(self) -> None:
-        await self._pubsub.aclose()
+        await self._pubsub.aclose()  # type: ignore[no-untyped-call]
         await self._conn.aclose()
         if self._listener is not None:
             self._listener.cancel()
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index a8bd3eb..b88b317 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -4,9 +4,11 @@
 import typing
 
 import pytest
+from redis import asyncio as redis
 
 from broadcaster import Broadcast, BroadcastBackend, Event
 from broadcaster.backends.kafka import KafkaBackend
+from broadcaster.backends.redis import RedisBackend
 
 
 class CustomBackend(BroadcastBackend):
@@ -56,6 +58,23 @@ async def test_redis():
             assert event.message == "hello"
 
 
+@pytest.mark.asyncio
+async def test_redis_configured_client():
+    backend = RedisBackend(conn=redis.Redis.from_url("redis://localhost:6379"))
+    async with Broadcast(backend=backend) as broadcast:
+        async with broadcast.subscribe("chatroom") as subscriber:
+            await broadcast.publish("chatroom", "hello")
+            event = await subscriber.get()
+            assert event.channel == "chatroom"
+            assert event.message == "hello"
+
+
+@pytest.mark.asyncio
+async def test_redis_requires_url_or_connection():
+    with pytest.raises(AssertionError, match="conn must be provided if url is not"):
+        RedisBackend()
+
+
 @pytest.mark.asyncio
 async def test_redis_stream():
     async with Broadcast("redis-stream://localhost:6379") as broadcast:

From 6b3ea71d4f8fb038fa7d357a1fb3750d58ac614d Mon Sep 17 00:00:00 2001
From: Alex Oleshkevich <alex.oleshkevich@gmail.com>
Date: Wed, 9 Apr 2025 19:09:16 +0200
Subject: [PATCH 33/33] bump version (#153)

---
 broadcaster/__init__.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/broadcaster/__init__.py b/broadcaster/__init__.py
index 0bcd9d2..3ee97d2 100644
--- a/broadcaster/__init__.py
+++ b/broadcaster/__init__.py
@@ -1,5 +1,5 @@
 from ._base import Broadcast, Event
 from .backends.base import BroadcastBackend
 
-__version__ = "0.3.1"
+__version__ = "0.3.2"
 __all__ = ["Broadcast", "Event", "BroadcastBackend"]