Use retry mechanism in async version of Connection objects (#2271)

2 files changed, 12 insertions, 2 deletions

diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index 35536fc..5bf2a0f 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -637,6 +637,8 @@ class Connection:
             retry_on_error = []
         if retry_on_timeout:
             retry_on_error.append(TimeoutError)
+            retry_on_error.append(socket.timeout)
+            retry_on_error.append(asyncio.TimeoutError)
         self.retry_on_error = retry_on_error
         if retry_on_error:
             if not retry:
@@ -706,7 +708,9 @@ class Connection:
         if self.is_connected:
             return
         try:
-            await self._connect()
+            await self.retry.call_with_retry(
+                lambda: self._connect(), lambda error: self.disconnect()
+            )
         except asyncio.CancelledError:
             raise
         except (socket.timeout, asyncio.TimeoutError):
diff --git a/redis/asyncio/sentinel.py b/redis/asyncio/sentinel.py
index 5aefd09..99c5074 100644
--- a/redis/asyncio/sentinel.py
+++ b/redis/asyncio/sentinel.py
@@ -44,7 +44,7 @@ class SentinelManagedConnection(Connection):
             if str_if_bytes(await self.read_response()) != "PONG":
                 raise ConnectionError("PING failed")
 
-    async def connect(self):
+    async def _connect_retry(self):
         if self._reader:
             return  # already connected
         if self.connection_pool.is_master:
@@ -57,6 +57,12 @@ class SentinelManagedConnection(Connection):
                     continue
             raise SlaveNotFoundError  # Never be here
 
+    async def connect(self):
+        return await self.retry.call_with_retry(
+            self._connect_retry,
+            lambda error: asyncio.sleep(0),
+        )
+
     async def read_response(self, disable_decoding: bool = False):
         try:
             return await super().read_response(disable_decoding=disable_decoding)