Reorganizing the code, in preparation for RESP3

7 files changed, 986 insertions, 0 deletions

diff --git a/redis/parsers/__init__.py b/redis/parsers/__init__.py
new file mode 100644
index 0000000..c6d950d
--- /dev/null
+++ b/redis/parsers/__init__.py
@@ -0,0 +1,16 @@
+from .commands import AsyncCommandsParser, CommandsParser
+from .encoders import Encoder
+from .hiredis import AsyncHiredisParser, HiredisParser
+from .protocol import BaseParser
+from .resp2 import AsyncRESP2Parser, RESP2Parser
+
+__all__ = [
+    "AsyncCommandsParser",
+    "AsyncHiredisParser",
+    "AsyncRESP2Parser",
+    "CommandsParser",
+    "Encoder",
+    "BaseParser",
+    "HiredisParser",
+    "RESP2Parser",
+]
diff --git a/redis/parsers/commands.py b/redis/parsers/commands.py
new file mode 100644
index 0000000..7981a52
--- /dev/null
+++ b/redis/parsers/commands.py
@@ -0,0 +1,259 @@
+from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
+
+from redis.exceptions import RedisError, ResponseError
+from redis.utils import str_if_bytes
+
+if TYPE_CHECKING:
+    from redis.asyncio.cluster import ClusterNode
+
+
+class AsyncCommandsParser:
+    """
+    Parses Redis commands to get command keys.
+
+    COMMAND output is used to determine key locations.
+    Commands that do not have a predefined key location are flagged with 'movablekeys',
+    and these commands' keys are determined by the command 'COMMAND GETKEYS'.
+
+    NOTE: Due to a bug in redis<7.0, this does not work properly
+    for EVAL or EVALSHA when the `numkeys` arg is 0.
+     - issue: https://github.com/redis/redis/issues/9493
+     - fix: https://github.com/redis/redis/pull/9733
+
+    So, don't use this with EVAL or EVALSHA.
+    """
+
+    __slots__ = ("commands", "node")
+
+    def __init__(self) -> None:
+        self.commands: Dict[str, Union[int, Dict[str, Any]]] = {}
+
+    async def initialize(self, node: Optional["ClusterNode"] = None) -> None:
+        if node:
+            self.node = node
+
+        commands = await self.node.execute_command("COMMAND")
+        for cmd, command in commands.items():
+            if "movablekeys" in command["flags"]:
+                commands[cmd] = -1
+            elif command["first_key_pos"] == 0 and command["last_key_pos"] == 0:
+                commands[cmd] = 0
+            elif command["first_key_pos"] == 1 and command["last_key_pos"] == 1:
+                commands[cmd] = 1
+        self.commands = {cmd.upper(): command for cmd, command in commands.items()}
+
+    # As soon as this PR is merged into Redis, we should reimplement
+    # our logic to use COMMAND INFO changes to determine the key positions
+    # https://github.com/redis/redis/pull/8324
+    async def get_keys(self, *args: Any) -> Optional[Tuple[str, ...]]:
+        if len(args) < 2:
+            # The command has no keys in it
+            return None
+
+        try:
+            command = self.commands[args[0]]
+        except KeyError:
+            # try to split the command name and to take only the main command
+            # e.g. 'memory' for 'memory usage'
+            args = args[0].split() + list(args[1:])
+            cmd_name = args[0].upper()
+            if cmd_name not in self.commands:
+                # We'll try to reinitialize the commands cache, if the engine
+                # version has changed, the commands may not be current
+                await self.initialize()
+                if cmd_name not in self.commands:
+                    raise RedisError(
+                        f"{cmd_name} command doesn't exist in Redis commands"
+                    )
+
+            command = self.commands[cmd_name]
+
+        if command == 1:
+            return (args[1],)
+        if command == 0:
+            return None
+        if command == -1:
+            return await self._get_moveable_keys(*args)
+
+        last_key_pos = command["last_key_pos"]
+        if last_key_pos < 0:
+            last_key_pos = len(args) + last_key_pos
+        return args[command["first_key_pos"] : last_key_pos + 1 : command["step_count"]]
+
+    async def _get_moveable_keys(self, *args: Any) -> Optional[Tuple[str, ...]]:
+        try:
+            keys = await self.node.execute_command("COMMAND GETKEYS", *args)
+        except ResponseError as e:
+            message = e.__str__()
+            if (
+                "Invalid arguments" in message
+                or "The command has no key arguments" in message
+            ):
+                return None
+            else:
+                raise e
+        return keys
+
+
+class CommandsParser:
+    """
+    Parses Redis commands to get command keys.
+    COMMAND output is used to determine key locations.
+    Commands that do not have a predefined key location are flagged with
+    'movablekeys', and these commands' keys are determined by the command
+    'COMMAND GETKEYS'.
+    """
+
+    def __init__(self, redis_connection):
+        self.commands = {}
+        self.initialize(redis_connection)
+
+    def initialize(self, r):
+        commands = r.execute_command("COMMAND")
+        uppercase_commands = []
+        for cmd in commands:
+            if any(x.isupper() for x in cmd):
+                uppercase_commands.append(cmd)
+        for cmd in uppercase_commands:
+            commands[cmd.lower()] = commands.pop(cmd)
+        self.commands = commands
+
+    def parse_subcommand(self, command, **options):
+        cmd_dict = {}
+        cmd_name = str_if_bytes(command[0])
+        cmd_dict["name"] = cmd_name
+        cmd_dict["arity"] = int(command[1])
+        cmd_dict["flags"] = [str_if_bytes(flag) for flag in command[2]]
+        cmd_dict["first_key_pos"] = command[3]
+        cmd_dict["last_key_pos"] = command[4]
+        cmd_dict["step_count"] = command[5]
+        if len(command) > 7:
+            cmd_dict["tips"] = command[7]
+            cmd_dict["key_specifications"] = command[8]
+            cmd_dict["subcommands"] = command[9]
+        return cmd_dict
+
+    # As soon as this PR is merged into Redis, we should reimplement
+    # our logic to use COMMAND INFO changes to determine the key positions
+    # https://github.com/redis/redis/pull/8324
+    def get_keys(self, redis_conn, *args):
+        """
+        Get the keys from the passed command.
+
+        NOTE: Due to a bug in redis<7.0, this function does not work properly
+        for EVAL or EVALSHA when the `numkeys` arg is 0.
+         - issue: https://github.com/redis/redis/issues/9493
+         - fix: https://github.com/redis/redis/pull/9733
+
+        So, don't use this function with EVAL or EVALSHA.
+        """
+        if len(args) < 2:
+            # The command has no keys in it
+            return None
+
+        cmd_name = args[0].lower()
+        if cmd_name not in self.commands:
+            # try to split the command name and to take only the main command,
+            # e.g. 'memory' for 'memory usage'
+            cmd_name_split = cmd_name.split()
+            cmd_name = cmd_name_split[0]
+            if cmd_name in self.commands:
+                # save the splitted command to args
+                args = cmd_name_split + list(args[1:])
+            else:
+                # We'll try to reinitialize the commands cache, if the engine
+                # version has changed, the commands may not be current
+                self.initialize(redis_conn)
+                if cmd_name not in self.commands:
+                    raise RedisError(
+                        f"{cmd_name.upper()} command doesn't exist in Redis commands"
+                    )
+
+        command = self.commands.get(cmd_name)
+        if "movablekeys" in command["flags"]:
+            keys = self._get_moveable_keys(redis_conn, *args)
+        elif "pubsub" in command["flags"] or command["name"] == "pubsub":
+            keys = self._get_pubsub_keys(*args)
+        else:
+            if (
+                command["step_count"] == 0
+                and command["first_key_pos"] == 0
+                and command["last_key_pos"] == 0
+            ):
+                is_subcmd = False
+                if "subcommands" in command:
+                    subcmd_name = f"{cmd_name}|{args[1].lower()}"
+                    for subcmd in command["subcommands"]:
+                        if str_if_bytes(subcmd[0]) == subcmd_name:
+                            command = self.parse_subcommand(subcmd)
+                            is_subcmd = True
+
+                # The command doesn't have keys in it
+                if not is_subcmd:
+                    return None
+            last_key_pos = command["last_key_pos"]
+            if last_key_pos < 0:
+                last_key_pos = len(args) - abs(last_key_pos)
+            keys_pos = list(
+                range(command["first_key_pos"], last_key_pos + 1, command["step_count"])
+            )
+            keys = [args[pos] for pos in keys_pos]
+
+        return keys
+
+    def _get_moveable_keys(self, redis_conn, *args):
+        """
+        NOTE: Due to a bug in redis<7.0, this function does not work properly
+        for EVAL or EVALSHA when the `numkeys` arg is 0.
+         - issue: https://github.com/redis/redis/issues/9493
+         - fix: https://github.com/redis/redis/pull/9733
+
+        So, don't use this function with EVAL or EVALSHA.
+        """
+        pieces = []
+        cmd_name = args[0]
+        # The command name should be splitted into separate arguments,
+        # e.g. 'MEMORY USAGE' will be splitted into ['MEMORY', 'USAGE']
+        pieces = pieces + cmd_name.split()
+        pieces = pieces + list(args[1:])
+        try:
+            keys = redis_conn.execute_command("COMMAND GETKEYS", *pieces)
+        except ResponseError as e:
+            message = e.__str__()
+            if (
+                "Invalid arguments" in message
+                or "The command has no key arguments" in message
+            ):
+                return None
+            else:
+                raise e
+        return keys
+
+    def _get_pubsub_keys(self, *args):
+        """
+        Get the keys from pubsub command.
+        Although PubSub commands have predetermined key locations, they are not
+        supported in the 'COMMAND's output, so the key positions are hardcoded
+        in this method
+        """
+        if len(args) < 2:
+            # The command has no keys in it
+            return None
+        args = [str_if_bytes(arg) for arg in args]
+        command = args[0].upper()
+        keys = None
+        if command == "PUBSUB":
+            # the second argument is a part of the command name, e.g.
+            # ['PUBSUB', 'NUMSUB', 'foo'].
+            pubsub_type = args[1].upper()
+            if pubsub_type in ["CHANNELS", "NUMSUB"]:
+                keys = args[2:]
+        elif command in ["SUBSCRIBE", "PSUBSCRIBE", "UNSUBSCRIBE", "PUNSUBSCRIBE"]:
+            # format example:
+            # SUBSCRIBE channel [channel ...]
+            keys = list(args[1:])
+        elif command == "PUBLISH":
+            # format example:
+            # PUBLISH channel message
+            keys = [args[1]]
+        return keys
diff --git a/redis/parsers/encoders.py b/redis/parsers/encoders.py
new file mode 100644
index 0000000..85fb292
--- /dev/null
+++ b/redis/parsers/encoders.py
@@ -0,0 +1,44 @@
+from redis.typing import EncodableT, EncodedT
+
+from ..exceptions import DataError
+
+
+class Encoder:
+    "Encode strings to bytes-like and decode bytes-like to strings"
+
+    def __init__(self, encoding, encoding_errors, decode_responses):
+        self.encoding = encoding
+        self.encoding_errors = encoding_errors
+        self.decode_responses = decode_responses
+
+    def encode(self, value):
+        "Return a bytestring or bytes-like representation of the value"
+        if isinstance(value, (bytes, memoryview)):
+            return value
+        elif isinstance(value, (int, float)):
+            value = repr(value).encode()
+        elif isinstance(value, str):
+            value = value.encode(self.encoding, self.encoding_errors)
+        elif isinstance(value, bool):
+            # special case bool since it is a subclass of int
+            raise DataError(
+                "Invalid input of type: 'bool'. Convert to a "
+                "bytes, string, int or float first."
+            )
+        elif not isinstance(value, str):
+            # a value we don't know how to deal with. throw an error
+            typename = type(value).__name__
+            raise DataError(
+                f"Invalid input of type: '{typename}'. "
+                f"Convert to a bytes, string, int or float first."
+            )
+        return value
+
+    def decode(self, value, force=False):
+        "Return a unicode string from the bytes-like representation"
+        if self.decode_responses or force:
+            if isinstance(value, memoryview):
+                value = value.tobytes()
+            if isinstance(value, bytes):
+                value = value.decode(self.encoding, self.encoding_errors)
+        return value
diff --git a/redis/parsers/hiredis.py b/redis/parsers/hiredis.py
new file mode 100644
index 0000000..44d5b81
--- /dev/null
+++ b/redis/parsers/hiredis.py
@@ -0,0 +1,217 @@
+import asyncio
+import socket
+from typing import Callable, List, Optional, TypedDict, Union
+
+import async_timeout
+
+from ..exceptions import (
+    AuthenticationError,
+    ConnectionError,
+    InvalidResponse,
+    RedisError,
+)
+from ..typing import EncodableT
+from ..utils import HIREDIS_AVAILABLE
+from .protocol import AsyncBaseParser, BaseParser
+from .socket import (
+    NONBLOCKING_EXCEPTION_ERROR_NUMBERS,
+    NONBLOCKING_EXCEPTIONS,
+    SENTINEL,
+    SERVER_CLOSED_CONNECTION_ERROR,
+)
+
+
+class _HiredisReaderArgs(TypedDict, total=False):
+    protocolError: Callable[[str], Exception]
+    replyError: Callable[[str], Exception]
+    encoding: Optional[str]
+    errors: Optional[str]
+
+
+class HiredisParser(BaseParser):
+    "Parser class for connections using Hiredis"
+
+    def __init__(self, socket_read_size):
+        if not HIREDIS_AVAILABLE:
+            raise RedisError("Hiredis is not installed")
+        self.socket_read_size = socket_read_size
+        self._buffer = bytearray(socket_read_size)
+
+    def __del__(self):
+        try:
+            self.on_disconnect()
+        except Exception:
+            pass
+
+    def on_connect(self, connection, **kwargs):
+        import hiredis
+
+        self._sock = connection._sock
+        self._socket_timeout = connection.socket_timeout
+        kwargs = {
+            "protocolError": InvalidResponse,
+            "replyError": self.parse_error,
+            "errors": connection.encoder.encoding_errors,
+        }
+
+        if connection.encoder.decode_responses:
+            kwargs["encoding"] = connection.encoder.encoding
+        self._reader = hiredis.Reader(**kwargs)
+        self._next_response = False
+
+    def on_disconnect(self):
+        self._sock = None
+        self._reader = None
+        self._next_response = False
+
+    def can_read(self, timeout):
+        if not self._reader:
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+
+        if self._next_response is False:
+            self._next_response = self._reader.gets()
+            if self._next_response is False:
+                return self.read_from_socket(timeout=timeout, raise_on_timeout=False)
+        return True
+
+    def read_from_socket(self, timeout=SENTINEL, raise_on_timeout=True):
+        sock = self._sock
+        custom_timeout = timeout is not SENTINEL
+        try:
+            if custom_timeout:
+                sock.settimeout(timeout)
+            bufflen = self._sock.recv_into(self._buffer)
+            if bufflen == 0:
+                raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+            self._reader.feed(self._buffer, 0, bufflen)
+            # data was read from the socket and added to the buffer.
+            # return True to indicate that data was read.
+            return True
+        except socket.timeout:
+            if raise_on_timeout:
+                raise TimeoutError("Timeout reading from socket")
+            return False
+        except NONBLOCKING_EXCEPTIONS as ex:
+            # if we're in nonblocking mode and the recv raises a
+            # blocking error, simply return False indicating that
+            # there's no data to be read. otherwise raise the
+            # original exception.
+            allowed = NONBLOCKING_EXCEPTION_ERROR_NUMBERS.get(ex.__class__, -1)
+            if not raise_on_timeout and ex.errno == allowed:
+                return False
+            raise ConnectionError(f"Error while reading from socket: {ex.args}")
+        finally:
+            if custom_timeout:
+                sock.settimeout(self._socket_timeout)
+
+    def read_response(self, disable_decoding=False):
+        if not self._reader:
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+
+        # _next_response might be cached from a can_read() call
+        if self._next_response is not False:
+            response = self._next_response
+            self._next_response = False
+            return response
+
+        if disable_decoding:
+            response = self._reader.gets(False)
+        else:
+            response = self._reader.gets()
+
+        while response is False:
+            self.read_from_socket()
+            if disable_decoding:
+                response = self._reader.gets(False)
+            else:
+                response = self._reader.gets()
+        # if the response is a ConnectionError or the response is a list and
+        # the first item is a ConnectionError, raise it as something bad
+        # happened
+        if isinstance(response, (AuthenticationError, ConnectionError)):
+            raise response
+        elif (
+            isinstance(response, list)
+            and response
+            and isinstance(response[0], (AuthenticationError, ConnectionError))
+        ):
+            raise response[0]
+        return response
+
+
+class AsyncHiredisParser(AsyncBaseParser):
+    """PAsync implementation of parser class for connections using Hiredis"""
+
+    __slots__ = AsyncBaseParser.__slots__ + ("_reader",)
+
+    def __init__(self, socket_read_size: int):
+        if not HIREDIS_AVAILABLE:
+            raise RedisError("Hiredis is not available.")
+
+        import hiredis
+
+        super().__init__(socket_read_size=socket_read_size)
+        self._reader: Optional[hiredis.Reader] = None
+
+    def on_connect(self, connection: "Connection"):
+        import hiredis
+
+        self._stream = connection._reader
+        kwargs: _HiredisReaderArgs = {
+            "protocolError": InvalidResponse,
+            "replyError": self.parse_error,
+        }
+        if connection.encoder.decode_responses:
+            kwargs["encoding"] = connection.encoder.encoding
+            kwargs["errors"] = connection.encoder.encoding_errors
+
+        self._reader = hiredis.Reader(**kwargs)
+
+    def on_disconnect(self):
+        self._stream = None
+        self._reader = None
+
+    async def can_read_destructive(self):
+        if not self._stream or not self._reader:
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+        if self._reader.gets():
+            return True
+        try:
+            async with async_timeout.timeout(0):
+                return await self.read_from_socket()
+        except asyncio.TimeoutError:
+            return False
+
+    async def read_from_socket(self):
+        buffer = await self._stream.read(self._read_size)
+        if not buffer or not isinstance(buffer, bytes):
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from None
+        self._reader.feed(buffer)
+        # data was read from the socket and added to the buffer.
+        # return True to indicate that data was read.
+        return True
+
+    async def read_response(
+        self, disable_decoding: bool = False
+    ) -> Union[EncodableT, List[EncodableT]]:
+        if not self._stream or not self._reader:
+            self.on_disconnect()
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from None
+
+        response = self._reader.gets()
+        while response is False:
+            await self.read_from_socket()
+            response = self._reader.gets()
+
+        # if the response is a ConnectionError or the response is a list and
+        # the first item is a ConnectionError, raise it as something bad
+        # happened
+        if isinstance(response, (AuthenticationError, ConnectionError)):
+            raise response
+        elif (
+            isinstance(response, list)
+            and response
+            and isinstance(response[0], (AuthenticationError, ConnectionError))
+        ):
+            raise response[0]
+        return response
diff --git a/redis/parsers/protocol.py b/redis/parsers/protocol.py
new file mode 100644
index 0000000..6faaddd
--- /dev/null
+++ b/redis/parsers/protocol.py
@@ -0,0 +1,107 @@
+from abc import ABC
+from asyncio import StreamReader
+from typing import List, Optional, Union
+
+from redis.typing import EncodableT
+
+from ..exceptions import (
+    AuthenticationError,
+    AuthenticationWrongNumberOfArgsError,
+    BusyLoadingError,
+    ConnectionError,
+    ExecAbortError,
+    ModuleError,
+    NoPermissionError,
+    NoScriptError,
+    ReadOnlyError,
+    ResponseError,
+)
+
+MODULE_LOAD_ERROR = "Error loading the extension. " "Please check the server logs."
+NO_SUCH_MODULE_ERROR = "Error unloading module: no such module with that name"
+MODULE_UNLOAD_NOT_POSSIBLE_ERROR = "Error unloading module: operation not " "possible."
+MODULE_EXPORTS_DATA_TYPES_ERROR = (
+    "Error unloading module: the module "
+    "exports one or more module-side data "
+    "types, can't unload"
+)
+# user send an AUTH cmd to a server without authorization configured
+NO_AUTH_SET_ERROR = {
+    # Redis >= 6.0
+    "AUTH <password> called without any password "
+    "configured for the default user. Are you sure "
+    "your configuration is correct?": AuthenticationError,
+    # Redis < 6.0
+    "Client sent AUTH, but no password is set": AuthenticationError,
+}
+
+
+class BaseParser(ABC):
+
+    EXCEPTION_CLASSES = {
+        "ERR": {
+            "max number of clients reached": ConnectionError,
+            "invalid password": AuthenticationError,
+            # some Redis server versions report invalid command syntax
+            # in lowercase
+            "wrong number of arguments "
+            "for 'auth' command": AuthenticationWrongNumberOfArgsError,
+            # some Redis server versions report invalid command syntax
+            # in uppercase
+            "wrong number of arguments "
+            "for 'AUTH' command": AuthenticationWrongNumberOfArgsError,
+            MODULE_LOAD_ERROR: ModuleError,
+            MODULE_EXPORTS_DATA_TYPES_ERROR: ModuleError,
+            NO_SUCH_MODULE_ERROR: ModuleError,
+            MODULE_UNLOAD_NOT_POSSIBLE_ERROR: ModuleError,
+            **NO_AUTH_SET_ERROR,
+        },
+        "WRONGPASS": AuthenticationError,
+        "EXECABORT": ExecAbortError,
+        "LOADING": BusyLoadingError,
+        "NOSCRIPT": NoScriptError,
+        "READONLY": ReadOnlyError,
+        "NOAUTH": AuthenticationError,
+        "NOPERM": NoPermissionError,
+    }
+
+    def parse_error(self, response):
+        "Parse an error response"
+        error_code = response.split(" ")[0]
+        if error_code in self.EXCEPTION_CLASSES:
+            response = response[len(error_code) + 1 :]
+            exception_class = self.EXCEPTION_CLASSES[error_code]
+            if isinstance(exception_class, dict):
+                exception_class = exception_class.get(response, ResponseError)
+            return exception_class(response)
+        return ResponseError(response)
+
+
+class AsyncBaseParser(BaseParser):
+    """Base parsing class for the python-backed async parser"""
+
+    __slots__ = "_stream", "_read_size"
+
+    def __init__(self, socket_read_size: int):
+        self._stream: Optional[StreamReader] = None
+        self._read_size = socket_read_size
+
+    def __del__(self):
+        try:
+            self.on_disconnect()
+        except Exception:
+            pass
+
+    def on_disconnect(self):
+        raise NotImplementedError()
+
+    def on_connect(self, connection: "Connection"):
+        raise NotImplementedError()
+
+    async def can_read_destructive(self) -> bool:
+        raise NotImplementedError()
+
+    async def read_response(
+        self, disable_decoding: bool = False
+    ) -> Union[EncodableT, ResponseError, None, List[EncodableT]]:
+        raise NotImplementedError()
diff --git a/redis/parsers/resp2.py b/redis/parsers/resp2.py
new file mode 100644
index 0000000..32872ea
--- /dev/null
+++ b/redis/parsers/resp2.py
@@ -0,0 +1,203 @@
+from asyncio import IncompleteReadError, TimeoutError
+from typing import Any, Optional, Union
+
+import async_timeout
+
+from redis.typing import EncodableT
+
+from ..exceptions import ConnectionError, InvalidResponse, RedisError, ResponseError
+from .encoders import Encoder
+from .protocol import AsyncBaseParser, BaseParser
+from .socket import SERVER_CLOSED_CONNECTION_ERROR, SocketBuffer
+
+
+class RESP2Parser(BaseParser):
+    "Plain Python parsing class"
+
+    def __init__(self, socket_read_size):
+        self.socket_read_size = socket_read_size
+        self.encoder = None
+        self._sock = None
+        self._buffer = None
+
+    def __del__(self):
+        try:
+            self.on_disconnect()
+        except Exception:
+            pass
+
+    def on_connect(self, connection):
+        "Called when the socket connects"
+        self._sock = connection._sock
+        self._buffer = SocketBuffer(
+            self._sock, self.socket_read_size, connection.socket_timeout
+        )
+        self.encoder = connection.encoder
+
+    def on_disconnect(self):
+        "Called when the socket disconnects"
+        self._sock = None
+        if self._buffer is not None:
+            self._buffer.close()
+            self._buffer = None
+        self.encoder = None
+
+    def can_read(self, timeout):
+        return self._buffer and self._buffer.can_read(timeout)
+
+    def read_response(self, disable_decoding=False):
+        raw = self._buffer.readline()
+        if not raw:
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+
+        byte, response = raw[:1], raw[1:]
+
+        if byte not in (b"-", b"+", b":", b"$", b"*"):
+            raise InvalidResponse(f"Protocol Error: {raw!r}")
+
+        # server returned an error
+        if byte == b"-":
+            response = response.decode("utf-8", errors="replace")
+            error = self.parse_error(response)
+            # if the error is a ConnectionError, raise immediately so the user
+            # is notified
+            if isinstance(error, ConnectionError):
+                raise error
+            # otherwise, we're dealing with a ResponseError that might belong
+            # inside a pipeline response. the connection's read_response()
+            # and/or the pipeline's execute() will raise this error if
+            # necessary, so just return the exception instance here.
+            return error
+        # single value
+        elif byte == b"+":
+            pass
+        # int value
+        elif byte == b":":
+            response = int(response)
+        # bulk response
+        elif byte == b"$":
+            length = int(response)
+            if length == -1:
+                return None
+            response = self._buffer.read(length)
+        # multi-bulk response
+        elif byte == b"*":
+            length = int(response)
+            if length == -1:
+                return None
+            response = [
+                self.read_response(disable_decoding=disable_decoding)
+                for i in range(length)
+            ]
+        if isinstance(response, bytes) and disable_decoding is False:
+            response = self.encoder.decode(response)
+        return response
+
+
+class AsyncRESP2Parser(AsyncBaseParser):
+    """Parsing class for the RESP2 protocol"""
+
+    __slots__ = AsyncBaseParser.__slots__ + ("encoder",)
+
+    def __init__(self, socket_read_size: int):
+        super().__init__(socket_read_size)
+        self.encoder: Optional[Encoder] = None
+
+    def on_connect(self, connection: "Connection"):
+        """Called when the stream connects"""
+        self._stream = connection._reader
+        if self._stream is None:
+            raise RedisError("Buffer is closed.")
+
+        self.encoder = connection.encoder
+
+    def on_disconnect(self):
+        """Called when the stream disconnects"""
+        if self._stream is not None:
+            self._stream = None
+        self.encoder = None
+
+    async def can_read_destructive(self) -> bool:
+        if self._stream is None:
+            raise RedisError("Buffer is closed.")
+        try:
+            async with async_timeout.timeout(0):
+                return await self._stream.read(1)
+        except TimeoutError:
+            return False
+
+    async def read_response(
+        self, disable_decoding: bool = False
+    ) -> Union[EncodableT, ResponseError, None]:
+        if not self._stream or not self.encoder:
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+        raw = await self._readline()
+        if not raw:
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+        response: Any
+        byte, response = raw[:1], raw[1:]
+
+        if byte not in (b"-", b"+", b":", b"$", b"*"):
+            raise InvalidResponse(f"Protocol Error: {raw!r}")
+
+        # server returned an error
+        if byte == b"-":
+            response = response.decode("utf-8", errors="replace")
+            error = self.parse_error(response)
+            # if the error is a ConnectionError, raise immediately so the user
+            # is notified
+            if isinstance(error, ConnectionError):
+                raise error
+            # otherwise, we're dealing with a ResponseError that might belong
+            # inside a pipeline response. the connection's read_response()
+            # and/or the pipeline's execute() will raise this error if
+            # necessary, so just return the exception instance here.
+            return error
+        # single value
+        elif byte == b"+":
+            pass
+        # int value
+        elif byte == b":":
+            response = int(response)
+        # bulk response
+        elif byte == b"$":
+            length = int(response)
+            if length == -1:
+                return None
+            response = await self._read(length)
+        # multi-bulk response
+        elif byte == b"*":
+            length = int(response)
+            if length == -1:
+                return None
+            response = [
+                (await self.read_response(disable_decoding)) for _ in range(length)
+            ]
+        if isinstance(response, bytes) and disable_decoding is False:
+            response = self.encoder.decode(response)
+        return response
+
+    async def _read(self, length: int) -> bytes:
+        """
+        Read `length` bytes of data.  These are assumed to be followed
+        by a '\r\n' terminator which is subsequently discarded.
+        """
+        if self._stream is None:
+            raise RedisError("Buffer is closed.")
+        try:
+            data = await self._stream.readexactly(length + 2)
+        except IncompleteReadError as error:
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from error
+        return data[:-2]
+
+    async def _readline(self) -> bytes:
+        """
+        read an unknown number of bytes up to the next '\r\n'
+        line separator, which is discarded.
+        """
+        if self._stream is None:
+            raise RedisError("Buffer is closed.")
+        data = await self._stream.readline()
+        if not data.endswith(b"\r\n"):
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+        return data[:-2]
diff --git a/redis/parsers/socket.py b/redis/parsers/socket.py
new file mode 100644
index 0000000..b021051
--- /dev/null
+++ b/redis/parsers/socket.py
@@ -0,0 +1,140 @@
+import errno
+import io
+import socket
+
+from ..exceptions import ConnectionError, TimeoutError
+from ..utils import SSL_AVAILABLE
+
+NONBLOCKING_EXCEPTION_ERROR_NUMBERS = {BlockingIOError: errno.EWOULDBLOCK}
+
+if SSL_AVAILABLE:
+    import ssl
+
+    if hasattr(ssl, "SSLWantReadError"):
+        NONBLOCKING_EXCEPTION_ERROR_NUMBERS[ssl.SSLWantReadError] = 2
+        NONBLOCKING_EXCEPTION_ERROR_NUMBERS[ssl.SSLWantWriteError] = 2
+    else:
+        NONBLOCKING_EXCEPTION_ERROR_NUMBERS[ssl.SSLError] = 2
+
+NONBLOCKING_EXCEPTIONS = tuple(NONBLOCKING_EXCEPTION_ERROR_NUMBERS.keys())
+
+SERVER_CLOSED_CONNECTION_ERROR = "Connection closed by server."
+SENTINEL = object()
+
+
+class SocketBuffer:
+    def __init__(self, socket, socket_read_size, socket_timeout):
+        self._sock = socket
+        self.socket_read_size = socket_read_size
+        self.socket_timeout = socket_timeout
+        self._buffer = io.BytesIO()
+        # number of bytes written to the buffer from the socket
+        self.bytes_written = 0
+        # number of bytes read from the buffer
+        self.bytes_read = 0
+
+    @property
+    def length(self):
+        return self.bytes_written - self.bytes_read
+
+    def _read_from_socket(self, length=None, timeout=SENTINEL, raise_on_timeout=True):
+        sock = self._sock
+        socket_read_size = self.socket_read_size
+        buf = self._buffer
+        buf.seek(self.bytes_written)
+        marker = 0
+        custom_timeout = timeout is not SENTINEL
+
+        try:
+            if custom_timeout:
+                sock.settimeout(timeout)
+            while True:
+                data = self._sock.recv(socket_read_size)
+                # an empty string indicates the server shutdown the socket
+                if isinstance(data, bytes) and len(data) == 0:
+                    raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+                buf.write(data)
+                data_length = len(data)
+                self.bytes_written += data_length
+                marker += data_length
+
+                if length is not None and length > marker:
+                    continue
+                return True
+        except socket.timeout:
+            if raise_on_timeout:
+                raise TimeoutError("Timeout reading from socket")
+            return False
+        except NONBLOCKING_EXCEPTIONS as ex:
+            # if we're in nonblocking mode and the recv raises a
+            # blocking error, simply return False indicating that
+            # there's no data to be read. otherwise raise the
+            # original exception.
+            allowed = NONBLOCKING_EXCEPTION_ERROR_NUMBERS.get(ex.__class__, -1)
+            if not raise_on_timeout and ex.errno == allowed:
+                return False
+            raise ConnectionError(f"Error while reading from socket: {ex.args}")
+        finally:
+            if custom_timeout:
+                sock.settimeout(self.socket_timeout)
+
+    def can_read(self, timeout):
+        return bool(self.length) or self._read_from_socket(
+            timeout=timeout, raise_on_timeout=False
+        )
+
+    def read(self, length):
+        length = length + 2  # make sure to read the \r\n terminator
+        # make sure we've read enough data from the socket
+        if length > self.length:
+            self._read_from_socket(length - self.length)
+
+        self._buffer.seek(self.bytes_read)
+        data = self._buffer.read(length)
+        self.bytes_read += len(data)
+
+        # purge the buffer when we've consumed it all so it doesn't
+        # grow forever
+        if self.bytes_read == self.bytes_written:
+            self.purge()
+
+        return data[:-2]
+
+    def readline(self):
+        buf = self._buffer
+        buf.seek(self.bytes_read)
+        data = buf.readline()
+        while not data.endswith(b"\r\n"):
+            # there's more data in the socket that we need
+            self._read_from_socket()
+            buf.seek(self.bytes_read)
+            data = buf.readline()
+
+        self.bytes_read += len(data)
+
+        # purge the buffer when we've consumed it all so it doesn't
+        # grow forever
+        if self.bytes_read == self.bytes_written:
+            self.purge()
+
+        return data[:-2]
+
+    def purge(self):
+        self._buffer.seek(0)
+        self._buffer.truncate()
+        self.bytes_written = 0
+        self.bytes_read = 0
+
+    def close(self):
+        try:
+            self.purge()
+            self._buffer.close()
+        except Exception:
+            # issue #633 suggests the purge/close somehow raised a
+            # BadFileDescriptor error. Perhaps the client ran out of
+            # memory or something else? It's probably OK to ignore
+            # any error being raised from purge/close since we're
+            # removing the reference to the instance below.
+            pass
+        self._buffer = None
+        self._sock = None