Add Async RedisCluster (#2099)

* Copy Cluster Client, Commands, Commands Parser, Tests for asyncio

* Async Cluster Tests: Async/Await

* Add Async RedisCluster

* cluster: use ERRORS_ALLOW_RETRY from self.__class__

* async_cluster: rework redis_connection, initialize, & close

- move redis_connection from NodesManager to ClusterNode & handle all related logic in ClusterNode class
- use Locks while initializing or closing
- in case of error, close connections instead of instantly reinitializing
- create ResourceWarning instead of manually deleting client object
- use asyncio.gather to run commands/initialize/close in parallel

- inline single use functions
- fix test_acl_log for py3.6

* async_cluster: add types

* async_cluster: add docs

* docs: update sphinx & add sphinx_autodoc_typehints

* async_cluster: move TargetNodesT to cluster module

* async_cluster/commands: inherit commands from sync class if possible

* async_cluster: add benchmark script with aredis & aioredis-cluster

* async_cluster: remove logging

* async_cluster: inline functions

* async_cluster: manage Connection instead of Redis Client

* async_cluster/commands: optimize parser

* async_cluster: use ensure_future & generators for gather

* async_conn: optimize

* async_cluster: optimize determine_slot

* async_cluster: optimize determine_nodes

* async_cluster/parser: optimize _get_moveable_keys

* async_cluster: inlined check_slots_coverage

* async_cluster: update docstrings

* async_cluster: add concurrent test & use read_response/_update_moved_slots without lock

Co-authored-by: Chayim <chayim@users.noreply.github.com>

1 files changed, 4 insertions, 8 deletions

diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index 8dde96e..6db5489 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -172,6 +172,7 @@ class Redis(
         username: Optional[str] = None,
         retry: Optional[Retry] = None,
         auto_close_connection_pool: bool = True,
+        redis_connect_func=None,
     ):
         """
         Initialize a new Redis client.
@@ -200,6 +201,7 @@ class Redis(
                 "max_connections": max_connections,
                 "health_check_interval": health_check_interval,
                 "client_name": client_name,
+                "redis_connect_func": redis_connect_func,
             }
             # based on input, setup appropriate connection args
             if unix_socket_path is not None:
@@ -263,11 +265,7 @@ class Redis(
         """Get the connection's key-word arguments"""
         return self.connection_pool.connection_kwargs
 
-    def load_external_module(
-        self,
-        funcname,
-        func,
-    ):
+    def load_external_module(self, funcname, func):
         """
         This function can be used to add externally defined redis modules,
         and their namespaces to the redis client.
@@ -426,9 +424,7 @@ class Redis(
     def __del__(self, _warnings: Any = warnings) -> None:
         if self.connection is not None:
             _warnings.warn(
-                f"Unclosed client session {self!r}",
-                ResourceWarning,
-                source=self,
+                f"Unclosed client session {self!r}", ResourceWarning, source=self
             )
             context = {"client": self, "message": self._DEL_MESSAGE}
             asyncio.get_event_loop().call_exception_handler(context)