Source code for rodario.actors.proxy

""" Actor proxy for rodario framework """

# stdlib
import types
import pickle
from multiprocessing import Queue
from threading import Thread
from uuid import uuid4
from time import sleep

# 3rd party
import redis

# local
from rodario.future import Future
from rodario.exceptions import InvalidActorException, InvalidProxyException


class ActorProxy(object):  # pylint: disable=R0903

    """ Proxy object that fires calls to an actor over redis pubsub """

    def __init__(self, actor=None, uuid=None):
        """
        Initialize instance of ActorProxy.

        Accepts either an Actor object to clone or a UUID, but not both.

        :param rodario.actors.Actor actor: Actor to clone
        :param str uuid: UUID of Actor to clone
        """

        #: Redis connection
        self._redis = redis.StrictRedis()
        #: Redis PubSub client
        self._pubsub = None
        #: This proxy object's UUID for creating unique channels
        self.proxyid = str(uuid4())
        #: Dict of response queues for sandboxing method calls
        self._response_queues = {}

        # avoid cyclic import
        actor_module = __import__('rodario.actors', fromlist=('Actor',))
        # pylint: disable=E1123
        self._pubsub = self._redis.pubsub(ignore_subscribe_messages=True)
        self._pubsub.subscribe(**{'proxy:%s' % self.proxyid: self._handler})
        # list of blocking methods
        self._blocking_methods = set()

        methods = set()

        def pubsub_thread():
            """ Call get_message in loop to fire _handler. """

            try:
                while self._pubsub:
                    self._pubsub.get_message()
                    sleep(0.001)
            except:  # pylint: disable=W0702
                pass

        # fire up the message handler thread as a daemon
        proc = Thread(target=pubsub_thread)
        proc.daemon = True
        proc.start()

        if isinstance(actor, actor_module.Actor):
            # proxying an Actor directly
            self.uuid = actor.uuid
            methods = actor._get_methods()  # pylint: disable=W0212
        elif isinstance(uuid, str):
            # proxying by UUID; get actor methods over pubsub
            self.uuid = uuid
            methods = self._proxy('_get_methods').get()
        else:
            raise InvalidProxyException('No actor or UUID provided')

        def get_lambda(name):
            """
            Generate a lambda function to proxy the given method.

            :param str name: Name of the method to proxy
            :rtype: :expression:`lambda`
            """

            return lambda _, *args, **kwargs: self._proxy(name, *args, **kwargs)

        # create proxy methods for each public method of the original Actor
        for name in methods:
            name_split = name.split(':')

            for attr in name_split[1:]:
                if attr == 'blocking':
                    self._blocking_methods.add(name_split[0])

            setattr(self, name_split[0],
                    types.MethodType(get_lambda(name_split[0]), self))

    def _handler(self, message):
        """
        Handle message response via Queue object.

        :param tuple message: The message to dissect
        """

        # throw its value in the associated response queue
        data = pickle.loads(message['data'])
        queue = data[0]
        self._response_queues[queue].put(data[1])
        self._response_queues.pop(queue, None)

    def _proxy(self, method_name, *args, **kwargs):
        """
        Proxy a method call to redis pubsub.

        This method is not meant to be called directly. Instead, it is used
        by the proxy's self-generated methods to provide the proxy with the
        same public API as the actor it represents.

        :param str method_name: The method to proxy
        :param tuple args: The arguments to pass
        :param dict kwargs: The keyword arguments to pass
        :rtype: :class:`multiprocessing.Queue`
        """

        # create a unique response queue for retrieving the return value async
        queue = str(uuid4())
        # fire off the method call to the original Actor over pubsub
        count = self._redis.publish('actor:%s' % self.uuid,
                                    pickle.dumps((self.proxyid, queue,
                                                  method_name, args,
                                                  kwargs,)))

        if count == 0:
            raise InvalidActorException('No such actor')

        self._response_queues[queue] = Queue()

        if method_name in self._blocking_methods:
            return self._response_queues[queue].get()

        return Future(self._response_queues[queue])