|
@ -78,14 +78,19 @@ class AsyncRedisManager(AsyncPubSubManager): # pragma: no cover |
|
|
self._redis_connect() |
|
|
self._redis_connect() |
|
|
return await self.redis.publish( |
|
|
return await self.redis.publish( |
|
|
self.channel, pickle.dumps(data)) |
|
|
self.channel, pickle.dumps(data)) |
|
|
except RedisError: |
|
|
except RedisError as exc: |
|
|
if retry: |
|
|
if retry: |
|
|
self._get_logger().error('Cannot publish to redis... ' |
|
|
self._get_logger().error( |
|
|
'retrying') |
|
|
'Cannot publish to redis... ' |
|
|
|
|
|
'retrying', |
|
|
|
|
|
extra={"redis_exception": str(exc)}) |
|
|
retry = False |
|
|
retry = False |
|
|
else: |
|
|
else: |
|
|
self._get_logger().error('Cannot publish to redis... ' |
|
|
self._get_logger().error( |
|
|
'giving up') |
|
|
'Cannot publish to redis... ' |
|
|
|
|
|
'giving up', |
|
|
|
|
|
extra={"redis_exception": str(exc)}) |
|
|
|
|
|
|
|
|
break |
|
|
break |
|
|
|
|
|
|
|
|
async def _redis_listen_with_retries(self): |
|
|
async def _redis_listen_with_retries(self): |
|
@ -99,10 +104,11 @@ class AsyncRedisManager(AsyncPubSubManager): # pragma: no cover |
|
|
retry_sleep = 1 |
|
|
retry_sleep = 1 |
|
|
async for message in self.pubsub.listen(): |
|
|
async for message in self.pubsub.listen(): |
|
|
yield message |
|
|
yield message |
|
|
except RedisError: |
|
|
except RedisError as exc: |
|
|
self._get_logger().error('Cannot receive from redis... ' |
|
|
self._get_logger().error('Cannot receive from redis... ' |
|
|
'retrying in ' |
|
|
'retrying in ' |
|
|
'{} secs'.format(retry_sleep)) |
|
|
'{} secs'.format(retry_sleep), |
|
|
|
|
|
extra={"redis_exception": str(exc)}) |
|
|
connect = True |
|
|
connect = True |
|
|
await asyncio.sleep(retry_sleep) |
|
|
await asyncio.sleep(retry_sleep) |
|
|
retry_sleep *= 2 |
|
|
retry_sleep *= 2 |
|
|