Reorganizing the code related to maintenance push notifications. (#3785)

* Extracting maintenance push notifications logic in separate abstract classes. Usingle multiple inheritance for Connection and ConnectionPool classes to expose the maintenance push notifications for the existing connections.

* Fixing credential tests and linters errors

* Adding better handling of configs and fixing tests

* Fixing linter- unused import,  and a docstring

* Fixing check for correct parser when protocol is 3

* Moving _should_reconnect related code back to generic ConnectionInterface as it is also used by AA

* Fixing e2e tests redis client setup

* Applying review comments

Author: petyaslavova

redis/client.py

@@ -58,7 +58,6 @@
 from redis.lock import Lock
 from redis.maint_notifications import (
     MaintNotificationsConfig,
-    MaintNotificationsPoolHandler,
 )
 from redis.retry import Retry
 from redis.utils import (
@@ -278,6 +277,17 @@ def __init__(
         single_connection_client:
             if `True`, connection pool is not used. In that case `Redis`
             instance use is not thread safe.
+        decode_responses:
+            if `True`, the response will be decoded to utf-8.
+            Argument is ignored when connection_pool is provided.
+        maint_notifications_config:
+            configuration the pool to support maintenance notifications - see
+            `redis.maint_notifications.MaintNotificationsConfig` for details.
+            Only supported with RESP3
+            If not provided and protocol is RESP3, the maintenance notifications
+            will be enabled by default (logic is included in the connection pool
+            initialization).
+            Argument is ignored when connection_pool is provided.
         """
         if event_dispatcher is None:
             self._event_dispatcher = EventDispatcher()
@@ -354,6 +364,22 @@ def __init__(
                             "cache_config": cache_config,
                         }
                     )
+                maint_notifications_enabled = (
+                    maint_notifications_config and maint_notifications_config.enabled
+                )
+                if maint_notifications_enabled and protocol not in [
+                    3,
+                    "3",
+                ]:
+                    raise RedisError(
+                        "Maintenance notifications handlers on connection are only supported with RESP version 3"
+                    )
+                if maint_notifications_config:
+                    kwargs.update(
+                        {
+                            "maint_notifications_config": maint_notifications_config,
+                        }
+                    )
             connection_pool = ConnectionPool(**kwargs)
             self._event_dispatcher.dispatch(
                 AfterPooledConnectionsInstantiationEvent(
@@ -377,23 +403,6 @@ def __init__(
         ]:
             raise RedisError("Client caching is only supported with RESP version 3")
 
-        if maint_notifications_config and self.connection_pool.get_protocol() not in [
-            3,
-            "3",
-        ]:
-            raise RedisError(
-                "Push handlers on connection are only supported with RESP version 3"
-            )
-        if maint_notifications_config and maint_notifications_config.enabled:
-            self.maint_notifications_pool_handler = MaintNotificationsPoolHandler(
-                self.connection_pool, maint_notifications_config
-            )
-            self.connection_pool.set_maint_notifications_pool_handler(
-                self.maint_notifications_pool_handler
-            )
-        else:
-            self.maint_notifications_pool_handler = None
-
         self.single_connection_lock = threading.RLock()
         self.connection = None
         self._single_connection_client = single_connection_client
@@ -591,15 +600,9 @@ def monitor(self):
         return Monitor(self.connection_pool)
 
     def client(self):
-        maint_notifications_config = (
-            None
-            if self.maint_notifications_pool_handler is None
-            else self.maint_notifications_pool_handler.config
-        )
         return self.__class__(
             connection_pool=self.connection_pool,
             single_connection_client=True,
-            maint_notifications_config=maint_notifications_config,
         )
 
     def __enter__(self):

redis/cluster.py

@@ -50,6 +50,7 @@
     WatchError,
 )
 from redis.lock import Lock
+from redis.maint_notifications import MaintNotificationsConfig
 from redis.retry import Retry
 from redis.utils import (
     deprecated_args,
@@ -1663,6 +1664,11 @@ def create_redis_node(self, host, port, **kwargs):
             backoff=NoBackoff(), retries=0, supported_errors=(ConnectionError,)
         )
 
+        protocol = kwargs.get("protocol", None)
+        if protocol in [3, "3"]:
+            kwargs.update(
+                {"maint_notifications_config": MaintNotificationsConfig(enabled=False)}
+            )
         if self.from_url:
             # Create a redis node with a costumed connection pool
             kwargs.update({"host": host})