fix taskiq-python#65: avoid worker crash in case connection is broken

khizunov · Jul 1, 2024 · ece9df8 · ece9df8
1 parent 65a0b6c
commit ece9df8
Showing 1 changed file with 9 additions and 5 deletions.
diff --git a/taskiq_redis/redis_broker.py b/taskiq_redis/redis_broker.py
@@ -123,8 +123,12 @@ async def listen(self) -> AsyncGenerator[bytes, None]:
         :yields: broker messages.
         """
         redis_brpop_data_position = 1
-        async with Redis(connection_pool=self.connection_pool) as redis_conn:
-            while True:
-                yield (await redis_conn.brpop(self.queue_name))[
-                    redis_brpop_data_position
-                ]
+        while True:
+            try:
+                async with Redis(connection_pool=self.connection_pool) as redis_conn:
+                    yield (await redis_conn.brpop(self.queue_name))[
+                        redis_brpop_data_position
+                    ]
+            except ConnectionError as exc:
+                logger.warning("Redis connection error: %s", exc)
+                continue