Allow tracking/reporting and closing of "lost" connections.

ConnectionPool keeps a WeakSet of in_use connections, allowing lost ones to be collected.
Collection produces a warning and closes the underlying transport.

redis/asyncio/client.py

@@ -546,6 +546,7 @@ def __del__(
                 _grl().call_exception_handler(context)
             except RuntimeError:
                 pass
+            self.connection._close()
 
     async def aclose(self, close_connection_pool: Optional[bool] = None) -> None:
         """

redis/asyncio/connection.py

@@ -5,6 +5,7 @@
 import socket
 import ssl
 import sys
+import warnings
 import weakref
 from abc import abstractmethod
 from itertools import chain
@@ -204,9 +205,27 @@ def __init__(
                 raise ConnectionError("protocol must be either 2 or 3")
             self.protocol = protocol
 
+    def __del__(self, _warnings: Any = warnings):
+        # For some reason, the individual streams don't get properly garbage
+        # collected and therefore produce no resource warnings.  We add one
+        # here, in the same style as those from the stdlib.
+        if self._writer:
+            _warnings.warn(
+                f"unclosed Connection {self!r}", ResourceWarning, source=self
+            )
+            self._close()
+
+    def _close(self):
+        """
+        Internal method to silently close the connection without waiting
+        """
+        if self._writer:
+            self._writer.close()
+            self._writer = self._reader = None
+
     def __repr__(self):
         repr_args = ",".join((f"{k}={v}" for k, v in self.repr_pieces()))
-        return f"{self.__class__.__name__}<{repr_args}>"
+        return f"<{self.__class__.__module__}.{self.__class__.__name__}({repr_args})>"
 
     @abstractmethod
     def repr_pieces(self):
@@ -1017,7 +1036,7 @@ def __repr__(self):
 
     def reset(self):
         self._available_connections = []
-        self._in_use_connections = set()
+        self._in_use_connections = weakref.WeakSet()
 
     def can_get_connection(self) -> bool:
         """Return True if a connection can be retrieved from the pool."""