Skip to content

@pcapriotti /

Embed URL


Subversion checkout URL

You can clone with
Download ZIP
Reactor Framework
"""The reactor framework.
This module introduces the *reactor framework*, a collection of utilities to be
used in conjunction with the greelet library to solve the problem of inversion
of control in event-driven code.
Traditionally, writing event-driven code typically consists of "connecting"
signals to handlers (i.e. callbacks), which are to be invoked by the framework
in use when a certain "event" occurs.
As long as the logic being implemented is simple enough to be easily expressed
in the form of "reactions" to specified events, this model is pretty natural
and straightforward to implement. An example is an application displaying
several buttons, each of which performs a different non-interactive operation.
However, more often than not, the desired reaction to user input depends not
only on the input itself, but also on whatever input was already provided,
either by the user, or by some other previous operations that is now concluded.
In that case, the logic detailing the desired behavior can usually be easily
expressed in a linear fashion, but that doesn't directly translate into the
traditional event-driven model, and it often requires a translation into the
equivalent of a state machine representation or something to that effect, to be
able to be implemented correctly and reliably.
For example, suppose we want to implement a button which will display a message
only after it has been clicked 5 times, and do nothing otherwise.
In the usual event-driven style of programming, we need to introduce a global
state containing the number of times the button has been clicked, increment it,
and check if it reached the desired count at each invocation of the handler.
Using the Qt API as an example, the handler and initialization function would
probably look like this:
def __init__(self):
self.count = 0
def on_button_clicked(self):
self.count += 1
if self.count == 5:
print "time is a face on the water"
This is a simple enough example that it doesn't look too bad, but still it's
easy to spot the signs that make such code so hard to follow and maintain when
the complexity is increased even so slightly.
It has all the components of a straightforward for loop (initializing a
counter, incrementing it at each step, checking for termination), but the
iteration itself is hidden in the framework main loop, and we cannot just use
the normal python iteration techniques, because the execution of the loop is
not contained within a single function frame, but is spread throughout multiple
function invocations.
The reactor framework solves this problem with the use of greenlet "threads".
When you use wait_for to connect to a signal, for example, control is
immediately returned to the calling function, but resumed to the point where it
left as soon as the signal is emitted.
This makes the code look like a blocking wait on the signal but without any of
the disadvantages, like actually blocking the main loop, resulting in a frozen
For comparison, here is how the previous example could have been implemented
using the reactor framework:
def __init__(self):
for i in xrange(5):
print "time is a face on the water"
The reactor framework is inspired by similar libraries available for other
languages and frameworks [1] [2], as well as a number of research articles on
the subject of solving the inversion of control problem with cooperative
multitasking [3] [4].
from greenlet import greenlet, getcurrent as gself
from decorator import decorator
def react(f):
"""Run the given function f in a new greenlet.
Convenience function to start a new reactive context.
Since wait_for or reactive_handler cannot be used inside the same greenlet
as the main loop, a basic use of this function is to execute the
application main routine, so that it run in its own greenlet, and is able
to take advantage of reactive event handling.
More generally, spawning a new greenlet is useful to make sure handling of
certain signals is performed independently of others.
For example, a code like the following:
print "button1 clicked"
print "button2 clicked"
will only react to presses of button1 and button2 in that specific order, whereas
wait_for(button1.clicked, button2.clicked)
print "button1 or button2 clicked"
will react equally to a click on either of the buttons. If we want to react
to buttons in either order, we can use something like the following:
def handle_button(button, name):
print name, "clicked"
handle_button(button1, "button1")
handle_button(button2, "button2")
def reactive(f, *args):
"""Modify a function to spawn a new greenlet when called.
Whenever the wrapped function is called, a new greenlet is spawned and the
function is executed in it.
Note that by using this decorator, you lose the ability to access the
return value of the wrapped function.
react(lambda: f(*args))
def wait_for(*signals, **kwargs):
"""Block the current greenlet on the given signals.
This function is used to interrupt the current greenlet execution until one
of the specified signals is emitted. When that happens, execution is
resumed immediately after the function call, which returns
It is possible to specify a list of extra signals that result in
exceptions, by adding a named parameter 'exceptions' containing a list of
tuples of the form
(signal_name, exception_class).
If a signal contained in that list is emitted first, the corresponding
exception class is instantiated using the signal arguments as parameters,
and raised in the greenlet of the caller after it resumes.
w = waiting_for(*signals, **kwargs)
with w:
return normalize_result(w.result)
class waiting_for(object):
"""A variation of wait_for to be used in a with statement.
Sometimes, we want to make sure we are listening to a certain signal before
performing an operation, but wait_for doesn't allow the possibily of
executing code between when the signal is connected, and when it is
This limitation is addressed by using waiting_for inside a with statement.
For example:
with waiting_for(message.sent):
print "message sent"
would set up a handler for the message.sent signal, then execute the code
inside the with statement (i.e. send the message), and finally resume
execution of the parent greenlet (e.g. the application main loop).
Eventually, When the signal is emitted, the print statement is executed.
As with wait_for, optional 'exceptional' signals can be specified, that
will result in a raised exception when emitted.
def __init__(self, *signals, **kwargs):
self.signals = signals
self.exceptions = kwargs.get("exceptions", [])
self.result = None
self.exception = None
def __enter__(self): = gself()
for signal in self.signals:
for signal, exc in self.exceptions:
return self
def __exit__(self, type, value, traceback):
if not self.exception is None:
raise self.exception
def disconnect_all(self):
for signal in self.signals:
for signal, _ in self.exceptions:
def handler(self, *args):
self.result = normalize_result(args) = gself()
def exception_handler(self, *args):
self.exception = exc(*args)
def events(signal):
"""Keep handling a signal indefinitely.
A common signal handling scenario is the need to perform some operation
every time the signal is emitted.
This is achieved normally without the use of reactor framework by simply
connecting the signal to a handler performing the desired operation.
However, there is a simple way to achieve the same result within the
reactor framework by iterating through the generator returned by the events
function as though they were all already available. The reactor framework
takes care of returning control to the main loop after each iteration, and
resuming the loop whenever a new message is available.
There is a number of of reasons why this approach might be preferable to
the direct one:
- consistency
- read/write access to the local frame
- ability to yield from within the handler code
- ability to manipulate the yield generator via combinators or comprehension
For example, the following code:
def incoming_messages(self):
return (e.message for e in events(self.message_received))
will return a generator containing all future incoming messages.
event_args = []
current = gself()
def handler(*args):
event_args[:] = args
current.parent = gself()
while True:
yield normalize_result(event_args)
class reactive_handler(object):
"""Adaptor to be used as a callback to non-reactive asynchronous function calls.
Many frameworks and libraries expose functions that take one or more
callbacks as arguments, and use them to notify the caller when the
operation is finished and what the result is.
Using a reactive_handler instance inside a with statement allows such a
function to be called within the reactor framework. The function will
execute and return to the main loop, and execution will resume after the
call only when the handler is actually invoked.
The handler may be used multiple times within the with statement, but can
only be called from a different greenlet.
def __init__(self, exception=None):
self.exception = exception
self.result = None
self.base = self
def __enter__(self): = gself()
return self
def __exit__(self, type, value, traceback):
if type is None:
if is None:
raise Exception("reactive_handler cannot be used from the root greenlet")
if self.exception: raise self.exception(*self.result)
return False
def __call__(self, *args):
self.base.result = normalize_result(args)
self.base.exception = self.exception
if gself() ==
raise Exception("reactive_handler invoked in the same greenlet where it was created") = gself()
"""Create a special handler which will result in a raised exception when invoked.
When a function takes a handler to be invoked in case of error, the handler
returned by this function can be used. The specified exception class will
be instantiated by the handler and raised after the greenlet is resumed.
def for_error(self, exception=None):
result = self.__class__(exception)
result.base = self
return result
class SignalAdaptor(object):
"""Adaptor for dbus-like signals.
The signals passed to the reactor framework are assumed to have an interface compatible with that of Qt signals in PySide.
This adaptor class allows the use of dbus signals in the reactor framework.
def __init__(self, obj, name):
self.obj = obj = name
self.connection = None
def connect(self, handler):
if self.connection:
raise Exception("SignalAdaptor does not support more than 1 connection")
self.connection = self.obj.connect_to_signal(, handler)
def disconnect(self, handler):
if self.connection:
def normalize_result(result):
if len(result) == 1:
return result[0]
elif len(result) == 0:
return None
return result
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Something went wrong with that request. Please try again.