Repository: redis/redis-py
Pull request #1825 (2021-12-22 09:46:32)

Support for SELECT
### Pull Request check-list

_Please make sure to review and check all of these items:_

- [x] Does `$ tox` pass with this change (including linting)?
- [x] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [x] Is the new or changed code fully tested?
- [x] 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)?

_NOTE: these things are not required to open a PR and can be done
afterwards / while the PR is open._

### Description of change

_Please provide a description of the change here._



Patch:
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 04f989a3..0f9ca0b1 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -32,7 +32,7 @@ Here's how to get started with your code contribution:
 
 1.  Create your own fork of redis-py
 2.  Do the changes in your fork
-3.  
+3.
     *Create a virtualenv and install the development dependencies from the dev_requirements.txt file:*
 
         a.  python -m venv .venv
@@ -69,12 +69,12 @@ configuration](https://redis.io/topics/sentinel).
 
 ## Testing
 
-Call `invoke tests` to run all tests, or `invoke all-tests` to run linters 
-tests as well. With the 'tests' and 'all-tests' targets, all Redis and 
-RedisCluster tests will be run. 
+Call `invoke tests` to run all tests, or `invoke all-tests` to run linters
+tests as well. With the 'tests' and 'all-tests' targets, all Redis and
+RedisCluster tests will be run.
 
-It is possible to run only Redis client tests (with cluster mode disabled) by 
-using `invoke standalone-tests`; similarly, RedisCluster tests can be run by using 
+It is possible to run only Redis client tests (with cluster mode disabled) by
+using `invoke standalone-tests`; similarly, RedisCluster tests can be run by using
 `invoke cluster-tests`.
 
 Each run of tox starts and stops the various dockers required. Sometimes
@@ -85,9 +85,7 @@ tests against multiple versions of python. Feel free to test your
 changes against all the python versions supported, as declared by the
 tox.ini file (eg: tox -e py39). If you have the various python versions
 on your desktop, you can run *tox* by itself, to test all supported
-versions. Alternatively, as your source code is mounted in the
-**lots-of-pythons** docker, you can start exploring from there, with all
-supported python versions!
+versions.
 
 ### Docker Tips
 
diff --git a/redis/client.py b/redis/client.py
index ae4fae2a..5116482c 100755
--- a/redis/client.py
+++ b/redis/client.py
@@ -768,6 +768,7 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
         "STRALGO": parse_stralgo,
         "PUBSUB NUMSUB": parse_pubsub_numsub,
         "RANDOMKEY": lambda r: r and r or None,
+        "RESET": str_if_bytes,
         "SCAN": parse_scan,
         "SCRIPT EXISTS": lambda r: list(map(bool, r)),
         "SCRIPT FLUSH": bool_ok,
@@ -868,6 +869,7 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
         errors=None,
         decode_responses=False,
         retry_on_timeout=False,
+        retry_on_error=[],
         ssl=False,
         ssl_keyfile=None,
         ssl_certfile=None,
@@ -886,8 +888,10 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
     ):
         """
         Initialize a new Redis client.
-        To specify a retry policy, first set `retry_on_timeout` to `True`
-        then set `retry` to a valid `Retry` object
+        To specify a retry policy for specific errors, first set
+        `retry_on_error` to a list of the error/s to retry on, then set
+        `retry` to a valid `Retry` object.
+        To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
         if not connection_pool:
             if charset is not None:
@@ -904,7 +908,8 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
                     )
                 )
                 encoding_errors = errors
-
+            if retry_on_timeout is True:
+                retry_on_error.append(TimeoutError)
             kwargs = {
                 "db": db,
                 "username": username,
@@ -913,7 +918,7 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
                 "encoding": encoding,
                 "encoding_errors": encoding_errors,
                 "decode_responses": decode_responses,
-                "retry_on_timeout": retry_on_timeout,
+                "retry_on_error": retry_on_error,
                 "retry": copy.deepcopy(retry),
                 "max_connections": max_connections,
                 "health_check_interval": health_check_interval,
@@ -1145,11 +1150,14 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
     def _disconnect_raise(self, conn, error):
         """
         Close the connection and raise an exception
-        if retry_on_timeout is not set or the error
-        is not a TimeoutError
+        if retry_on_error is not set or the error
+        is not one of the specified error types
         """
         conn.disconnect()
-        if not (conn.retry_on_timeout and isinstance(error, TimeoutError)):
+        if (
+            conn.retry_on_error is None
+            or isinstance(error, tuple(conn.retry_on_error)) is False
+        ):
             raise error
 
     # COMMAND EXECUTION AND PROTOCOL PARSING
@@ -1280,18 +1288,17 @@ class PubSub:
         self.shard_hint = shard_hint
         self.ignore_subscribe_messages = ignore_subscribe_messages
         self.connection = None
+        self.subscribed_event = threading.Event()
         # we need to know the encoding options for this connection in order
         # to lookup channel and pattern names for callback handlers.
         self.encoder = encoder
         if self.encoder is None:
             self.encoder = self.connection_pool.get_encoder()
+        self.health_check_response_b = self.encoder.encode(self.HEALTH_CHECK_MESSAGE)
         if self.encoder.decode_responses:
             self.health_check_response = ["pong", self.HEALTH_CHECK_MESSAGE]
         else:
-            self.health_check_response = [
-                b"pong",
-                self.encoder.encode(self.HEALTH_CHECK_MESSAGE),
-            ]
+            self.health_check_response = [b"pong", self.health_check_response_b]
         self.reset()
 
     def __enter__(self):
@@ -1316,9 +1323,11 @@ class PubSub:
             self.connection_pool.release(self.connection)
             self.connection = None
         self.channels = {}
+        self.health_check_response_counter = 0
         self.pending_unsubscribe_channels = set()
         self.patterns = {}
         self.pending_unsubscribe_patterns = set()
+        self.subscribed_event.clear()
 
     def close(self):
         self.reset()
@@ -1344,7 +1353,7 @@ class PubSub:
     @property
     def subscribed(self):
         "Indicates if there are subscriptions to any channels or patterns"
-        return bool(self.channels or self.patterns)
+        return self.subscribed_event.is_set()
 
     def execute_command(self, *args):
         "Execute a publish/subscribe command"
@@ -1362,8 +1371,28 @@ class PubSub:
             self.connection.register_connect_callback(self.on_connect)
         connection = self.connection
         kwargs = {"check_health": not self.subscribed}
+        if not self.subscribed:
+            self.clean_health_check_responses()
         self._execute(connection, connection.send_command, *args, **kwargs)
 
+    def clean_health_check_responses(self):
+        """
+        If any health check responses are present, clean them
+        """
+        ttl = 10
+        conn = self.connection
+        while self.health_check_response_counter > 0 and ttl > 0:
+            if self._execute(conn, conn.can_read, timeout=conn.socket_timeout):
+                response = self._execute(conn, conn.read_response)
+                if self.is_health_check_response(response):
+                    self.health_check_response_counter -= 1
+                else:
+                    raise PubSubError(
+                        "A non health check response was cleaned by "
+                        "execute_command: {0}".format(response)
+                    )
+            ttl -= 1
+
     def _disconnect_raise_connect(self, conn, error):
         """
         Close the connection and raise an exception
@@ -1403,11 +1432,23 @@ class PubSub:
             return None
         response = self._execute(conn, conn.read_response)
 
-        if conn.health_check_interval and response == self.health_check_response:
+        if self.is_health_check_response(response):
             # ignore the health check message as user might not expect it
+            self.health_check_response_counter -= 1
             return None
         return response
 
+    def is_health_check_response(self, response):
+        """
+        Check if the response is a health check response.
+        If there are no subscriptions redis responds to PING command with a
+        bulk response, instead of a multi-bulk with "pong" and the response.
+        """
+        return response in [
+            self.health_check_response,  # If there was a subscription
+            self.health_check_response_b,  # If there wasn't
+        ]
+
     def check_health(self):
         conn = self.connection
         if conn is None:
@@ -1418,6 +1459,7 @@ class PubSub:
 
         if conn.health_check_interval and time.time() > conn.next_health_check:
             conn.send_command("PING", self.HEALTH_CHECK_MESSAGE, check_health=False)
+            self.health_check_response_counter += 1
 
     def _normalize_keys(self, data):
         """
@@ -1447,6 +1489,11 @@ class PubSub:
         # for the reconnection.
         new_patterns = self._normalize_keys(new_patterns)
         self.patterns.update(new_patterns)
+        if not self.subscribed:
+            # Set the subscribed_event flag to True
+            self.subscribed_event.set()
+            # Clear the health check counter
+            self.health_check_response_counter = 0
         self.pending_unsubscribe_patterns.difference_update(new_patterns)
         return ret_val
 
@@ -1481,6 +1528,11 @@ class PubSub:
         # for the reconnection.
         new_channels = self._normalize_keys(new_channels)
         self.channels.update(new_channels)
+        if not self.subscribed:
+            # Set the subscribed_event flag to True
+            self.subscribed_event.set()
+            # Clear the health check counter
+            self.health_check_response_counter = 0
         self.pending_unsubscribe_channels.difference_update(new_channels)
         return ret_val
 
@@ -1512,6 +1564,20 @@ class PubSub:
         before returning. Timeout should be specified as a floating point
         number.
         """
+        if not self.subscribed:
+            # Wait for subscription
+            start_time = time.time()
+            if self.subscribed_event.wait(timeout) is True:
+                # The connection was subscribed during the timeout time frame.
+                # The timeout should be adjusted based on the time spent
+                # waiting for the subscription
+                time_spent = time.time() - start_time
+                timeout = max(0.0, timeout - time_spent)
+            else:
+                # The connection isn't subscribed to any channels or patterns,
+                # so no messages are available
+                return None
+
         response = self.parse_response(block=False, timeout=timeout)
         if response:
             return self.handle_message(response, ignore_subscribe_messages)
@@ -1530,6 +1596,8 @@ class PubSub:
         with a message handler, the handler is invoked instead of a parsed
         message being returned.
         """
+        if response is None:
+            return None
         message_type = str_if_bytes(response[0])
         if message_type == "pmessage":
             message = {
@@ -1565,6 +1633,10 @@ class PubSub:
                 if channel in self.pending_unsubscribe_channels:
                     self.pending_unsubscribe_channels.remove(channel)
                     self.channels.pop(channel, None)
+            if not self.channels and not self.patterns:
+                # There are no subscriptions anymore, set subscribed_event flag
+                # to false
+                self.subscribed_event.clear()
 
         if message_type in self.PUBLISH_MESSAGE_TYPES:
             # if there's a message handler, invoke it
diff --git a/redis/cluster.py b/redis/cluster.py
index 0c2fc715..5707a9da 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -17,6 +17,7 @@ from redis.exceptions import (
     ClusterCrossSlotError,
     ClusterDownError,
     ClusterError,
+    ConnectionError,
     DataError,
     MasterDownError,
     MovedError,
@@ -374,6 +375,12 @@ class RedisCluster(RedisClusterCommands):
         ),
     )
 
+    ERRORS_ALLOW_RETRY = (
+        ConnectionError,
+        TimeoutError,
+        ClusterDownError,
+    )
+
     def __init__(
         self,
         host=None,
@@ -385,8 +392,6 @@ class RedisCluster(RedisClusterCommands):
         reinitialize_steps=10,
         read_from_replicas=False,
         url=None,
-        retry_on_timeout=False,
-        retry=None,
         **kwargs,
     ):
         """
@@ -417,11 +422,6 @@ class RedisCluster(RedisClusterCommands):
         :cluster_error_retry_attempts: 'int'
              Retry command execution attempts when encountering ClusterDownError
              or ConnectionError
-        :retry_on_timeout: 'bool'
-             To specify a retry policy, first set `retry_on_timeout` to `True`
-             then set `retry` to a valid `Retry` object
-        :retry: 'Retry'
-              a `Retry` object
         :reinitialize_steps: 'int'
             Specifies the number of MOVED errors that need to occur before
             reinitializing the whole cluster topology. If a MOVED error occurs
@@ -452,9 +452,6 @@ class RedisCluster(RedisClusterCommands):
                 "Argument 'db' is not possible to use in cluster mode"
             )
 
-        if retry_on_timeout:
-            kwargs.update({"retry_on_timeout": retry_on_timeout, "retry": retry})
-
         # Get the startup node/s
         from_url = False
         if url is not None:
@@ -850,7 +847,7 @@ class RedisCluster(RedisClusterCommands):
 
     def execute_command(self, *args, **kwargs):
         """
-        Wrapper for ClusterDownError and ConnectionError error handling.
+        Wrapper for ERRORS_ALLOW_RETRY error handling.
 
         It will try the number of times specified by the config option
         "self.cluster_error_retry_attempts" which defaults to 3 unless manually
@@ -865,18 +862,19 @@ class RedisCluster(RedisClusterCommands):
             dict<Any, ClusterNode>
         """
         target_nodes_specified = False
-        target_nodes = kwargs.pop("target_nodes", None)
-        if target_nodes is not None and not self._is_nodes_flag(target_nodes):
-            target_nodes = self._parse_target_nodes(target_nodes)
+        target_nodes = None
+        passed_targets = kwargs.pop("target_nodes", None)
+        if passed_targets is not None and not self._is_nodes_flag(passed_targets):
+            target_nodes = self._parse_target_nodes(passed_targets)
             target_nodes_specified = True
-        # If ClusterDownError/ConnectionError were thrown, the nodes
-        # and slots cache were reinitialized. We will retry executing the
-        # command with the updated cluster setup only when the target nodes
-        # can be determined again with the new cache tables. Therefore,
-        # when target nodes were passed to this function, we cannot retry
-        # the command execution since the nodes may not be valid anymore
-        # after the tables were reinitialized. So in case of passed target
-        # nodes, retry_attempts will be set to 1.
+        # If an error that allows retrying was thrown, the nodes and slots
+        # cache were reinitialized. We will retry executing the command with
+        # the updated cluster setup only when the target nodes can be
+        # determined again with the new cache tables. Therefore, when target
+        # nodes were passed to this function, we cannot retry the command
+        # execution since the nodes may not be valid anymore after the tables
+        # were reinitialized. So in case of passed target nodes,
+        # retry_attempts will be set to 1.
         retry_attempts = (
             1 if target_nodes_specified else self.cluster_error_retry_attempts
         )
@@ -887,7 +885,7 @@ class RedisCluster(RedisClusterCommands):
                 if not target_nodes_specified:
                     # Determine the nodes to execute the command on
                     target_nodes = self._determine_nodes(
-                        *args, **kwargs, nodes_flag=target_nodes
+                        *args, **kwargs, nodes_flag=passed_targets
                     )
                     if not target_nodes:
                         raise RedisClusterException(
@@ -897,11 +895,14 @@ class RedisCluster(RedisClusterCommands):
                     res[node.name] = self._execute_command(node, *args, **kwargs)
                 # Return the processed result
                 return self._process_result(args[0], res, **kwargs)
-            except (ClusterDownError, ConnectionError) as e:
-                # The nodes and slots cache were reinitialized.
-                # Try again with the new cluster setup. All other errors
-                # should be raised.
-                exception = e
+            except BaseException as e:
+                if type(e) in RedisCluster.ERRORS_ALLOW_RETRY:
+                    # The nodes and slots cache were reinitialized.
+                    # Try again with the new cluster setup.
+                    exception = e
+                else:
+                    # All other errors should be raised.
+                    raise e
 
         # If it fails the configured number of times then raise exception back
         # to caller of this method
@@ -953,11 +954,11 @@ class RedisCluster(RedisClusterCommands):
                     )
                 return response
 
-            except (RedisClusterException, BusyLoadingError):
-                log.exception("RedisClusterException || BusyLoadingError")
+            except (RedisClusterException, BusyLoadingError) as e:
+                log.exception(type(e))
                 raise
-            except ConnectionError:
-                log.exception("ConnectionError")
+            except (ConnectionError, TimeoutError) as e:
+                log.exception(type(e))
                 # ConnectionError can also be raised if we couldn't get a
                 # connection from the pool before timing out, so check that
                 # this is an actual connection before attempting to disconnect.
@@ -976,13 +977,6 @@ class RedisCluster(RedisClusterCommands):
                     # and try again with the new setup
                     self.nodes_manager.initialize()
                     raise
-            except TimeoutError:
-                log.exception("TimeoutError")
-                if connection is not None:
-                    connection.disconnect()
-
-                if ttl < self.RedisClusterRequestTTL / 2:
-                    time.sleep(0.05)
             except MovedError as e:
                 # First, we will try to patch the slots/nodes cache with the
                 # redirected node output and try again. If MovedError exceeds
@@ -1016,7 +1010,7 @@ class RedisCluster(RedisClusterCommands):
                 # ClusterDownError can occur during a failover and to get
                 # self-healed, we will try to reinitialize the cluster layout
                 # and retry executing the command
-                time.sleep(0.05)
+                time.sleep(0.25)
                 self.nodes_manager.initialize()
                 raise e
             except ResponseError as e:
@@ -1342,7 +1336,7 @@ class NodesManager:
                     raise RedisClusterException(
                         "Cluster mode is not enabled on this node"
                     )
-                cluster_slots = r.execute_command("CLUSTER SLOTS")
+                cluster_slots = str_if_bytes(r.execute_command("CLUSTER SLOTS"))
                 startup_nodes_reachable = True
             except (ConnectionError, TimeoutError) as e:
                 msg = e.__str__
@@ -1631,21 +1625,20 @@ class ClusterPubSub(PubSub):
             return self.node.redis_connection
 
 
-ERRORS_ALLOW_RETRY = (
-    ConnectionError,
-    TimeoutError,
-    MovedError,
-    AskError,
-    TryAgainError,
-)
-
-
 class ClusterPipeline(RedisCluster):
     """
     Support for Redis pipeline
     in cluster mode
     """
 
+    ERRORS_ALLOW_RETRY = (
+        ConnectionError,
+        TimeoutError,
+        MovedError,
+        AskError,
+        TryAgainError,
+    )
+
     def __init__(
         self,
         nodes_manager,
@@ -1653,7 +1646,7 @@ class ClusterPipeline(RedisCluster):
         cluster_response_callbacks=None,
         startup_nodes=None,
         read_from_replicas=False,
-        cluster_error_retry_attempts=3,
+        cluster_error_retry_attempts=5,
         reinitialize_steps=10,
         **kwargs,
     ):
@@ -1915,7 +1908,11 @@ class ClusterPipeline(RedisCluster):
         # collect all the commands we are allowed to retry.
         # (MOVED, ASK, or connection errors or timeout errors)
         attempt = sorted(
-            (c for c in attempt if isinstance(c.result, ERRORS_ALLOW_RETRY)),
+            (
+                c
+                for c in attempt
+                if isinstance(c.result, ClusterPipeline.ERRORS_ALLOW_RETRY)
+            ),
             key=lambda x: x.position,
         )
         if attempt and allow_redirections:
diff --git a/redis/commands/core.py b/redis/commands/core.py
index 835ea612..4f0accd9 100644
--- a/redis/commands/core.py
+++ b/redis/commands/core.py
@@ -479,6 +479,104 @@ class ManagementCommands:
         """
         return self.execute_command("CLIENT ID", **kwargs)
 
+    def client_tracking_on(
+        self,
+        clientid=None,
+        prefix=[],
+        bcast=False,
+        optin=False,
+        optout=False,
+        noloop=False,
+    ):
+        """
+        Turn on the tracking mode.
+        For more information about the options look at client_tracking func.
+
+        See https://redis.io/commands/client-tracking
+        """
+        return self.client_tracking(
+            True, clientid, prefix, bcast, optin, optout, noloop
+        )
+
+    def client_tracking_off(
+        self,
+        clientid=None,
+        prefix=[],
+        bcast=False,
+        optin=False,
+        optout=False,
+        noloop=False,
+    ):
+        """
+        Turn off the tracking mode.
+        For more information about the options look at client_tracking func.
+
+        See https://redis.io/commands/client-tracking
+        """
+        return self.client_tracking(
+            False, clientid, prefix, bcast, optin, optout, noloop
+        )
+
+    def client_tracking(
+        self,
+        on=True,
+        clientid=None,
+        prefix=[],
+        bcast=False,
+        optin=False,
+        optout=False,
+        noloop=False,
+        **kwargs,
+    ):
+        """
+        Enables the tracking feature of the Redis server, that is used
+        for server assisted client side caching.
+
+        ``on`` indicate for tracking on or tracking off. The dafualt is on.
+
+        ``clientid`` send invalidation messages to the connection with
+        the specified ID.
+
+        ``bcast`` enable tracking in broadcasting mode. In this mode
+        invalidation messages are reported for all the prefixes
+        specified, regardless of the keys requested by the connection.
+
+        ``optin``  when broadcasting is NOT active, normally don't track
+        keys in read only commands, unless they are called immediately
+        after a CLIENT CACHING yes command.
+
+        ``optout`` when broadcasting is NOT active, normally track keys in
+        read only commands, unless they are called immediately after a
+        CLIENT CACHING no command.
+
+        ``noloop`` don't send notifications about keys modified by this
+        connection itself.
+
+        ``prefix``  for broadcasting, register a given key prefix, so that
+        notifications will be provided only for keys starting with this string.
+
+        See https://redis.io/commands/client-tracking
+        """
+
+        if len(prefix) != 0 and bcast is False:
+            raise DataError("Prefix can only be used with bcast")
+
+        pieces = ["ON"] if on else ["OFF"]
+        if clientid is not None:
+            pieces.extend(["REDIRECT", clientid])
+        for p in prefix:
+            pieces.extend(["PREFIX", p])
+        if bcast:
+            pieces.append("BCAST")
+        if optin:
+            pieces.append("OPTIN")
+        if optout:
+            pieces.append("OPTOUT")
+        if noloop:
+            pieces.append("NOLOOP")
+
+        return self.execute_command("CLIENT TRACKING", *pieces)
+
     def client_trackinginfo(self, **kwargs):
         """
         Returns the information about the current client connection's
@@ -510,16 +608,28 @@ class ManagementCommands:
             args.append(b"ERROR")
         return self.execute_command(*args, **kwargs)
 
-    def client_pause(self, timeout, **kwargs):
+    def client_pause(self, timeout, all=True, **kwargs):
         """
         Suspend all the Redis clients for the specified amount of time
         :param timeout: milliseconds to pause clients
 
         For more information check https://redis.io/commands/client-pause
-        """
+        :param all: If true (default) all client commands are blocked.
+             otherwise, clients are only blocked if they attempt to execute
+             a write command.
+             For the WRITE mode, some commands have special behavior:
+                 EVAL/EVALSHA: Will block client for all scripts.
+                 PUBLISH: Will block client.
+                 PFCOUNT: Will block client.
+                 WAIT: Acknowledgments will be delayed, so this command will
+                 appear blocked.
+        """
+        args = ["CLIENT PAUSE", str(timeout)]
         if not isinstance(timeout, int):
             raise DataError("CLIENT PAUSE timeout must be an integer")
-        return self.execute_command("CLIENT PAUSE", str(timeout), **kwargs)
+        if not all:
+            args.append("WRITE")
+        return self.execute_command(*args, **kwargs)
 
     def client_unpause(self, **kwargs):
         """
@@ -670,6 +780,13 @@ class ManagementCommands:
         """
         return self.execute_command("SWAPDB", first, second, **kwargs)
 
+    def select(self, index, **kwargs):
+        """Select the Redis logical database at index.
+
+        See: https://redis.io/commands/select
+        """
+        return self.execute_command("SELECT", index, **kwargs)
+
     def info(self, section=None, **kwargs):
         """
         Returns a dictionary containing information about the Redis server
@@ -707,6 +824,13 @@ class ManagementCommands:
         else:
             return self.execute_command("LOLWUT", **kwargs)
 
+    def reset(self):
+        """Perform a full reset on the connection's server side contenxt.
+
+        See: https://redis.io/commands/reset
+        """
+        return self.execute_command("RESET")
+
     def migrate(
         self,
         host,
diff --git a/redis/connection.py b/redis/connection.py
index 3fe8543c..a349a0f2 100755
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -513,6 +513,7 @@ class Connection:
         socket_keepalive_options=None,
         socket_type=0,
         retry_on_timeout=False,
+        retry_on_error=[],
         encoding="utf-8",
         encoding_errors="strict",
         decode_responses=False,
@@ -526,8 +527,10 @@ class Connection:
     ):
         """
         Initialize a new Connection.
-        To specify a retry policy, first set `retry_on_timeout` to `True`
-        then set `retry` to a valid `Retry` object
+        To specify a retry policy for specific errors, first set
+        `retry_on_error` to a list of the error/s to retry on, then set
+        `retry` to a valid `Retry` object.
+        To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
         self.pid = os.getpid()
         self.host = host
@@ -543,11 +546,17 @@ class Connection:
         self.socket_type = socket_type
         self.retry_on_timeout = retry_on_timeout
         if retry_on_timeout:
+            # Add TimeoutError to the errors list to retry on
+            retry_on_error.append(TimeoutError)
+        self.retry_on_error = retry_on_error
+        if retry_on_error:
             if retry is None:
                 self.retry = Retry(NoBackoff(), 1)
             else:
                 # deep-copy the Retry object as it is mutable
                 self.retry = copy.deepcopy(retry)
+            # Update the retry's supported errors with the specified errors
+            self.retry.update_supported_erros(retry_on_error)
         else:
             self.retry = Retry(NoBackoff(), 0)
         self.health_check_interval = health_check_interval
@@ -969,6 +978,7 @@ class UnixDomainSocketConnection(Connection):
         encoding_errors="strict",
         decode_responses=False,
         retry_on_timeout=False,
+        retry_on_error=[],
         parser_class=DefaultParser,
         socket_read_size=65536,
         health_check_interval=0,
@@ -978,8 +988,10 @@ class UnixDomainSocketConnection(Connection):
     ):
         """
         Initialize a new UnixDomainSocketConnection.
-        To specify a retry policy, first set `retry_on_timeout` to `True`
-        then set `retry` to a valid `Retry` object
+        To specify a retry policy for specific errors, first set
+        `retry_on_error` to a list of the error/s to retry on, then set
+        `retry` to a valid `Retry` object.
+        To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
         self.pid = os.getpid()
         self.path = path
@@ -990,11 +1002,17 @@ class UnixDomainSocketConnection(Connection):
         self.socket_timeout = socket_timeout
         self.retry_on_timeout = retry_on_timeout
         if retry_on_timeout:
+            # Add TimeoutError to the errors list to retry on
+            retry_on_error.append(TimeoutError)
+        self.retry_on_error = retry_on_error
+        if self.retry_on_error:
             if retry is None:
                 self.retry = Retry(NoBackoff(), 1)
             else:
                 # deep-copy the Retry object as it is mutable
                 self.retry = copy.deepcopy(retry)
+            # Update the retry's supported errors with the specified errors
+            self.retry.update_supported_erros(retry_on_error)
         else:
             self.retry = Retry(NoBackoff(), 0)
         self.health_check_interval = health_check_interval
@@ -1052,6 +1070,7 @@ URL_QUERY_ARGUMENT_PARSERS = {
     "socket_connect_timeout": float,
     "socket_keepalive": to_bool,
     "retry_on_timeout": to_bool,
+    "retry_on_error": list,
     "max_connections": int,
     "health_check_interval": int,
     "ssl_check_hostname": to_bool,
diff --git a/redis/retry.py b/redis/retry.py
index 75504c77..6147fbd9 100644
--- a/redis/retry.py
+++ b/redis/retry.py
@@ -19,6 +19,14 @@ class Retry:
         self._retries = retries
         self._supported_errors = supported_errors
 
+    def update_supported_erros(self, specified_errors: list):
+        """
+        Updates the supported errors with the specified error types
+        """
+        self._supported_errors = tuple(
+            set(self._supported_errors + tuple(specified_errors))
+        )
+
     def call_with_retry(self, do, fail):
         """
         Execute an operation that might fail and returns its result, or
diff --git a/tox.ini b/tox.ini
index b574d17b..bceab0bf 100644
--- a/tox.ini
+++ b/tox.ini
@@ -73,12 +73,6 @@ ports =
     36379:6379/tcp
 healtcheck_cmd = python -c "import socket;print(True) if 0 == socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect_ex(('127.0.0.1',36379)) else False"
 
-[docker:lots-of-pythons]
-name = lots-of-pythons
-image = redisfab/lots-of-pythons:latest
-volumes =
-    bind:rw:{toxinidir}:/data
-
 [docker:redis_cluster]
 name = redis_cluster
 image = redisfab/redis-py-cluster:6.2.6-buster


Test patch:
diff --git a/tests/test_cluster.py b/tests/test_cluster.py
index e1c8c347..496ed981 100644
--- a/tests/test_cluster.py
+++ b/tests/test_cluster.py
@@ -22,6 +22,7 @@ from redis.crc import key_slot
 from redis.exceptions import (
     AskError,
     ClusterDownError,
+    ConnectionError,
     DataError,
     MovedError,
     NoPermissionError,
@@ -555,46 +556,24 @@ class TestRedisClusterObj:
         for node in r.get_primaries():
             assert node in nodes
 
-    def test_cluster_down_overreaches_retry_attempts(self):
+    @pytest.mark.parametrize("error", RedisCluster.ERRORS_ALLOW_RETRY)
+    def test_cluster_down_overreaches_retry_attempts(self, error):
         """
-        When ClusterDownError is thrown, test that we retry executing the
-        command as many times as configured in cluster_error_retry_attempts
+        When error that allows retry is thrown, test that we retry executing
+        the command as many times as configured in cluster_error_retry_attempts
         and then raise the exception
         """
         with patch.object(RedisCluster, "_execute_command") as execute_command:
 
-            def raise_cluster_down_error(target_node, *args, **kwargs):
+            def raise_error(target_node, *args, **kwargs):
                 execute_command.failed_calls += 1
-                raise ClusterDownError(
-                    "CLUSTERDOWN The cluster is down. Use CLUSTER INFO for "
-                    "more information"
-                )
+                raise error("mocked error")
 
-            execute_command.side_effect = raise_cluster_down_error
+            execute_command.side_effect = raise_error
 
             rc = get_mocked_redis_client(host=default_host, port=default_port)
 
-            with pytest.raises(ClusterDownError):
-                rc.get("bar")
-                assert execute_command.failed_calls == rc.cluster_error_retry_attempts
-
-    def test_connection_error_overreaches_retry_attempts(self):
-        """
-        When ConnectionError is thrown, test that we retry executing the
-        command as many times as configured in cluster_error_retry_attempts
-        and then raise the exception
-        """
-        with patch.object(RedisCluster, "_execute_command") as execute_command:
-
-            def raise_conn_error(target_node, *args, **kwargs):
-                execute_command.failed_calls += 1
-                raise ConnectionError()
-
-            execute_command.side_effect = raise_conn_error
-
-            rc = get_mocked_redis_client(host=default_host, port=default_port)
-
-            with pytest.raises(ConnectionError):
+            with pytest.raises(error):
                 rc.get("bar")
                 assert execute_command.failed_calls == rc.cluster_error_retry_attempts
 
diff --git a/tests/test_commands.py b/tests/test_commands.py
index b8dc69f9..f4ffb632 100644
--- a/tests/test_commands.py
+++ b/tests/test_commands.py
@@ -396,6 +396,28 @@ class TestRedisCommands:
         assert len(res) > 2
         assert "prefixes" in res
 
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("6.0.0")
+    def test_client_tracking(self, r, r2):
+
+        # simple case
+        assert r.client_tracking_on()
+        assert r.client_tracking_off()
+
+        # id based
+        client_id = r.client_id()
+        assert r.client_tracking_on(client_id)
+        assert r.client_tracking_off(client_id)
+
+        # id exists
+        client_id = r2.client_id()
+        assert r.client_tracking_on(client_id)
+        assert r2.client_tracking_off(client_id)
+
+        # now with some prefixes
+        with pytest.raises(exceptions.DataError):
+            assert r.client_tracking_on(prefix=["foo", "bar", "blee"])
+
     @pytest.mark.onlynoncluster
     @skip_if_server_version_lt("5.0.0")
     def test_client_unblock(self, r):
@@ -557,6 +579,13 @@ class TestRedisCommands:
         with pytest.raises(exceptions.RedisError):
             r.client_pause(timeout="not an integer")
 
+    @skip_if_server_version_lt("6.2.0")
+    def test_client_pause_all(self, r, r2):
+        assert r.client_pause(1, all=False)
+        assert r2.set("foo", "bar")
+        assert r2.get("foo") == b"bar"
+        assert r.get("foo") == b"bar"
+
     @pytest.mark.onlynoncluster
     @skip_if_server_version_lt("6.2.0")
     @skip_if_redis_enterprise()
@@ -640,6 +669,11 @@ class TestRedisCommands:
         lolwut = r.lolwut(5, 6, 7, 8).decode("utf-8")
         assert "Redis ver." in lolwut
 
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("6.2.0")
+    def test_reset(self, r):
+        assert r.reset() == "RESET"
+
     def test_object(self, r):
         r["a"] = "foo"
         assert isinstance(r.object("refcount", "a"), int)
@@ -661,6 +695,12 @@ class TestRedisCommands:
         assert isinstance(r.role()[1], int)
         assert isinstance(r.role()[2], list)
 
+    @pytest.mark.onlynoncluster
+    def test_select(self, r):
+        assert r.select(5)
+        assert r.select(2)
+        assert r.select(9)
+
     @pytest.mark.onlynoncluster
     def test_slowlog_get(self, r, slowlog):
         assert r.slowlog_reset()
diff --git a/tests/test_json.py b/tests/test_json.py
index 8bd1d777..6980e673 100644
--- a/tests/test_json.py
+++ b/tests/test_json.py
@@ -1428,6 +1428,5 @@ def test_set_path(client):
     open(nojsonfile, "a+").write("hello")
 
     result = {jsonfile: True, nojsonfile: False}
-    print(result)
     assert client.json().set_path(Path.rootPath(), root) == result
     assert client.json().get(jsonfile.rsplit(".")[0]) == {"hello": "world"}
diff --git a/tests/test_pubsub.py b/tests/test_pubsub.py
index 20ae0a05..23af4615 100644
--- a/tests/test_pubsub.py
+++ b/tests/test_pubsub.py
@@ -2,6 +2,7 @@ import platform
 import threading
 import time
 from unittest import mock
+from unittest.mock import patch
 
 import pytest
 
@@ -348,15 +349,6 @@ class TestPubSubMessages:
             "pmessage", channel, "test message", pattern=pattern
         )
 
-    def test_get_message_without_subscribe(self, r):
-        p = r.pubsub()
-        with pytest.raises(RuntimeError) as info:
-            p.get_message()
-        expect = (
-            "connection not set: " "did you forget to call subscribe() or psubscribe()?"
-        )
-        assert expect in info.exconly()
-
 
 class TestPubSubAutoDecoding:
     "These tests only validate that we get unicode values back"
@@ -549,6 +541,39 @@ class TestPubSubTimeouts:
         assert wait_for_message(p) == make_message("subscribe", "foo", 1)
         assert p.get_message(timeout=0.01) is None
 
+    def test_get_message_not_subscribed_return_none(self, r):
+        p = r.pubsub()
+        assert p.subscribed is False
+        assert p.get_message() is None
+        assert p.get_message(timeout=0.1) is None
+        with patch.object(threading.Event, "wait") as mock:
+            mock.return_value = False
+            assert p.get_message(timeout=0.01) is None
+            assert mock.called
+
+    def test_get_message_subscribe_during_waiting(self, r):
+        p = r.pubsub()
+
+        def poll(ps, expected_res):
+            assert ps.get_message() is None
+            message = ps.get_message(timeout=1)
+            assert message == expected_res
+
+        subscribe_response = make_message("subscribe", "foo", 1)
+        poller = threading.Thread(target=poll, args=(p, subscribe_response))
+        poller.start()
+        time.sleep(0.2)
+        p.subscribe("foo")
+        poller.join()
+
+    def test_get_message_wait_for_subscription_not_being_called(self, r):
+        p = r.pubsub()
+        p.subscribe("foo")
+        with patch.object(threading.Event, "wait") as mock:
+            assert p.subscribed is True
+            assert wait_for_message(p) == make_message("subscribe", "foo", 1)
+            assert mock.called is False
+
 
 class TestPubSubWorkerThread:
     @pytest.mark.skipif(
diff --git a/tests/test_retry.py b/tests/test_retry.py
index c4650bc6..00947872 100644
--- a/tests/test_retry.py
+++ b/tests/test_retry.py
@@ -1,10 +1,20 @@
+from unittest.mock import patch
+
 import pytest
 
 from redis.backoff import NoBackoff
+from redis.client import Redis
 from redis.connection import Connection, UnixDomainSocketConnection
-from redis.exceptions import ConnectionError
+from redis.exceptions import (
+    BusyLoadingError,
+    ConnectionError,
+    ReadOnlyError,
+    TimeoutError,
+)
 from redis.retry import Retry
 
+from .conftest import _get_client
+
 
 class BackoffMock:
     def __init__(self):
@@ -39,6 +49,37 @@ class TestConnectionConstructorWithRetry:
         assert isinstance(c.retry, Retry)
         assert c.retry._retries == retries
 
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error(self, Class):
+        c = Class(retry_on_error=[ReadOnlyError])
+        assert c.retry_on_error == [ReadOnlyError]
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == 1
+
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error_empty_value(self, Class):
+        c = Class(retry_on_error=[])
+        assert c.retry_on_error == []
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == 0
+
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error_and_timeout(self, Class):
+        c = Class(
+            retry_on_error=[ReadOnlyError, BusyLoadingError], retry_on_timeout=True
+        )
+        assert c.retry_on_error == [ReadOnlyError, BusyLoadingError, TimeoutError]
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == 1
+
+    @pytest.mark.parametrize("retries", range(10))
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error_retry(self, Class, retries):
+        c = Class(retry_on_error=[ReadOnlyError], retry=Retry(NoBackoff(), retries))
+        assert c.retry_on_error == [ReadOnlyError]
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == retries
+
 
 class TestRetry:
     "Test that Retry calls backoff and retries the expected number of times"
@@ -65,3 +106,85 @@ class TestRetry:
         assert self.actual_failures == 1 + retries
         assert backoff.reset_calls == 1
         assert backoff.calls == retries
+
+
+@pytest.mark.onlynoncluster
+class TestRedisClientRetry:
+    "Test the standalone Redis client behavior with retries"
+
+    def test_client_retry_on_error_with_success(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+
+            def mock_parse_response(connection, *args, **options):
+                def ok_response(connection, *args, **options):
+                    return "MOCK_OK"
+
+                parse_response.side_effect = ok_response
+                raise ReadOnlyError()
+
+            parse_response.side_effect = mock_parse_response
+            r = _get_client(Redis, request, retry_on_error=[ReadOnlyError])
+            assert r.get("foo") == "MOCK_OK"
+            assert parse_response.call_count == 2
+
+    def test_client_retry_on_error_raise(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = BusyLoadingError()
+            retries = 3
+            r = _get_client(
+                Redis,
+                request,
+                retry_on_error=[ReadOnlyError, BusyLoadingError],
+                retry=Retry(NoBackoff(), retries),
+            )
+            with pytest.raises(BusyLoadingError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == retries + 1
+
+    def test_client_retry_on_error_different_error_raised(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = TimeoutError()
+            retries = 3
+            r = _get_client(
+                Redis,
+                request,
+                retry_on_error=[ReadOnlyError],
+                retry=Retry(NoBackoff(), retries),
+            )
+            with pytest.raises(TimeoutError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == 1
+
+    def test_client_retry_on_error_and_timeout(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = TimeoutError()
+            retries = 3
+            r = _get_client(
+                Redis,
+                request,
+                retry_on_error=[ReadOnlyError],
+                retry_on_timeout=True,
+                retry=Retry(NoBackoff(), retries),
+            )
+            with pytest.raises(TimeoutError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == retries + 1
+
+    def test_client_retry_on_timeout(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = TimeoutError()
+            retries = 3
+            r = _get_client(
+                Redis, request, retry_on_timeout=True, retry=Retry(NoBackoff(), retries)
+            )
+            with pytest.raises(TimeoutError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == retries + 1