Repository: redis/redis-py
Pull request: Replace OSError exceptions from `can_read` with `redis.ConnectionError`
### Pull Request check-list

_Please make sure to review and check all of these items:_

- [ ] Does `$ tox` pass with this change (including linting)?
- [ ] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [ ] Is the new or changed code fully tested?
- [ ] Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
- [ ] Is there an example added to the examples folder (if applicable)?
- [ ] Was the change added to CHANGES file?

_NOTE: these things are not required to open a PR and can be done
afterwards / while the PR is open._

### Description of change

This change catches OS errors in `Connection.can_read`, closes connection and raises a `redis.ConnectionError`.
Fixes #2138 


Problem statement:
`connection.can_read()` raises OSError when connection is broken

**Version**: 4.2.2

**Platform**: Windows 11
**Description**: 
`redis.asyncio.connection.Connection()`, the class provides a `can_read` method, which is used, for example, in `Client.parse_response`.

However, unlike `Connection.read_response()` there is no error handling present to turn any OSErrors into `asyncio.ConnectionError`.
And so, the OSError bubbles up and causes problems, where we expect only `asyncio.ConnectionError` to occur.
This plays havoc with retry-mechanisms and automatic reconnect.

The issue is evident when the connection to the redis server is interrupted during a blocking call to `pubsub.run()` where a `get_message` call is made with a positive timeout.

Hints from issue discussion:

Commit message: Replace OSError exceptions from `can_read` with `redis.ConnectionError`

Patch to review:
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index f0e6d3da4f..7b1463be4d 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -911,7 +911,13 @@ async def can_read(self, timeout: float = 0):
         """Poll the socket to see if there's data that can be read."""
         if not self.is_connected:
             await self.connect()
-        return await self._parser.can_read(timeout)
+        try:
+            return await self._parser.can_read(timeout)
+        except OSError as e:
+            await self.disconnect()
+            raise ConnectionError(
+                f"Error while reading from {self.host}:{self.port} : {e.args}"
+            )
 
     async def read_response(self, disable_decoding: bool = False):
         """Read the response from a previously sent command"""
diff --git a/redis/connection.py b/redis/connection.py
index 0f97361e6e..b9ad14ea41 100755
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -808,7 +808,11 @@ def can_read(self, timeout=0):
         sock = self._sock
         if not sock:
             self.connect()
-        return self._parser.can_read(timeout)
+        try:
+            return self._parser.can_read(timeout)
+        except OSError as e:
+            self.disconnect()
+            raise ConnectionError(f"Error while reading from {self.host}:{self.port} : {e.args}")
 
     def read_response(self, disable_decoding=False):
         """Read the response from a previously sent command"""
@@ -1282,7 +1286,7 @@ class initializer. In the case of conflicting arguments, querystring
     def __init__(
         self, connection_class=Connection, max_connections=None, **connection_kwargs
     ):
-        max_connections = max_connections or 2 ** 31
+        max_connections = max_connections or 2**31
         if not isinstance(max_connections, int) or max_connections < 0:
             raise ValueError('"max_connections" must be a positive integer')
 


Code review comments:
diff_hunk: @@ -911,7 +911,13 @@ async def can_read(self, timeout: float = 0):
         """Poll the socket to see if there's data that can be read."""
         if not self.is_connected:
             await self.connect()
-        return await self._parser.can_read(timeout)
+        try:
+            return await self._parser.can_read(timeout)
+        except OSError as e:
+            await self.disconnect()
+            raise ConnectionError(
+                f"Error while reading from {self.host}:{self.port} : {e.args}"
path: redis/asyncio/connection.py
text: @user1:
Nice. Small nit to remove a space.
```
                f"Error while reading from {self.host}:{self.port}: {e.args}"
```

diff_hunk: @@ -808,7 +808,11 @@ def can_read(self, timeout=0):
         sock = self._sock
         if not sock:
             self.connect()
-        return self._parser.can_read(timeout)
+        try:
+            return self._parser.can_read(timeout)
+        except OSError as e:
+            self.disconnect()
+            raise ConnectionError(f"Error while reading from {self.host}:{self.port} : {e.args}")
path: redis/connection.py
text: @user1:
Same
```
                f"Error while reading from {self.host}:{self.port}: {e.args}"
```