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, 10 insertions, 2 deletions

diff --git a/redis/typing.py b/redis/typing.py
index 73ae411..6748612 100644
--- a/redis/typing.py
+++ b/redis/typing.py
@@ -1,13 +1,14 @@
 # from __future__ import annotations
 
 from datetime import datetime, timedelta
-from typing import TYPE_CHECKING, Iterable, TypeVar, Union
+from typing import TYPE_CHECKING, Any, Awaitable, Iterable, TypeVar, Union
 
 from redis.compat import Protocol
 
 if TYPE_CHECKING:
     from redis.asyncio.connection import ConnectionPool as AsyncConnectionPool
-    from redis.connection import ConnectionPool
+    from redis.asyncio.connection import Encoder as AsyncEncoder
+    from redis.connection import ConnectionPool, Encoder
 
 
 EncodedT = Union[bytes, memoryview]
@@ -43,3 +44,10 @@ class CommandsProtocol(Protocol):
 
     def execute_command(self, *args, **options):
         ...
+
+
+class ClusterCommandsProtocol(CommandsProtocol):
+    encoder: Union["AsyncEncoder", "Encoder"]
+
+    def execute_command(self, *args, **options) -> Union[Any, Awaitable]:
+        ...