Skip to main content

Pubsub system for aio framework

Project description

aio.signals

Pubsub system for the aio asyncio framework

Build status

https://travis-ci.org/phlax/aio.signals.svg?branch=master

Installation

Requires python >= 3.4

Install with:

pip install aio.signals

Quickstart

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!'))

Run with

python my_signals.py

aio.signals usage

Using

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

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Filename, size & hash SHA256 hash help File type Python version Upload date
aio.signals-0.1.3.tar.gz (3.6 kB) Copy SHA256 hash SHA256 Source None May 25, 2015

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page