make can_read() destructive for simplicity, and rename the method.

Remove timeout argument, always timeout immediately.

Author: kristjanvalur

redis/asyncio/connection.py

@@ -197,7 +197,7 @@ def on_disconnect(self):
     def on_connect(self, connection: "Connection"):
         raise NotImplementedError()
 
-    async def can_read(self, timeout: float) -> bool:
+    async def can_read_destructive(self) -> bool:
         raise NotImplementedError()
 
     async def read_response(
@@ -275,9 +275,9 @@ async def _read_from_socket(
                 return False
             raise ConnectionError(f"Error while reading from socket: {ex.args}")
 
-    async def can_read(self, timeout: float) -> bool:
+    async def can_read_destructive(self) -> bool:
         return bool(self.length) or await self._read_from_socket(
-            timeout=timeout, raise_on_timeout=False
+            timeout=0, raise_on_timeout=False
         )
 
     async def read(self, length: int) -> bytes:
@@ -375,8 +375,8 @@ def on_disconnect(self):
             self._buffer = None
         self.encoder = None
 
-    async def can_read(self, timeout: float):
-        return self._buffer and bool(await self._buffer.can_read(timeout))
+    async def can_read_destructive(self):
+        return self._buffer and bool(await self._buffer.can_read_destructive())
 
     async def read_response(
         self, disable_decoding: bool = False
@@ -433,9 +433,7 @@ async def read_response(
 class HiredisParser(BaseParser):
     """Parser class for connections using Hiredis"""
 
-    __slots__ = BaseParser.__slots__ + ("_next_response", "_reader", "_socket_timeout")
-
-    _next_response: bool
+    __slots__ = BaseParser.__slots__ + ("_reader", "_socket_timeout")
 
     def __init__(self, socket_read_size: int):
         if not HIREDIS_AVAILABLE:
@@ -455,23 +453,18 @@ def on_connect(self, connection: "Connection"):
             kwargs["errors"] = connection.encoder.encoding_errors
 
         self._reader = hiredis.Reader(**kwargs)
-        self._next_response = False
         self._socket_timeout = connection.socket_timeout
 
     def on_disconnect(self):
         self._stream = None
         self._reader = None
-        self._next_response = False
 
-    async def can_read(self, timeout: float):
+    async def can_read_destructive(self):
         if not self._stream or not self._reader:
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
-
-        if self._next_response is False:
-            self._next_response = self._reader.gets()
-        if self._next_response is False:
-            return await self.read_from_socket(timeout=timeout, raise_on_timeout=False)
-        return True
+        if self._reader.gets():
+            return True
+        return await self.read_from_socket(timeout=0, raise_on_timeout=False)
 
     async def read_from_socket(
         self,
@@ -514,12 +507,6 @@ async def read_response(
             self.on_disconnect()
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from None
 
-        # _next_response might be cached from a can_read() call
-        if self._next_response is not False:
-            response = self._next_response
-            self._next_response = False
-            return response
-
         response = self._reader.gets()
         while response is False:
             await self.read_from_socket()
@@ -916,12 +903,10 @@ async def send_command(self, *args: Any, **kwargs: Any) -> None:
             self.pack_command(*args), check_health=kwargs.get("check_health", True)
         )
 
-    async def can_read(self, timeout: float = 0):
+    async def can_read_destructive(self):
         """Poll the socket to see if there's data that can be read."""
-        if not self.is_connected:
-            await self.connect()
         try:
-            return await self._parser.can_read(timeout)
+            return await self._parser.can_read_destructive()
         except OSError as e:
             await self.disconnect()
             raise ConnectionError(
@@ -1523,12 +1508,12 @@ async def get_connection(self, command_name, *keys, **options):
             # pool before all data has been read or the socket has been
             # closed. either way, reconnect and verify everything is good.
             try:
-                if await connection.can_read():
+                if await connection.can_read_destructive():
                     raise ConnectionError("Connection has data") from None
             except ConnectionError:
                 await connection.disconnect()
                 await connection.connect()
-                if await connection.can_read():
+                if await connection.can_read_destructive():
                     raise ConnectionError("Connection not ready") from None
         except BaseException:
             # release the connection back to the pool so that we don't
@@ -1724,12 +1709,12 @@ async def get_connection(self, command_name, *keys, **options):
             # pool before all data has been read or the socket has been
             # closed. either way, reconnect and verify everything is good.
             try:
-                if await connection.can_read():
+                if await connection.can_read_destructive():
                     raise ConnectionError("Connection has data") from None
             except ConnectionError:
                 await connection.disconnect()
                 await connection.connect()
-                if await connection.can_read():
+                if await connection.can_read_destructive():
                     raise ConnectionError("Connection not ready") from None
         except BaseException:
             # release the connection back to the pool so that we don't leak it

tests/test_asyncio/test_connection_pool.py

@@ -103,7 +103,7 @@ async def connect(self):
     async def disconnect(self):
         pass
 
-    async def can_read(self, timeout: float = 0):
+    async def can_read_destructive(self, timeout: float = 0):
         return False
 
 

tests/test_asyncio/test_pubsub.py

@@ -842,7 +842,7 @@ async def test_reconnect_socket_error(self, r: redis.Redis, method):
                 self.state = 1
                 with mock.patch.object(self.pubsub.connection, "_parser") as mockobj:
                     mockobj.read_response.side_effect = socket.error
-                    mockobj.can_read.side_effect = socket.error
+                    mockobj.can_read_destructive.side_effect = socket.error
                     # wait until task noticies the disconnect until we undo the patch
                     await self.cond.wait_for(lambda: self.state >= 2)
                     assert not self.pubsub.connection.is_connected