Standalone client POC

Author: petyaslavova

redis/_parsers/base.py

@@ -159,22 +159,28 @@ async def read_response(
 
 
 _INVALIDATION_MESSAGE = [b"invalidate", "invalidate"]
+_MOVING_MESSAGE = [b"MOVING", "MOVING"]
 
 
 class PushNotificationsParser(Protocol):
     """Protocol defining RESP3-specific parsing functionality"""
 
     pubsub_push_handler_func: Callable
     invalidation_push_handler_func: Optional[Callable] = None
+    node_replacement_push_handler_func: Optional[Callable] = None
 
     def handle_pubsub_push_response(self, response):
         """Handle pubsub push responses"""
         raise NotImplementedError()
 
     def handle_push_response(self, response, **kwargs):
-        if response[0] not in _INVALIDATION_MESSAGE:
+        msg_type = response[0]
+        if msg_type not in _INVALIDATION_MESSAGE and msg_type not in _MOVING_MESSAGE:
             return self.pubsub_push_handler_func(response)
-        if self.invalidation_push_handler_func:
+        if msg_type in _MOVING_MESSAGE and self.node_replacement_push_handler_func:
+            # push notification for enterprise cluster node replacement
+            return self.node_replacement_push_handler_func(response)
+        if msg_type in _INVALIDATION_MESSAGE and self.invalidation_push_handler_func:
             return self.invalidation_push_handler_func(response)
 
     def set_pubsub_push_handler(self, pubsub_push_handler_func):
@@ -183,22 +189,32 @@ def set_pubsub_push_handler(self, pubsub_push_handler_func):
     def set_invalidation_push_handler(self, invalidation_push_handler_func):
         self.invalidation_push_handler_func = invalidation_push_handler_func
 
+    def set_node_replacement_push_handler_func(
+        self, node_replacement_push_handler_func
+    ):
+        self.node_replacement_push_handler_func = node_replacement_push_handler_func
+
 
 class AsyncPushNotificationsParser(Protocol):
     """Protocol defining async RESP3-specific parsing functionality"""
 
     pubsub_push_handler_func: Callable
     invalidation_push_handler_func: Optional[Callable] = None
+    node_replacement_push_handler_func: Optional[Callable] = None
 
     async def handle_pubsub_push_response(self, response):
         """Handle pubsub push responses asynchronously"""
         raise NotImplementedError()
 
     async def handle_push_response(self, response, **kwargs):
         """Handle push responses asynchronously"""
-        if response[0] not in _INVALIDATION_MESSAGE:
+        msg_type = response[0]
+        if msg_type not in _INVALIDATION_MESSAGE and msg_type not in _MOVING_MESSAGE:
             return await self.pubsub_push_handler_func(response)
-        if self.invalidation_push_handler_func:
+        if msg_type in _MOVING_MESSAGE and self.node_replacement_push_handler_func:
+            # push notification for enterprise cluster node replacement
+            return await self.node_replacement_push_handler_func(response)
+        if msg_type in _INVALIDATION_MESSAGE and self.invalidation_push_handler_func:
             return await self.invalidation_push_handler_func(response)
 
     def set_pubsub_push_handler(self, pubsub_push_handler_func):
@@ -209,6 +225,11 @@ def set_invalidation_push_handler(self, invalidation_push_handler_func):
         """Set the invalidation push handler function"""
         self.invalidation_push_handler_func = invalidation_push_handler_func
 
+    def set_node_replacement_push_handler_func(
+        self, node_replacement_push_handler_func
+    ):
+        self.node_replacement_push_handler_func = node_replacement_push_handler_func
+
 
 class _AsyncRESPBase(AsyncBaseParser):
     """Base class for async resp parsing"""

redis/_parsers/resp3.py

@@ -3,6 +3,7 @@
 
 from ..exceptions import ConnectionError, InvalidResponse, ResponseError
 from ..typing import EncodableT
+from ..notifications import NodeReplacementNotification
 from .base import (
     AsyncPushNotificationsParser,
     PushNotificationsParser,
@@ -18,13 +19,25 @@ class _RESP3Parser(_RESPBase, PushNotificationsParser):
     def __init__(self, socket_read_size):
         super().__init__(socket_read_size)
         self.pubsub_push_handler_func = self.handle_pubsub_push_response
+        self.node_replacement_push_handler_func = (
+            self.handle_node_replacement_push_response
+        )
         self.invalidation_push_handler_func = None
 
     def handle_pubsub_push_response(self, response):
         logger = getLogger("push_response")
         logger.debug("Push response: " + str(response))
         return response
 
+    def handle_node_replacement_push_response(self, response):
+        logger = getLogger("push_response")
+        logger.debug("Push response: " + str(response))
+
+        host, port = response[2].split(":")
+        ttl = response[1]
+        pushed_notification = NodeReplacementNotification(host, port, ttl)
+        return pushed_notification
+
     def read_response(self, disable_decoding=False, push_request=False):
         pos = self._buffer.get_pos() if self._buffer else None
         try:
@@ -116,13 +129,17 @@ def _read_response(self, disable_decoding=False, push_request=False):
                 )
                 for _ in range(int(response))
             ]
-            response = self.handle_push_response(response)
+            push_response = self.handle_push_response(response)
             if not push_request:
-                return self._read_response(
+                response = self._read_response(
                     disable_decoding=disable_decoding, push_request=push_request
                 )
+                if isinstance(push_response, NodeReplacementNotification):
+                    return push_response
+                else:
+                    return response
             else:
-                return response
+                return push_response
         else:
             raise InvalidResponse(f"Protocol Error: {raw!r}")
 

redis/client.py

@@ -244,6 +244,7 @@ def __init__(
         cache: Optional[CacheInterface] = None,
         cache_config: Optional[CacheConfig] = None,
         event_dispatcher: Optional[EventDispatcher] = None,
+        enable_node_replacement_push_notifications: bool = False,
     ) -> None:
         """
         Initialize a new Redis client.
@@ -297,6 +298,7 @@ def __init__(
                 "redis_connect_func": redis_connect_func,
                 "credential_provider": credential_provider,
                 "protocol": protocol,
+                "enable_node_replacement_push_notifications": enable_node_replacement_push_notifications,
             }
             # based on input, setup appropriate connection args
             if unix_socket_path is not None:

redis/connection.py

@@ -1374,6 +1374,7 @@ def __init__(
         connection_class=Connection,
         max_connections: Optional[int] = None,
         cache_factory: Optional[CacheFactoryInterface] = None,
+        enable_node_replacement_push_notifications: bool = False,
         **connection_kwargs,
     ):
         max_connections = max_connections or 2**31
@@ -1383,6 +1384,9 @@ def __init__(
         self.connection_class = connection_class
         self.connection_kwargs = connection_kwargs
         self.max_connections = max_connections
+        self.enable_node_replacement_push_notifications = (
+            enable_node_replacement_push_notifications
+        )
         self.cache = None
         self._cache_factory = cache_factory
 
@@ -1408,6 +1412,12 @@ def __init__(
         connection_kwargs.pop("cache", None)
         connection_kwargs.pop("cache_config", None)
 
+        if self.enable_node_replacement_push_notifications:
+            if connection_kwargs.get("protocol") not in [3, "3"]:
+                raise RedisError(
+                    "Push notifications on node replacement are only supported with RESP version 3"
+                )
+
         self._event_dispatcher = self.connection_kwargs.get("event_dispatcher", None)
         if self._event_dispatcher is None:
             self._event_dispatcher = EventDispatcher()
@@ -1694,13 +1704,15 @@ def __init__(
         timeout=20,
         connection_class=Connection,
         queue_class=LifoQueue,
+        enable_node_replacement_push_notifications=False,
         **connection_kwargs,
     ):
         self.queue_class = queue_class
         self.timeout = timeout
         super().__init__(
             connection_class=connection_class,
             max_connections=max_connections,
+            enable_node_replacement_push_notifications=enable_node_replacement_push_notifications,
             **connection_kwargs,
         )
 

redis/notifications.py

@@ -0,0 +1,27 @@
+import time
+
+
+class NodeReplacementNotification:
+    """
+    Initialize a new NodeReplacementNotification.
+
+    Args:
+        new_node_host (str): Hostname or IP address of the new replacement node
+        new_node_port (int): Port number of the new replacement node
+        ttl (int): Time-to-live in seconds for this notification
+    """
+
+    def __init__(self, new_node_host: str, new_node_port: int, ttl: int):
+        self.new_node_host = new_node_host
+        self.new_node_port = new_node_port
+        self.ttl = ttl
+        self.creation_time = time.time()
+
+    def is_expired(self) -> bool:
+        """
+        Check if this notification has expired based on its TTL.
+
+        Returns:
+            bool: True if the notification has expired, False otherwise
+        """
+        return time.time() > (self.creation_time + self.ttl)