# Path: aredis/compat.py

# 
# Path: aredis/exceptions.py
# class PubSubError(RedisError):
#     pass
# 
# class ConnectionError(RedisError):
#     pass
# 
# class TimeoutError(RedisError):
#     pass
# 
# Path: aredis/utils.py
# def list_or_args(keys, args):
#     # returns a single list combining keys and args
#     try:
#         iter(keys)
#         # a string or bytes instance can be iterated, but indicates
#         # keys wasn't passed as a list
#         if isinstance(keys, (str, bytes)):
#             keys = [keys]
#     except TypeError:
#         keys = [keys]
#     if args:
#         keys.extend(args)
#     return keys
# 
# def iteritems(x):
#     return iter(x.items())
# 
# def iterkeys(x):
#     return iter(x.keys())
# 
# def nativestr(x):
#     return x if isinstance(x, str) else x.decode('utf-8', 'replace')


import asyncio
import threading
from aredis.compat import CancelledError
from aredis.exceptions import PubSubError, ConnectionError, TimeoutError
from aredis.utils import (list_or_args,
                          iteritems,
                          iterkeys,
                          nativestr)
        """Indicates if there are subscriptions to any channels or patterns"""
        return bool(self.channels or self.patterns)

    async def execute_command(self, *args, **kwargs):
        """Executes a publish/subscribe command"""

        # NOTE: don't parse the response in this function -- it could pull a
        # legitimate message off the stack if the connection is already
        # subscribed to one or more channels

        if self.connection is None:
            self.connection = self.connection_pool.get_connection()
            # register a callback that re-subscribes to any channels we
            # were listening to when we were disconnected
            self.connection.register_connect_callback(self.on_connect)
        connection = self.connection
        await self._execute(connection, connection.send_command, *args)

    async def _execute(self, connection, command, *args):
        try:
            return await command(*args)
        except CancelledError:
            # do not retry if coroutine is cancelled
            if await connection.can_read():
                # disconnect if buffer is not empty in case of error
                # when connection is reused
                connection.disconnect()
            return None
        except (ConnectionError, TimeoutError) as e:
            connection.disconnect()

# Path: aredis/compat.py

# 
# Path: aredis/exceptions.py
# class PubSubError(RedisError):
#     pass
# 
# class ConnectionError(RedisError):
#     pass
# 
# class TimeoutError(RedisError):
#     pass
# 
# Path: aredis/utils.py
# def list_or_args(keys, args):
#     # returns a single list combining keys and args
#     try:
#         iter(keys)
#         # a string or bytes instance can be iterated, but indicates
#         # keys wasn't passed as a list
#         if isinstance(keys, (str, bytes)):
#             keys = [keys]
#     except TypeError:
#         keys = [keys]
#     if args:
#         keys.extend(args)
#     return keys
# 
# def iteritems(x):
#     return iter(x.items())
# 
# def iterkeys(x):
#     return iter(x.keys())
# 
# def nativestr(x):
#     return x if isinstance(x, str) else x.decode('utf-8', 'replace')


# Path: aredis/pubsub.py
import asyncio
import threading
from aredis.compat import CancelledError
from aredis.exceptions import PubSubError, ConnectionError, TimeoutError
from aredis.utils import (list_or_args,
                          iteritems,
                          iterkeys,
                          nativestr)

        """Indicates if there are subscriptions to any channels or patterns"""
        return bool(self.channels or self.patterns)

    async def execute_command(self, *args, **kwargs):
        """Executes a publish/subscribe command"""

        # NOTE: don't parse the response in this function -- it could pull a
        # legitimate message off the stack if the connection is already
        # subscribed to one or more channels

        if self.connection is None:
            self.connection = self.connection_pool.get_connection()
            # register a callback that re-subscribes to any channels we
            # were listening to when we were disconnected
            self.connection.register_connect_callback(self.on_connect)
        connection = self.connection
        await self._execute(connection, connection.send_command, *args)

    async def _execute(self, connection, command, *args):
        try:
            return await command(*args)
        except CancelledError:
            # do not retry if coroutine is cancelled
            if await connection.can_read():
                # disconnect if buffer is not empty in case of error
                # when connection is reused
                connection.disconnect()
            return None
        except (ConnectionError, TimeoutError) as e:
            connection.disconnect()

            if not connection.retry_on_timeout and isinstance(e, TimeoutError):