I’ve just implemented my own, extremely simple event manager.
Highlighted source: bpaste.net/show/24313/
import types
class Event(object):
"""Base class for all event classes. You can add methods and attributes in
derived classes.
"""
pass
class Sender(object):
"""Use instances of this to send events to a messenger.
Preferred way of instancing this class is through the LightMessenger class.
"""
def __init__(self, messenger):
self.messenger = messenger
def send(self, event):
"""Dispatch an event.
Arguments:
event -- can be either a string or an instance of Event.
"""
self.messenger._send(event)
class Reciever(object):
"""Use instances of this to recieve events.
Preferred way of instancing this class is through the LightMessenger class.
WARNING: multiple calls to accept() with the same event parameter will
result in multiple functions assigned.
"""
def __init__(self, messenger):
self.messenger = messenger
def accept(self, event, function):
"""Listen for an event and assign a function to be called.
Arguments:
event -- can be either a string, the Event class, or an Event derived
class.
function -- can be anything callable. For extra arguments use lambda.
"""
self.messenger._accept(event, function)
def ignore(self, event):
"""Do no longer listen for this event.
Arguments:
event -- reverts a previously attached function(s) from that event.
"""
self.messenger._ignore(event)
class LightMessenger(object):
"""Easy and lightweight event manager. Every single messenger instance is
independent of others. Unlike Panda's internal messenger this is not
prepared to be used in a multithreading environment.
Example usage:
>>> from __future__ import print_function
>>> messenger = LightMessenger()
>>> rec = messenger.getReciever()
>>> sender = messenger.getSender()
>>> reciever.accept("custom event", lambda arg: print(arg))
>>> sender.send("custom event")
custom event
>>> reciever.accept(Event, lambda arg: print(arg))
>>> sender.send(Event())
<lightmessenger.Event object at 0x7f63f745e790>
"""
def __init__(self):
self.listeners_actions = {}
def _accept(self, event, func):
"""This method should be called by recievers."""
if self.listeners_actions.has_key(event):
self.listeners_actions[event] += [func]
else:
self.listeners_actions[event] = [func]
def _ignore(self, event):
"""This method should be called by recievers."""
if event in self.listeners_actions:
del self.listeners_actions[event]
def _send(self, event):
"""This method should be called by senders."""
if isinstance(event, types.StringTypes) and \
event in self.listeners_actions:
for func in self.listeners_actions[event]:
func.__call__(event)
elif isinstance(event, Event) and \
event.__class__ in self.listeners_actions:
for func in self.listeners_actions[event.__class__]:
func.__call__(event)
def getReciever(self):
"""Returns a reciever object that is intended for requesting new
listeners.
"""
return Reciever(self)
def getSender(self):
"""Returns a sender object that is intended for sending events to this
messenger.
"""
return Sender(self)
def reset(self):
"""Resets this messenger to its initial state. All recievers and senders
stay in tact.
"""
self.listeners_actions = {}