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 PyActorTimeoutError(Exception):
    """Wait time expired."""

    def __init__(self, method='Not specified'):
        self.method = method

    def __str__(self):
        return f"Timeout triggered: {self.method!r}" class FutureError(Exception):
    """Some problem with the Future."""

    def __init__(self, value='Not specified'):
        self.value = value

    def __str__(self):
        return f"Future ERROR: {self.value!r}" TELL = 'TELL' FUTURE = 'FUTURE' TYPE = 'TYPE' METHOD = 'METHOD' PARAMS = 'PARAMS' CHANNEL = 'CHANNEL' TO = 'TO' def get_current():
    if core_type == 'thread':
        current = current_thread()
    else:
        current = getcurrent()
    for host in hosts.values():
        if current in host.threads.keys():
            return host.actors[host.threads[current]]
        elif current in host.pthreads.keys():
            return host.actors[host.pthreads[current]] def get_host():
    if core_type == 'thread':
        current = current_thread()
    else:
        current = getcurrent()
    for host in hosts.values():
        if current in host.threads.keys():
            return host
        elif current in host.pthreads.keys():
            return host
    return main_host RPC_ID = 'RPC_ID' RESULT = 'RESULT'
import uuid
from gevent import spawn
from gevent.event import Event
from .channel import Channel
from ..exceptions import PyActorTimeoutError, FutureError
from ..util import TELL, FUTURE, TYPE, METHOD, PARAMS, CHANNEL, TO
from ..util import get_current, get_host, RPC_ID, RESULT
            return self.__get__result()

        self.__condition.wait(timeout)

        if self.__state == FINISHED:
            return self.__get__result()
        else:
            raise PyActorTimeoutError(f"Future: {self.__method!r}")

    def exception(self, timeout=None):
        """
        Return a exception raised by the call that the future
        represents.
        :param timeout: The number of seconds to wait for the exception
            if the future has not been completed. None, the default,
            sets no limit.
        :returns: The exception raised by the call that the future
            represents or None if the call completed without raising.
        :raises: TimeoutError: If the timeout is reached before the
            future ends execution.
        """
        # with self.__condition:
        if self.__state == FINISHED:
            return self.__exception

        self.__condition.wait(timeout)

        if self.__state == FINISHED:
            return self.__exception
        else:
            raise PyActorTimeoutError(f"Future: {self.__method!r}")

    def send_work(self):
        """
        Sends the query to the actor for it to start executing the work.

        It is possible to execute once again a future that has finished
        if necessary (overwriting the results), but only one execution
        at a time.
        """
        if self.__set_running():
            # msg = FutureRequest(FUTURE, self.__method, self.__params,
            #                     self.__channel, self.__target, self.__id)
            msg = {TYPE: FUTURE, METHOD: self.__method, PARAMS: self.__params,
                   CHANNEL: self.__channel, TO: self.__target,
                   RPC_ID: self.__id}
            self.__actor_channel.send(msg)
        else:
            raise FutureError("Future already running.")

    def __set_running(self):
        # """This is only called internally from send_work().
        # It marks the future as running or returns false if it
        # already was running."""
        # with self.__condition:
        if self.__state in [PENDING, FINISHED]:
            self.__condition.clear()
            self.__state = RUNNING
            return True
        elif self.__state == RUNNING:
            return False

    def set_result(self, result):
        """
        Sets the return value of work associated with the future.
        Only called internally.
        """
        # with self.__condition:
        self.__result = result
        self.__state = FINISHED
        self.__condition.set()
        self._invoke_callbacks()

    def set_exception(self, exception):
        """
        Sets the result of the future as being the given exception.
        Only called internally.
        """
        # with self.__condition:
        self.__exception = exception
        self.__state = FINISHED
        self.__condition.set()
        self._invoke_callbacks()


class FutureRef(Future):
    def result(self, timeout=None):
        """
        Returns the result of the call that the future represents.

        :param timeout: The number of seconds to wait for the result
            if the future has not been completed. None, the default,
            sets no limit.
        :returns: The result of the call that the future represents.
        :raises: TimeoutError: If the timeout is reached before the
            future ends execution.
        :raises: Exception: If the call raises the Exception.
        """
        result = super(FutureRef, self).result(timeout)
        return get_host().loads(result)


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]