broadcaster

This example demonstrates how to integrate Broadcaster with Starlette to create a real-time chat application. It uses broadcast.publish to receive messages from a WebSocket and broadcast.subscribe to send messages from the broadcast backend back to the WebSocket.

Requirements: starlette, uvicorn, jinja2 Run command: uvicorn example:app

import anyio
from broadcaster import Broadcast
from starlette.applications import Starlette
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()

    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):
    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],
)

The docker-compose.yaml file provides a pre-configured environment for running Broadcaster with various backends. It includes services for Zookeeper, Kafka, Redis, and PostgreSQL.

Note that these services are configured with specific environment variables and ports that may be required for the Broadcaster backends to connect successfully.

version: '3'
services:
  zookeeper:
    image: "confluentinc/cp-zookeeper"
    hostname: zookeeper
    ports:
      - 32181:32181
    environment:
      - ZOOKEEPER_CLIENT_PORT=32181
      - ALLOW_ANONYMOUS_LOGIN=yes
  kafka:
    image: confluentinc/cp-kafka
    hostname: kafka
    ports:
    - 9092:9092
    - 29092:29092
    depends_on:
    - zookeeper
    environment:
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:32181
      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT_HOST://localhost:29092,PLAINTEXT://localhost:9092
      - KAFKA_BROKER_ID=1
      - ALLOW_PLAINTEXT_LISTENER=yes
  redis:
    image: "redis:alpine"
    ports:
      - 6379:6379
  postgres:
    image: "postgres:12"
    environment:
      - POSTGRES_DB=broadcaster
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_HOST_AUTH_METHOD=trust
      - POSTGRES_USER=postgres
    ports:
      - 5432:5432