Pubsub system for aio framework
Pubsub system for the aio asyncio framework
Requires python >= 3.4
pip install aio.signals
The listen function is called synchronously, but the callback listener will be called as a coroutine if it isnt one
The callback listener receives a signal object that has the name of the signal and the object that the signal was emitted with
The emit function is a coroutine
Add the following code to a file my_signals.py
import asyncio from aio.signals import Signals def listener(signal): yield from asyncio.sleep(1) print(signal.data) signals = Signals() signals.listen("my-signal", listener) loop = asyncio.get_event_loop() loop.run_until_complete( signals.emit("my-signal", 'BOOM!'))
>>> import asyncio >>> import aio.testing >>> from aio.signals import Signals
The handler receives a signal object
signal.name is the name of the signal
signal.data contains the emitted object
>>> @asyncio.coroutine ... def callback(signal): ... print("%s received with %s" % (signal.name, signal.data))
>>> @aio.testing.run_until_complete ... def run_test(_signals, name, message): ... yield from _signals.emit(name, message)
>>> signals = Signals() >>> signals.listen("test-signal", callback)
>>> run_test(signals, "test-signal", 'BOOM!') test-signal received with BOOM!
The handler is wrapped in a co-routine if its not one already
>>> def callback(signal): ... yield from asyncio.sleep(1) ... print("%s received with %s" % (signal.name, signal.data))
>>> signals = Signals() >>> signals.listen("test-signal-2", callback)
>>> run_test(signals, "test-signal-2", 'BOOM AGAIN!') test-signal-2 received with BOOM AGAIN!
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
|Filename, size||File type||Python version||Upload date||Hashes|
|Filename, size aio.signals-0.1.3.tar.gz (3.6 kB)||File type Source||Python version None||Upload date||Hashes View|