similarly make non-async connect callbacks internal, use same system as for async.

Author: kristjanvalur

redis/client.py

@@ -690,7 +690,7 @@ def __del__(self):
     def reset(self):
         if self.connection:
             self.connection.disconnect()
-            self.connection.clear_connect_callbacks()
+            self.connection._deregister_connect_callback(self.on_connect)
             self.connection_pool.release(self.connection)
             self.connection = None
         self.health_check_response_counter = 0
@@ -748,7 +748,7 @@ def execute_command(self, *args):
             )
             # register a callback that re-subscribes to any channels we
             # were listening to when we were disconnected
-            self.connection.register_connect_callback(self.on_connect)
+            self.connection._register_connect_callback(self.on_connect)
             if self.push_handler_func is not None and not HIREDIS_AVAILABLE:
                 self.connection._parser.set_push_handler(self.push_handler_func)
         connection = self.connection

redis/cluster.py

@@ -1771,7 +1771,7 @@ def execute_command(self, *args):
             )
             # register a callback that re-subscribes to any channels we
             # were listening to when we were disconnected
-            self.connection.register_connect_callback(self.on_connect)
+            self.connection._register_connect_callback(self.on_connect)
             if self.push_handler_func is not None and not HIREDIS_AVAILABLE:
                 self.connection._parser.set_push_handler(self.push_handler_func)
         connection = self.connection

redis/connection.py

@@ -237,11 +237,16 @@ def _construct_command_packer(self, packer):
         else:
             return PythonRespSerializer(self._buffer_cutoff, self.encoder.encode)
 
-    def register_connect_callback(self, callback):
-        self._connect_callbacks.append(weakref.WeakMethod(callback))
+    def _register_connect_callback(self, callback):
+        wm = weakref.WeakMethod(callback)
+        if wm not in self._connect_callbacks:
+            self._connect_callbacks.append(wm)
 
-    def clear_connect_callbacks(self):
-        self._connect_callbacks = []
+    def _deregister_connect_callback(self, callback):
+        try:
+            self._connect_callbacks.remove(weakref.WeakMethod(callback))
+        except ValueError:
+            pass
 
     def set_parser(self, parser_class):
         """
@@ -279,6 +284,8 @@ def connect(self):
 
         # run any user callbacks. right now the only internal callback
         # is for pubsub channel/pattern resubscription
+        # first, remove any dead weakrefs
+        self._connect_callbacks = [ref for ref in self._connect_callbacks if ref()]
         for ref in self._connect_callbacks:
             callback = ref()
             if callback: