SignalManager - light weight signal management

signal_manager.py

"""
signal_manager.py - SignalManager - light weight signal management

Extremely light weight.  No enforcement of signal arguments, or
even explicit listing of which signals exist.

Terry Brown, [email protected], Thu Mar 23 21:13:38 2017
"""

from collections import defaultdict

class SignalData:
    def __init__(self):
        self.listeners = defaultdict(lambda: list())
        self.locked = False

class MsgSignalHandled:
    """A listener can return SignalManager.MsgSignalHandled to prevent
    other listeners from being called
    """
    pass

def _signal_setup(obj):
    if not hasattr(obj, '_signal_data'):
        obj._signal_data = SignalData()

def signal_emit(source, signal_name, *args, **kwargs):
    """Emit signal to all listeners"""
    if not hasattr(source, '_signal_data'):
        return
    if '_sig_lock' in kwargs:
        obj_to_lock = kwargs.pop('_sig_lock')
        _signal_setup(obj_to_lock)
        obj_to_lock._signal_data.locked = True
    else:
        obj_to_lock = None

    for listener in source._signal_data.listeners[signal_name]:
        response = listener(*args, **kwargs)
        if response == MsgSignalHandled:
            break

    if obj_to_lock is not None:
        obj_to_lock._signal_data.locked = False

def signal_connect(source, signal_name, listener):
    """Connect to signal"""
    _signal_setup(source)
    source._signal_data.listeners[signal_name].append(listener)

def signal_is_locked(obj):
    return hasattr(obj, '_signal_data') and obj._signal_data.locked

class SignalManager(object):
    """SignalManager - light weight signal management mixin."""

    def signal_emit(self, signal_name, *args, **kwargs):
        """Emit signal to all listeners"""
        signal_emit(self, signal_name, *args, **kwargs)
    def signal_connect(self, signal_name, listener):
        """Connect to signal"""
        signal_connect(self, signal_name, listener)

def main():
    """test of SignalManager"""

    # simple use

    class Emitter(SignalManager):
        def some_emission(self):
            self.signal_emit('the_emission', 12, [1,2,3])
    
    def hear_emit(n, l):
        print("Got %s %s" % (n,l))
    
    emitter = Emitter()
    emitter.signal_connect('the_emission', hear_emit)
    emitter.some_emission()

    # use with proxy and locking

    class Tester:
        def __init__(self, name, relay):
            self.name = name
            self.relay = relay
            signal_connect(self.relay, 'work_done', self.check_work)
        def do_work(self):
            signal_emit(self.relay, 'work_done', 4.2, animal='otters', _sig_lock=self)
        def check_work(self, num, animal='eels'):
            if signal_is_locked(self):
                return
            print("%s heard about %s %s" % (self.name, num, animal))

    class SomeProxy:
        """Like a public notice board"""
        pass

    relay = SomeProxy()
    a = Tester('A', relay)
    a.do_work()
    b = Tester('B', relay)
    a.do_work()
    b.do_work()

if __name__ == '__main__':
    main()