class Channel(Queue):
    """
    Channel is the main communication mechanism between actors. It is
    actually a simple facade to the gevent.Queue class.
    """

    def __init__(self):
        Queue.__init__(self)

    def send(self, msg):
        """
        It sends a message to the current channel.

        :param msg: The message sent to an actor. It is a dictionary
            using the constants in util.py (:mod:`pyactor.util`).
        """
        self.put(msg)

    def receive(self, timeout=None):
        """
        It receives a message from the channel, blocking the calling
        thread until the response is received, or the timeout is
        triggered.

        :param int timeout: timeout to wait for messages. If none
            provided it will block until a message arrives.
        :return: returns a message sent to the channel. It is a
            dictionary the constants in util.py (:mod:`pyactor.util`).
        """
        return self.get(timeout=timeout) class FutureManager(object):
    """
    A manager that controls the creation and execution of the futures in a host.
    """

    def __init__(self):
        self.running = False
        self.channel = Channel()
        self.futures = {}
        self.t = None

    def __queue_management(self):
        self.running = True
        while self.running:
            response = self.channel.receive()
            if response == 'stop':
                self.running = False
            else:
                result = response[RESULT]
                future = self.futures[response[RPC_ID]]
                if isinstance(result, Exception):
                    future.set_exception(result)
                else:
                    future.set_result(result)

    def new_future(self, future_ref, ref=False):
        future_id = str(uuid.uuid4())
        if not ref:
            future = Future(future_id, future_ref, self.channel)
        else:
            future = FutureRef(future_id, future_ref, self.channel)
        future.send_work()
        self.futures[future_id] = future

        if not self.running:
            self.t = spawn(self.__queue_management)
        return future

    def stop(self):
        self.channel.send('stop')
        if self.t is not None:
            self.t.join()
            self.t = None
        self.futures = {} ASK = 'ASK' TELL = 'TELL' FUTURE = 'FUTURE' TYPE = 'TYPE' ASK_RESPONSE = 'ASK_RESPONSE' FUTURE_RESPONSE = 'FUTURE_RESPONSE' METHOD = 'METHOD' PARAMS = 'PARAMS' RESULT = 'RESULT' CHANNEL = 'CHANNEL' RPC_ID = 'RPC_ID' def ref_l(self, f):
    def wrap_ref_l(*args):
        new_args = list(args)
        try:
            if new_args[0][METHOD] in self._ref:
                new_args[0][PARAMS] = get_host().loads(list(args[0][PARAMS]))
            return f(*new_args)
        except HostError:
            pass
            # If there is a problem deserializing the params, the method
            # is not executed.
    return wrap_ref_l def ref_d(self, f):
    def wrap_ref_d(*args):
        new_args = list(args)
        if new_args[1][METHOD] in self._ref:
            new_args[0] = get_host().dumps(args[0])
        return f(*new_args)
    return wrap_ref_d
from copy import copy
from gevent import spawn
from .channel import Channel
from .future import FutureManager
from ..util import ASK, TELL, FUTURE, TYPE, ASK_RESPONSE, FUTURE_RESPONSE
from ..util import METHOD, PARAMS, RESULT, CHANNEL, RPC_ID
from ..util import ref_l, ref_d
    URL where it is located, the communication :class:`~.Channel` and
    the class of the actor as also the synchronous and asynchronous
    methods the class implements. When no channel is specified a new
    one will be created which is also the default procedure.

    .. note:: This is a superclass of :py:class:`Actor` and has no
        direct functionality.

    """

    def __init__(self, url, klass, channel=None):
        self.url = url
        self.tell = set()
        self.ask = set()
        self.klass = klass
        if channel:
            self.channel = channel
        else:
            self.channel = Channel()
        if hasattr(klass, '_tell') and klass._tell:
            self.tell = copy(klass._tell)
        if hasattr(klass, '_ask') and klass._ask:
            self.ask = copy(klass._ask)

        if hasattr(klass, '_ref'):
            self.receive = ref_l(self, self.receive)
            self.send_response = ref_d(self, self.send_response)

            self.tell_ref = self.tell & klass._ref
            self.ask_ref = self.ask & klass._ref
            for method in self.ask_ref:
                self.ask.remove(method)
            for method in self.tell_ref:
                self.tell.remove(method)
        else:
            self.ask_ref = set()
            self.tell_ref = set()

        self.tell.add('stop')

    def receive(self, msg):
        raise NotImplementedError()

    def send_response(self, result, msg):
        raise NotImplementedError()

    @property
    def _ref(self):
        return self.tell_ref | self.ask_ref

    def __str__(self):
        return f"Actor {self.url} ({self.klass.__name__})"

    def __repr__(self):
        return f"Actor(url={self.url}, class={self.klass})"


class Actor(ActorRef):
    """
    Actor is the instance of an object to which is possible to access
    and invoke its methods remotely. Main element of the model. The
    host is the one to create them (spawning -> see :meth:`~.spawn`).

    :param str. url: URL where the actor is running.
    :param class klass: class type for the actor.
    :param klass obj: instance of the *klass* class to attach to the
        actor.
    """

    def __init__(self, url, klass, obj):
        super(Actor, self).__init__(url, klass)
        self._obj = obj
        self.id = obj.id
        self.running = True
        self.thread = None
        self.future_manager = FutureManager()

    def __process_queue(self):
        while self.running:
            message = self.channel.receive()
            self.receive(message)

    def is_alive(self):
        """
        :return: (*bool.*) identifies the current state of the actor.
            **True** if it is running.
        """
        return self.running

    def receive(self, msg):
        """
        The message received from the queue specifies a method of the
        class the actor represents. This invokes it. If the
        communication is an ASK, sends the result back
        to the channel included in the message as an ASK_RESPONSE.

        If it is a FUTURE, generates a FUTURE_RESPONSE
        to send the result to the manager.

        :param msg: The message is a dictionary using the constants
            defined in util.py (:mod:`pyactor.util`).
        """
        if msg[TYPE] == TELL and msg[METHOD] == 'stop':
            self.running = False
            self.future_manager.stop()
        else:
            try:
                invoke = getattr(self._obj, msg[METHOD])
                params = msg[PARAMS]
                result = invoke(*params[0], **params[1])
            except Exception as e:
                if msg[TYPE] == TELL:
                    print(e)
                    return
                result = e
            self.send_response(result, msg)

    def send_response(self, result, msg):
        if msg[TYPE] == ASK:
            response = {TYPE: ASK_RESPONSE, RESULT: result,

                        RPC_ID: msg[RPC_ID] if RPC_ID in msg.keys() else None}