diff --git a/requirements.txt b/requirements.txt index 6093d21..5945b3a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ sentry-sdk -aioredis +aredis ariadne starlette uvicorn diff --git a/services/presence.py b/services/presence.py index 0447e19..dab45a1 100644 --- a/services/presence.py +++ b/services/presence.py @@ -6,6 +6,6 @@ async def notify_message(message, chat_id: str): channel_name = f"chat:{chat_id}" data = {**message, "kind": "new_message"} try: - await redis.execute_pubsub("PUBLISH", channel_name, json.dumps(data)) + await redis.publish(channel_name, json.dumps(data)) except Exception as e: print(f"Failed to publish to channel {channel_name}: {e}") diff --git a/services/redis.py b/services/redis.py index f61389c..620fbd5 100644 --- a/services/redis.py +++ b/services/redis.py @@ -1,5 +1,5 @@ import asyncio -import aioredis +import aredis from settings import REDIS_URL @@ -10,13 +10,10 @@ class RedisCache: self._redis = None async def connect(self): - pool = aioredis.ConnectionPool.from_url( - self._uri, encoding="utf-8", max_connections=10 - ) - self._redis = aioredis.Redis(connection_pool=pool) + self._redis = aredis.StrictRedis.from_url(self._uri, decode_responses=True) async def disconnect(self): - await self._redis.wait_closed() + self._redis.connection_pool.disconnect() self._redis = None async def execute(self, command, *args, **kwargs): @@ -28,29 +25,25 @@ class RedisCache: except Exception: pass - async def execute_pubsub(self, command, *args, **kwargs): - while not self._redis: - await asyncio.sleep(1) - try: - print("[redis] " + command + " " + " ".join(args)) - return await self._redis.execute_pubsub(command, *args, **kwargs) - except Exception: - pass - async def subscribe(self, *channels): if not self._redis: await self.connect() for channel in channels: - await self._redis.execute_pubsub("SUBSCRIBE", channel) + await self._redis.subscribe(channel) self.pubsub_channels.append(channel) async def unsubscribe(self, *channels): if not self._redis: return for channel in channels: - await self._redis.execute_pubsub("UNSUBSCRIBE", channel) + await self._redis.unsubscribe(channel) self.pubsub_channels.remove(channel) + async def publish(self, channel, data): + if not self._redis: + return + await self._redis.publish(channel, data) + async def lrange(self, key, start, stop): print(f"[redis] LRANGE {key} {start} {stop}") return await self._redis.lrange(key, start, stop)