def _handle_failure(self, exception_type, exception_value, traceback):
     """Logs unexpected failures, unregisters and stops the actor."""
     logger.error(
         'Unhandled exception in {}:'.format(self),
         exc_info=(exception_type, exception_value, traceback),
     )
     ActorRegistry.unregister(self.actor_ref)
     self.actor_stopped.set()

 def test_broadcast_sends_message_to_all_actors_of_given_class(self):
     ActorRegistry.broadcast({'command': 'foo'}, target_class=self.AnActor)
     for actor_ref in ActorRegistry.get_by_class(self.AnActor):
         received_messages = actor_ref.proxy().received_messages.get()
         self.assert_({'command': 'foo'} in received_messages)
     for actor_ref in ActorRegistry.get_by_class(self.BeeActor):
         received_messages = actor_ref.proxy().received_messages.get()
         self.assert_({'command': 'foo'} not in received_messages)

    def _actor_loop(self):
        """
        The actor's event loop.

        This is the method that will be executed by the thread or greenlet.
        """
        try:
            self.on_start()
        except Exception:
            self._handle_failure(*sys.exc_info())

        while not self.actor_stopped.is_set():
            envelope = self.actor_inbox.get()
            try:
                response = self._handle_receive(envelope.message)
                if envelope.reply_to is not None:
                    envelope.reply_to.set(response)
            except Exception:
                if envelope.reply_to is not None:
                    logger.info(
                        'Exception returned from {} to caller:'.format(self),
                        exc_info=sys.exc_info(),
                    )
                    envelope.reply_to.set_exception()
                else:
                    self._handle_failure(*sys.exc_info())
                    try:
                        self.on_failure(*sys.exc_info())
                    except Exception:
                        self._handle_failure(*sys.exc_info())
            except BaseException:
                exception_value = sys.exc_info()[1]
                logger.debug(
                    '{!r} in {}. Stopping all actors.'.format(
                        exception_value, self
                    )
                )
                self._stop()
                ActorRegistry.stop_all()

        while not self.actor_inbox.empty():
            envelope = self.actor_inbox.get()
            if envelope.reply_to is not None:
                if isinstance(envelope.message, messages._ActorStop):
                    envelope.reply_to.set(None)
                else:
                    envelope.reply_to.set_exception(
                        exc_info=(
                            ActorDeadError,
                            ActorDeadError(
                                '{} stopped before handling the message'.format(
                                    self.actor_ref
                                )
                            ),
                            None,
                        )
                    )

 def test_actor_may_be_unregistered_multiple_times_without_error(self):
     ActorRegistry.unregister(self.ref)
     self.assert_(self.ref not in ActorRegistry.get_all())
     ActorRegistry.unregister(self.ref)
     self.assert_(self.ref not in ActorRegistry.get_all())
     ActorRegistry.register(self.ref)
     self.assert_(self.ref in ActorRegistry.get_all())

 def _stop(self):
     """
     Stops the actor immediately without processing the rest of the inbox.
     """
     ActorRegistry.unregister(self.actor_ref)
     self.actor_stopped.set()
     logger.debug('Stopped {}'.format(self))
     try:
         self.on_stop()
     except Exception:
         self._handle_failure(*sys.exc_info())

def test_broadcast_sends_message_to_all_actors_if_no_target(
    a_actor_refs, b_actor_refs
):
    ActorRegistry.broadcast({'command': 'foo'})

    running_actors = ActorRegistry.get_all()
    assert running_actors

    for actor_ref in running_actors:
        received_messages = actor_ref.proxy().received_messages.get()
        assert {'command': 'foo'} in received_messages

    def test_fail(self):   # pylint: disable=R0201
        """ Test closing stream when fail. """
        def buggy_write(_):
            """ A buggy writable """
            raise ValueError()
        c = Collector.start(self.controller,
                            flexmock(write=buggy_write)
                            .should_receive('close').once().mock())

        c.tell(Record(None, 'Fail'))
        ActorRegistry.stop_all()

def test_broadcast_sends_message_to_all_actors_of_given_class_name(
    actor_a_class, actor_b_class
):
    ActorRegistry.broadcast({'command': 'foo'}, target_class='ActorA')

    for actor_ref in ActorRegistry.get_by_class(actor_a_class):
        received_messages = actor_ref.proxy().received_messages.get()
        assert {'command': 'foo'} in received_messages

    for actor_ref in ActorRegistry.get_by_class(actor_b_class):
        received_messages = actor_ref.proxy().received_messages.get()
        assert {'command': 'foo'} not in received_messages

def test_all_actors_are_stopped_on_base_exception(events, actor_ref):
    assert len(ActorRegistry.get_all()) == 1
    assert not events.on_stop_was_called.is_set()

    actor_ref.tell({'command': 'raise base exception'})

    events.on_stop_was_called.wait(5)
    assert events.on_stop_was_called.is_set()
    assert len(ActorRegistry.get_all()) == 0

    events.on_stop_was_called.wait(5)
    assert events.on_stop_was_called.is_set()
    assert len(ActorRegistry.get_all()) == 0

    def test_work(self):
        class Worker(object):
            def __init__(self, arg):
                self.arg = arg

            def work_on(self, task):
                print task
                return task

        flexmock(Worker).should_call('work_on').once().with_args(1)
        cr = Crawler.start(self.controller, self.tasksource, self.collector,
                           Worker, self.initargs)
        cr.tell(Task(None, 1))
        cr.stop()
        ActorRegistry.stop_all()

 def send_event(self, event_name, data, player=None, client=None):
     if player:
         data['player'].update(player.get_stats_data())
     if client:
         c = data.get('client', {})
         c['device_type'] = client.device_type
         c['conn_type'] = client.conn_type
         c['client_timestamp'] = client.client_timestamp
         c['client_version'] = client.client_version
         c['device_os'] = client.device_os
         c['device_os_version'] = client.device_os_version
         c['openudid'] = client.openudid
         c['odin'] = client.odin
         c['mac_address'] = client.mac_address
         c['ios_ifa'] = client.ios_ifa
         c['android_id'] = client.android_id
         c['imei'] = client.imei
         c['android_ifa'] = client.android_ifa
         data['client'] = c
     msg = {"event_name": event_name,
            "timestamp": time.time(),
            "data": data}
     actor = ActorRegistry.get_by_class(StatsActor)
     actor = actor and actor[0] or StatsActor.start()
     actor.tell(msg)

    def test_stop_all_stops_last_started_actor_first_if_blocking(
            self, mock_method):
        stopped_actors = []
        started_actors = [mock.Mock(name=i) for i in range(3)]
        started_actors[0].stop.side_effect = lambda *a, **kw: \
            stopped_actors.append(started_actors[0])
        started_actors[1].stop.side_effect = lambda *a, **kw: \
            stopped_actors.append(started_actors[1])
        started_actors[2].stop.side_effect = lambda *a, **kw: \
            stopped_actors.append(started_actors[2])
        ActorRegistry.get_all.return_value = started_actors

        ActorRegistry.stop_all(block=True)

        self.assertEqual(stopped_actors[0], started_actors[2])
        self.assertEqual(stopped_actors[1], started_actors[1])
        self.assertEqual(stopped_actors[2], started_actors[0])

 def __init__(self):
     super(SleepyElder, self).__init__()
     self._lair = ActorRegistry.get_by_class_name('SpiderLair')[0]
     self._lair_command = {
         'command': 'kick_all',
         'data': None
     }
     self._work = True

 def get_pactor_ref(self):
     """Get a reference to a pykka actor for this."""
     if not self.is_active():
         raise NotRunningError("This game is not active.")
     else:
         actor_urn = cache.get(str(self.uuid))
         actor_ref = ActorRegistry.get_by_urn(actor_urn)
     return actor_ref

 def test_write(self):  # pylint: disable=R0201
     """ Test writing to a writable.  """
     mock_open = flexmock(sys.modules['__builtin__'])
     mock_open.should_call('open')
     (mock_open.should_receive('open')
      .with_args('newfile', 'wb')
      .and_return(
          flexmock(write=lambda x: None)
          .should_receive('write').with_args('READY').once().mock()
          .should_receive('flush').once().mock()
          .should_receive('close').once().mock()
          )
      )
     wr = FileWriter('newfile')
     c = Collector.start(self.controller, wr)
     c.tell(Record(None, 'READY'))
     ActorRegistry.stop_all()

def test_actors_may_be_looked_up_by_class_name(
    actor_a_class, a_actor_refs, b_actor_refs
):
    result = ActorRegistry.get_by_class_name('ActorA')

    for a_actor in a_actor_refs:
        assert a_actor in result
    for b_actor in b_actor_refs:
        assert b_actor not in result

    def test_actor_processes_all_messages_before_stop_on_self_stops_it(self):
        self.actor_ref.ask({'command': 'message self then stop'})

        self.greetings_was_received.wait(5)
        self.assertTrue(self.greetings_was_received.is_set())

        self.on_stop_was_called.wait(5)

        self.assertEqual(0, len(ActorRegistry.get_all()))

 def is_active(self):
     actor_urn = cache.get(str(self.uuid))
     if actor_urn:
         if ActorRegistry.get_by_urn(actor_urn):
             return True
         else:
             cache.delete(str(self.uuid))
             return False
     else:
         return False

def test_actor_processes_all_messages_before_stop_on_self_stops_it(
    actor_ref, events
):
    actor_ref.ask({'command': 'message self then stop'})

    events.greetings_was_received.wait(5)
    assert events.greetings_was_received.is_set()

    events.on_stop_was_called.wait(5)
    assert len(ActorRegistry.get_all()) == 0

def test_actor_is_stopped_when_unhandled_exceptions_are_raised(
    actor_ref, events
):
    assert not events.on_failure_was_called.is_set()

    actor_ref.tell({'command': 'raise exception'})

    events.on_failure_was_called.wait(5)
    assert events.on_failure_was_called.is_set()
    assert len(ActorRegistry.get_all()) == 0