Remove __del__ handler, fix pubsub weakref callback handling

Author: kristjanvalur

redis/asyncio/client.py

@@ -720,15 +720,11 @@ async def __aenter__(self):
     async def __aexit__(self, exc_type, exc_value, traceback):
         await self.reset()
 
-    def __del__(self):
-        if self.connection:
-            self.connection.clear_connect_callbacks()
-
     async def reset(self):
         async with self._lock:
             if self.connection:
                 await self.connection.disconnect()
-                self.connection.clear_connect_callbacks()
+                self.connection.deregister_connect_callback(self.on_connect)
                 await self.connection_pool.release(self.connection)
                 self.connection = None
             self.channels = {}

redis/asyncio/connection.py

@@ -216,10 +216,15 @@ def is_connected(self):
         return self._reader is not None and self._writer is not None
 
     def register_connect_callback(self, callback):
-        self._connect_callbacks.append(weakref.WeakMethod(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: Type[BaseParser]) -> None:
         """
@@ -262,6 +267,8 @@ async 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()
             task = callback(self)