Skip to content

Implement Redis-backed WebSocket Pub/Sub helpers#32

Open
emiliano-go wants to merge 6 commits into
redis:mainfrom
emiliano-go:feat/pubsub-websocket
Open

Implement Redis-backed WebSocket Pub/Sub helpers#32
emiliano-go wants to merge 6 commits into
redis:mainfrom
emiliano-go:feat/pubsub-websocket

Conversation

@emiliano-go

@emiliano-go emiliano-go commented Jul 23, 2026

Copy link
Copy Markdown

Implement Redis-backed WebSocket Pub/Sub helpers (Issue #24)

5 commits:

feat: add PubSubManager for Redis Pub/Sub messaging
pubsub.py, __init__.py, deps.py, telemetry.py

PubSubManager with subscribe/listen/publish/close. FastAPI DI via get_pubsub_manager (sub-dependency on Depends(get_async_redis)). record_pubsub_publish telemetry counter.

test: add 25 PubSub unit tests
tests/unit/test_pubsub.py

Core lifecycle, 13 edge cases, memory leak via weakref, DI integration.

test: add PubSub integration tests
tests/integration/test_pubsub.py

Cross-connection bridge, subscriber count, unsub isolation (requires real Redis).

docs: add PubSub user guide and API reference
docs/guide/pubsub.md, docs/api/reference.md, mkdocs.yml

Guide with WebSocket examples, connection pool limitation note, DI override chains.

examples: add PubSub WebSocket chat and notify endpoint
examples/main.py

/notify HTTP publish and /ws/chat/{room} WebSocket broadcast loop.

Design decisions

  • get_pubsub_manager uses Depends(get_async_redis) instead of calling get_async_redis directly, so callers can override the Redis client without re-registering the PubSub provider.
  • listen() filters system events and yields only message-type events.
  • Each subscribe() call consumes one dedicated pool connection (inherent Redis Pub/Sub limitation), documented in the guide.
  • No new middleware or lifespan changes needed; Pub/Sub reuses the existing shared pool.

Closes #24

PubSubManager with subscribe/listen/publish/close methods.
FastAPI DI via get_pubsub_manager (sub-dependency on Depends(get_async_redis) for proper override chains).
record_pubsub_publish OpenTelemetry counter.

Each subscribe() call consumes one dedicated pool connection
(inherent Redis Pub/Sub limitation).
Core lifecycle, 13 edge cases (empty channels, unicode, binary-safe,
burst, concurrent subscribers, 1MB payload), memory leak via weakref,
DI integration, and error handling.
Cross-connection bridge, subscriber count, and unsub isolation.
Requires a real Redis server.
Guide with WebSocket examples, connection pool limitation note,
and DI override chains. API reference section with PubSubManagerDep,
get_pubsub_manager, and PubSubManager.
/notify HTTP endpoint for publishing messages to Redis channels.
/ws/chat/{room} WebSocket endpoint for bidirectional chat
via Redis Pub/Sub broadcast.
get_async_redis now takes starlette.requests.HTTPConnection instead
of fastapi.Request, so it works in both HTTP endpoints and WebSocket
handlers. PubSubManagerDep and PubSubManager can now be injected
directly in WebSocket endpoints without manually calling
_get_pool_state.

This approach was inspired by PR redis#27 (massooti).
@emiliano-go

emiliano-go commented Jul 23, 2026

Copy link
Copy Markdown
Author

I just noticed PR #27 after opening this one (started working yesterday at night, too tired). The two PRs target the same feature but go about it differently.
I've incorporated the HTTPConnection approach from PR #27 into get_async_redis so it works in both HTTP and WebSocket endpoints directly. Credit to @massooti for that improvement.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Suggest adding Redis-backed WebSocket pub/sub helpers

1 participant