Listen to events in Redis and process them.
()
| 46 | |
| 47 | |
| 48 | async def process_events(): |
| 49 | """Listen to events in Redis and process them.""" |
| 50 | redis = aioredis.from_url("redis://127.0.0.1:6379/1") |
| 51 | pubsub = redis.pubsub() |
| 52 | await pubsub.subscribe("events") |
| 53 | async for message in pubsub.listen(): |
| 54 | if message["type"] != "message": |
| 55 | continue |
| 56 | payload = message["data"].decode() |
| 57 | # Broadcast event to all users who have permissions to see it. |
| 58 | event = json.loads(payload) |
| 59 | recipients = ( |
| 60 | websocket |
| 61 | for websocket, connection in CONNECTIONS.items() |
| 62 | if event["content_type_id"] in connection["content_type_ids"] |
| 63 | ) |
| 64 | broadcast(recipients, payload) |
| 65 | |
| 66 | |
| 67 | async def main(): |