Source code for rodario.registry
""" Actor registry for rodario framework """
# 3rd party
import redis
# local
from rodario.exceptions import RegistrationException
# pylint: disable=C1001
[docs]class _Singleton(object):
""" Singleton for actor registry """
[docs] def __init__(self):
""" Initialize the registry. """
self._redis = redis.StrictRedis()
@property
def actors(self):
"""
Retrieve a list of registered actors.
:rtype: :class:`set`
"""
return self._redis.smembers('actors')
[docs] def register(self, uuid):
"""
Register a new actor.
:param str uuid: The UUID of the actor to register
"""
if self._redis.sadd('actors', uuid) == 0:
raise RegistrationException('Failed adding member to set')
[docs] def unregister(self, uuid):
"""
Unregister an existing actor.
:param str uuid: The UUID of the actor to unregister
"""
self._redis.srem('actors', uuid)
[docs] def exists(self, uuid):
"""
Test whether an actor exists in the registry.
:param str uuid: UUID of the actor to check for
:rtype: :class:`bool`
"""
return self._redis.sismember('actors', uuid) == 1
# pylint: disable=R0201
[docs] def get_proxy(self, uuid):
"""
Return an ActorProxy for the given UUID.
:param str uuid: The UUID to return a proxy object for
:rtype: :class:`rodario.actors.ActorProxy`
"""
# avoid cyclic import
proxy_module = __import__('rodario.actors',
fromlist=('ActorProxy',))
return proxy_module.ActorProxy(uuid=uuid)
# pylint: disable=R0903
[docs]class Registry(object):
""" Actor registry class (singleton wrapper) """
_instance = None
[docs] def __new__(cls):
"""
Retrieve the singleton instance for Registry.
:rtype: :class:`rodario.registry._Singleton`
"""
if not cls._instance:
cls._instance = _Singleton()
return cls._instance