Hitless upgrade support implementation for synchronous Redis client.

redis/_parsers/base.py

@@ -202,17 +202,22 @@ def handle_push_response(self, response, **kwargs):
         if msg_type in _INVALIDATION_MESSAGE and self.invalidation_push_handler_func:
             return self.invalidation_push_handler_func(response)
         if msg_type in _MOVING_MESSAGE and self.node_moving_push_handler_func:
-            if msg_type in _MOVING_MESSAGE:
-                host, port = response[2].split(":")
-                ttl = response[1]
-                notification = NodeMovingEvent(host, port, ttl)
-                return self.node_moving_push_handler_func(notification)
+            # TODO: PARSE latest format when available
+            host, port = response[2].decode().split(":")
+            ttl = response[1]
+            id = 1  # Hardcoded value until the notification starts including the id
+            notification = NodeMovingEvent(id, host, port, ttl)
+            return self.node_moving_push_handler_func(notification)
         if msg_type in _MAINTENANCE_MESSAGES and self.maintenance_push_handler_func:
             if msg_type in _MIGRATING_MESSAGE:
+                # TODO: PARSE latest format when available
                 ttl = response[1]
-                notification = NodeMigratingEvent(ttl)
+                id = 2  # Hardcoded value until the notification starts including the id
+                notification = NodeMigratingEvent(id, ttl)
             elif msg_type in _MIGRATED_MESSAGE:
-                notification = NodeMigratedEvent()
+                # TODO: PARSE latest format when available
+                id = 3  # Hardcoded value until the notification starts including the id
+                notification = NodeMigratedEvent(id)
             else:
                 notification = None
             if notification is not None:
@@ -258,18 +263,21 @@ async def handle_push_response(self, response, **kwargs):
             return await self.invalidation_push_handler_func(response)
         if msg_type in _MOVING_MESSAGE and self.node_moving_push_handler_func:
             # push notification from enterprise cluster for node moving
+            # TODO: PARSE latest format when available
             host, port = response[2].split(":")
             ttl = response[1]
-            id = 1  # TODO: get unique id from push notification
+            id = 1  # Hardcoded value for async parser
             notification = NodeMovingEvent(id, host, port, ttl)
             return await self.node_moving_push_handler_func(notification)
         if msg_type in _MAINTENANCE_MESSAGES and self.maintenance_push_handler_func:
             if msg_type in _MIGRATING_MESSAGE:
+                # TODO: PARSE latest format when available
                 ttl = response[1]
-                id = 1  # TODO: get unique id from push notification
+                id = 2  # Hardcoded value for async parser
                 notification = NodeMigratingEvent(id, ttl)
             elif msg_type in _MIGRATED_MESSAGE:
-                id = 1  # TODO: get unique id from push notification
+                # TODO: PARSE latest format when available
+                id = 3  # Hardcoded value for async parser
                 notification = NodeMigratedEvent(id)
             return await self.maintenance_push_handler_func(notification)
 
@@ -281,7 +289,7 @@ 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_moving_push_handler_func(self, node_moving_push_handler_func):
+    def set_node_moving_push_handler(self, node_moving_push_handler_func):
         self.node_moving_push_handler_func = node_moving_push_handler_func
 
     def set_maintenance_push_handler(self, maintenance_push_handler_func):

redis/_parsers/hiredis.py

@@ -152,6 +152,7 @@ def read_response(self, disable_decoding=False, push_request=False):
                     disable_decoding=disable_decoding,
                     push_request=push_request,
                 )
+            return response
 
         if disable_decoding:
             response = self._reader.gets(False)

redis/asyncio/connection.py

@@ -1308,6 +1308,8 @@ def __init__(
         )
         self._condition = asyncio.Condition()
         self.timeout = timeout
+        self._in_maintenance = False
+        self._locked = False
 
     @deprecated_args(
         args_to_warn=["*"],

redis/client.py

@@ -1,5 +1,4 @@
 import copy
-import logging
 import re
 import threading
 import time
@@ -668,9 +667,6 @@ def _execute_command(self, *args, **options):
 
         finally:
             if conn and conn.should_reconnect():
-                logging.debug(
-                    f"***** Redis reconnect before exit _execute_command --> notification for {conn._sock.getpeername()}"
-                )
                 self._close_connection(conn)
                 conn.connect()
             if self._single_connection_client:
@@ -963,9 +959,6 @@ def _execute(self, conn, command, *args, **kwargs):
             lambda _: self._reconnect(conn),
         )
         if conn.should_reconnect():
-            logging.debug(
-                f"***** PubSub --> Reconnect on notification for {conn._sock.getpeername()}"
-            )
             self._reconnect(conn)
 
         return response
@@ -1669,7 +1662,7 @@ def execute(self, raise_on_error: bool = True) -> List[Any]:
                 lambda error: self._disconnect_raise_on_watching(conn, error),
             )
         finally:
-            # in reset() the connection is diconnected before returned to the pool if
+            # in reset() the connection is disconnected before returned to the pool if
             # it is marked for reconnect.
             self.reset()