广播器

从原始的broadcaster分支以获取可重用的补丁修复

广播器通过提供简单广播API到多个不同的后端服务，帮助您开发实时流功能。

它目前支持Redis PUB/SUB、Apache Kafka和Postgres LISTEN/NOTIFY，以及一个简单的内存后端，您可以使用它进行本地开发或在测试期间使用。

以下是一个简单的WebSocket聊天应用程序后端代码的完整示例

app.py

# Requires: `starlette`, `uvicorn`, `jinja2`
# Run with `uvicorn example:app`
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


broadcast = Broadcast("redis://localhost:6379")
templates = Jinja2Templates("templates")


async def homepage(request):
    template = "index.html"
    context = {"request": request}
    return templates.TemplateResponse(template, context)


async def chatroom_ws(websocket):
    await websocket.accept()
    await run_until_first_complete(
        (chatroom_ws_receiver, {"websocket": websocket}),
        (chatroom_ws_sender, {"websocket": websocket}),
    )


async def chatroom_ws_receiver(websocket):
    async for message in websocket.iter_text():
        await broadcast.publish(channel="chatroom", message=message)


async def chatroom_ws_sender(websocket):
    async with broadcast.subscribe(channel="chatroom") as subscriber:
        async for event in subscriber:
            await websocket.send_text(event.message)


routes = [
    Route("/", homepage),
    WebSocketRoute("/", chatroom_ws, name='chatroom_ws'),
]


app = Starlette(
    routes=routes, on_startup=[broadcast.connect], on_shutdown=[broadcast.disconnect],
)

前端HTML模板在此处提供，并改编自Pieter Noordhuis的PUB/SUB演示。

需求

Python 3.7+

安装

• pip安装broadcaster
• pip安装broadcaster[redis]
• pip安装broadcaster[postgres]
• pip安装broadcaster[kafka]

可用的后端

• Broadcast('memory://')
• Broadcast("redis://localhost:6379")
• Broadcast("postgres://localhost:5432/broadcaster")
• Broadcast("kafka://localhost:9092")

接下来做什么？

目前 广播器 处于 Alpha 阶段，应被视为工作设计文档。

API 可能会发生变化。如果您确实想在当前状态下使用广播器，请确保将您的需求严格绑定到 broadcaster==0.2.0。

为了提高能力，我们非常希望添加一些额外的后端，提供 API 支持从持久存储读取最近的事件历史，并提供序列化/反序列化 API...

• 支持序列化和反序列化以支持广播结构化数据。
• 后端支持 Redis Streams、Apache Kafka 和 RabbitMQ。
• 为支持持久化的后端添加对 subscribe('chatroom', history=100) 的支持。（Redis Streams、Apache Kafka）这将允许应用程序订阅频道更新，同时也能获得最近事件的初始窗口。我们可能还想支持一些基本的分页操作，以便应用程序能够扫描事件历史。
• 支持后端支持的模式订阅。