Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_proxy_constructor_raises_exception_if_actor_is_dead(actor_class):
actor_ref = actor_class.start()
actor_ref.stop()
with pytest.raises(ActorDeadError) as exc_info:
ActorProxy(actor_ref)
assert str(exc_info.value) == '{} not found'.format(actor_ref)
def proxy(actor_class):
proxy = ActorProxy(actor_class.start())
yield proxy
proxy.stop()
def replay_events(self, until=None):
while True:
try:
e = self.events.get(timeout=0.1)
cls, event, kwargs = e
if event == until:
break
for actor in self.actor_register:
if isinstance(actor, pykka.ActorProxy):
if isinstance(actor._actor, cls):
actor.on_event(event, **kwargs).get()
else:
if isinstance(actor, cls):
actor.on_event(event, **kwargs)
except Queue.Empty:
# All events replayed.
break
def replay_events(self, until=None):
while True:
try:
e = self.events.get(timeout=0.1)
cls, event, kwargs = e
for actor in self.actor_register:
if isinstance(actor, pykka.ActorProxy):
if isinstance(actor._actor, cls):
actor.on_event(event, **kwargs).get()
else:
actor.on_event(event, **kwargs)
if e[0] == until:
break
except Queue.Empty:
# All events replayed.
break
"""
Wraps the :class:`ActorRef` in an :class:`ActorProxy
`.
Using this method like this::
proxy = AnActor.start().proxy()
is analogous to::
proxy = ActorProxy(AnActor.start())
:raise: :exc:`pykka.ActorDeadError` if actor is not available
:return: :class:`pykka.ActorProxy`
"""
return ActorProxy(self)