class ConnectionError(RedisError):
    pass class TimeoutError(RedisError):
    pass class RedisError(Exception):
    pass class ExecAbortError(ResponseError):
    pass class BusyLoadingError(ConnectionError):
    pass class NoScriptError(ResponseError):
    pass class ReadOnlyError(ResponseError):
    pass class ResponseError(RedisError):
    pass class InvalidResponse(RedisError):
    pass class AskError(ResponseError):
    """
    src node: MIGRATING to dst node
        get > ASK error
        ask dst node > ASKING command
    dst node: IMPORTING from src node
        asking command only affects next command
        any op will be allowed after asking command
    """

    def __init__(self, resp):
        """should only redirect to master node"""
        self.args = (resp,)
        self.message = resp
        slot_id, new_node = resp.split(' ')
        host, port = new_node.rsplit(':', 1)
        self.slot_id = int(slot_id)
        self.node_addr = self.host, self.port = host, int(port) class MovedError(AskError):
    pass class TryAgainError(ResponseError):

    def __init__(self, *args, **kwargs):
        pass class ClusterDownError(ClusterError, ResponseError):

    def __init__(self, resp):
        self.args = (resp,)
        self.message = resp class ClusterCrossSlotError(ResponseError):
    message = "Keys in request don't hash to the same slot" def b(x):
    return x.encode('latin-1') if not isinstance(x, bytes) else x def nativestr(x):
    return x if isinstance(x, str) else x.decode('utf-8', 'replace') LOOP_DEPRECATED = sys.version_info >= (3, 8)
import asyncio
import inspect
import os
import socket
import ssl
import sys
import time
import warnings
import aredis.compat
    import hiredis
from io import BytesIO
from aredis.exceptions import (ConnectionError, TimeoutError,
                               RedisError, ExecAbortError,
                               BusyLoadingError, NoScriptError,
                               ReadOnlyError, ResponseError,
                               InvalidResponse, AskError,
                               MovedError, TryAgainError,
                               ClusterDownError, ClusterCrossSlotError)
from aredis.utils import b, nativestr, LOOP_DEPRECATED


async def exec_with_timeout(coroutine, timeout, *, loop=None):
    try:
        if LOOP_DEPRECATED:
            return await asyncio.wait_for(coroutine, timeout)
        else:
            return await asyncio.wait_for(coroutine, timeout, loop=loop)
    except asyncio.TimeoutError as exc:
        raise TimeoutError(exc)


class SocketBuffer:
    def __init__(self, stream_reader, read_size):
        self._stream = stream_reader
        self.read_size = read_size
        self._buffer = 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

    async def _read_from_socket(self, length=None):
        buf = self._buffer
        buf.seek(self.bytes_written)
        marker = 0

        try:
            while True:
                data = await self._stream.read(self.read_size)
                # an empty string indicates the server shutdown the socket
                if isinstance(data, bytes) and len(data) == 0:
                    raise ConnectionError('Socket closed on remote end')
                buf.write(data)
                data_length = len(data)
                self.bytes_written += data_length
                marker += data_length

                if length is not None and length > marker:
                    continue
                break
        except socket.error:
            e = sys.exc_info()[1]
            raise ConnectionError("Error while reading from socket: %s" %
                                  (e.args,))

    async 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:
            await 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]

    async def readline(self):
        buf = self._buffer
        buf.seek(self.bytes_read)
        data = buf.readline()
        while not data.endswith(SYM_CRLF):
            # there's more data in the socket that we need
            await 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:
            # 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


class BaseParser:
    """Plain Python parsing class"""

    EXCEPTION_CLASSES = {
        'ERR': {
            'max number of clients reached': ConnectionError
        },
        'EXECABORT': ExecAbortError,
        'LOADING': BusyLoadingError,
        'NOSCRIPT': NoScriptError,
        'READONLY': ReadOnlyError,
        'ASK': AskError,
        'TRYAGAIN': TryAgainError,

        'MOVED': MovedError,