Perform connect() call in PubSub code rather than read_response.

Author: kristjanvalur

redis/asyncio/client.py

@@ -754,9 +754,15 @@ async def parse_response(self, block: bool = True, timeout: float = 0):
 
         await self.check_health()
 
-        if not block and not await self._execute(conn, conn.can_read, timeout=timeout):
-            return None
-        response = await self._execute(conn, conn.read_response)
+        async def try_read():
+            if not block:
+                if not await conn.can_read(timeout=timeout):
+                    return None
+            else:
+                await conn.connect()
+            return await conn.read_response()
+
+        response = await self._execute(conn, try_read)
 
         if conn.health_check_interval and response == self.health_check_response:
             # ignore the health check message as user might not expect it

redis/asyncio/connection.py

@@ -930,8 +930,6 @@ async def can_read(self, timeout: float = 0):
 
     async def read_response(self, disable_decoding: bool = False):
         """Read the response from a previously sent command"""
-        if not self.is_connected:
-            await self.connect()
         try:
             async with self._lock:
                 if self.socket_timeout:

redis/client.py

@@ -1497,9 +1497,15 @@ def parse_response(self, block=True, timeout=0):
 
         self.check_health()
 
-        if not block and not self._execute(conn, conn.can_read, timeout=timeout):
-            return None
-        response = self._execute(conn, conn.read_response)
+        def try_read():
+            if not block:
+                if not conn.can_read(timeout=timeout):
+                    return None
+            else:
+                conn.connect()
+            return conn.read_response()
+
+        response = self._execute(conn, try_read)
 
         if self.is_health_check_response(response):
             # ignore the health check message as user might not expect it

redis/connection.py

@@ -815,8 +815,6 @@ def can_read(self, timeout=0):
 
     def read_response(self, disable_decoding=False):
         """Read the response from a previously sent command"""
-        if not self._sock:
-            self.connect()
         try:
             hosterr = f"{self.host}:{self.port}"
         except AttributeError: