Reorganizing the parsers code, and add support for RESP3

.github/workflows/integration.yaml

@@ -51,6 +51,7 @@ jobs:
  timeout-minutes: 30
  strategy:
  max-parallel: 15
+ fail-fast: false
  matrix:
  python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', 'pypy-3.7', 'pypy-3.8', 'pypy-3.9']
  test-type: ['standalone', 'cluster']
@@ -108,6 +109,7 @@ jobs:
  name: Install package from commit hash
  runs-on: ubuntu-latest
  strategy:
+ fail-fast: false
  matrix:
  python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', 'pypy-3.7', 'pypy-3.8', 'pypy-3.9']
  steps:

benchmarks/socket_read_size.py

@@ -1,12 +1,12 @@
 from base import Benchmark
 
-from redis.connection import HiredisParser, PythonParser
+from redis.connection import PythonParser, _HiredisParser
 
 
 class SocketReadBenchmark(Benchmark):
 
  ARGUMENTS = (
- {"name": "parser", "values": [PythonParser, HiredisParser]},
+ {"name": "parser", "values": [PythonParser, _HiredisParser]},
  {
  "name": "value_size",
  "values": [10, 100, 1000, 10000, 100000, 1000000, 10000000, 100000000],

redis/asyncio/__init__.py

@@ -7,7 +7,6 @@
  SSLConnection,
  UnixDomainSocketConnection,
 )
-from redis.asyncio.parser import CommandsParser
 from redis.asyncio.sentinel import (
  Sentinel,
  SentinelConnectionPool,
@@ -38,7 +37,6 @@
  "BlockingConnectionPool",
  "BusyLoadingError",
  "ChildDeadlockedError",
- "CommandsParser",
  "Connection",
  "ConnectionError",
  "ConnectionPool",

redis/asyncio/client.py

@@ -253,6 +253,9 @@ def __init__(
 
  self.response_callbacks = CaseInsensitiveDict(self.__class__.RESPONSE_CALLBACKS)
 
+ if self.connection_pool.connection_kwargs.get("protocol") == "3":
+ self.response_callbacks.update(self.__class__.RESP3_RESPONSE_CALLBACKS)
+
  # If using a single connection client, we need to lock creation-of and use-of
  # the client in order to avoid race conditions such as using asyncio.gather
  # on a set of redis commands

redis/asyncio/cluster.py

@@ -17,15 +17,8 @@
 )
 
 from redis.asyncio.client import ResponseCallbackT
-from redis.asyncio.connection import (
- Connection,
- DefaultParser,
- Encoder,
- SSLConnection,
- parse_url,
-)
+from redis.asyncio.connection import Connection, DefaultParser, SSLConnection, parse_url
 from redis.asyncio.lock import Lock
-from redis.asyncio.parser import CommandsParser
 from redis.asyncio.retry import Retry
 from redis.backoff import default_backoff
 from redis.client import EMPTY_RESPONSE, NEVER_DECODE, AbstractRedis
@@ -60,6 +53,7 @@
  TimeoutError,
  TryAgainError,
 )
+from redis.parsers import AsyncCommandsParser, Encoder
 from redis.typing import AnyKeyT, EncodableT, KeyT
 from redis.utils import dict_merge, safe_str, str_if_bytes
 
@@ -250,6 +244,7 @@ def __init__(
  ssl_certfile: Optional[str] = None,
  ssl_check_hostname: bool = False,
  ssl_keyfile: Optional[str] = None,
+ protocol: Optional[int] = 2,
  ) -> None:
  if db:
  raise RedisClusterException(
@@ -290,6 +285,7 @@ def __init__(
  "socket_keepalive_options": socket_keepalive_options,
  "socket_timeout": socket_timeout,
  "retry": retry,
+ "protocol": protocol,
  }
 
  if ssl:
@@ -344,7 +340,7 @@ def __init__(
  self.cluster_error_retry_attempts = cluster_error_retry_attempts
  self.connection_error_retry_attempts = connection_error_retry_attempts
  self.reinitialize_counter = 0
- self.commands_parser = CommandsParser()
+ self.commands_parser = AsyncCommandsParser()
  self.node_flags = self.__class__.NODE_FLAGS.copy()
  self.command_flags = self.__class__.COMMAND_FLAGS.copy()
  self.response_callbacks = kwargs["response_callbacks"]