How to use pykka - 10 common examples

"position=%(position)s",
            {
                "rate": segment.rate,
                "format": Gst.Format.get_name(segment.format),
                "start": segment.start,
                "stop": segment.stop,
                "position": segment.position,
            },
        )
        position_ms = segment.position // Gst.MSECOND
        logger.debug("Audio event: position_changed(position=%r)", position_ms)
        AudioListener.send("position_changed", position=position_ms)


# TODO: create a player class which replaces the actors internals
class Audio(pykka.ThreadingActor):

    """
    Audio output through `GStreamer `_.
    """

    #: The GStreamer state mapped to :class:`mopidy.audio.PlaybackState`
    state = PlaybackState.STOPPED

    #: The software mixing interface :class:`mopidy.audio.actor.SoftwareMixer`
    mixer = None

    def __init__(self, config, mixer):
        super().__init__()

        self._config = config
        self._target_state = Gst.State.NULL

def time_it(func):
    start = time.time()
    func()
    print('{!r} took {:.3f}s'.format(func.__name__, time.time() - start))


class SomeObject(object):
    pykka_traversable = False
    cat = 'bar.cat'

    def func(self):
        pass


class AnActor(ThreadingActor):
    bar = SomeObject()
    bar.pykka_traversable = True

    foo = 'foo'

    def __init__(self):
        super(AnActor, self).__init__()
        self.cat = 'quox'

    def func(self):
        pass


def test_direct_plain_attribute_access():
    actor = AnActor.start().proxy()
    for _ in range(10000):

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_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_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

used in tests of the frontends.
"""

from __future__ import absolute_import, unicode_literals

from mopidy import backend
from mopidy.models import Playlist, Ref, SearchResult

import pykka


def create_proxy(config=None, audio=None):
    return DummyBackend.start(config=config, audio=audio).proxy()


class DummyBackend(pykka.ThreadingActor, backend.Backend):

    def __init__(self, config, audio):
        super(DummyBackend, self).__init__()

        self.library = DummyLibraryProvider(backend=self)
        if audio:
            self.playback = backend.PlaybackProvider(audio=audio, backend=self)
        else:
            self.playback = DummyPlaybackProvider(audio=audio, backend=self)
        self.playlists = DummyPlaylistsProvider(backend=self)

        self.uri_schemes = ['dummy']


class DummyLibraryProvider(backend.LibraryProvider):
    root_directory = Ref.directory(uri='dummy:/', name='dummy')

def tearDown(self):  # noqa: N802
        pykka.ActorRegistry.stop_all()

def tearDown(self):  # noqa: N802
        pykka.ActorRegistry.stop_all()
        actor.LocalBackend.libraries = []

def tearDown(self):  # noqa: N802
        pykka.ActorRegistry.stop_all()
        shutil.rmtree(self.temp_dir)