/usr/share/pyshared/coherence/extern/louie.py is in python-coherence 0.6.6.2-6.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | """
Wrapper module for the louie implementation
"""
import warnings
from coherence.dispatcher import Dispatcher
class Any(object): pass
class All(object): pass
class Anonymous(object): pass
# fake the API
class Dummy(object): pass
signal = Dummy()
sender = Dummy()
#senders
sender.Anonymous = Anonymous
sender.Any = Any
#signals
signal.All = All
# a slightly less raise-y-ish implementation as louie was not so picky, too
class GlobalDispatcher(Dispatcher):
def connect(self, signal, callback, *args, **kw):
if not signal in self.receivers:
# ugly hack
self.receivers[signal] = []
return Dispatcher.connect(self, signal, callback, *args, **kw)
def _get_receivers(self, signal):
try:
return self.receivers[signal]
except KeyError:
return []
global _global_dispatcher
_global_dispatcher = GlobalDispatcher()
_global_receivers_pool = {}
def _display_deprecation_warning():
warnings.warn("extern.louie will soon be deprecated in favor of coherence.dispatcher.")
def connect(receiver, signal=All, sender=Any, weak=True):
callback = receiver
if signal in (Any, All):
raise NotImplemented("This is not allowed. Signal HAS to be something")
if sender not in (Any, All):
_display_deprecation_warning()
receiver = _global_dispatcher.connect(signal, callback)
_global_receivers_pool[(callback, signal)] = receiver
return receiver
def disconnect(receiver, signal=All, sender=Any, weak=True):
callback = receiver
if signal in (Any, All):
raise NotImplemented("This is not allowed. Signal HAS to be something")
if sender not in (Any, All):
_display_deprecation_warning()
receiver = _global_receivers_pool.pop((callback, signal))
return _global_dispatcher.disconnect(receiver)
def send(signal=All, sender=Anonymous, *arguments, **named):
if signal in (Any, All):
raise NotImplemented("This is not allowed. Signal HAS to be something")
if sender not in (Anonymous, None):
_display_deprecation_warning()
# the first value of the callback shall always be the signal:
return _global_dispatcher.save_emit(signal, *arguments, **named)
def send_minimal(signal=All, sender=Anonymous, *arguments, **named):
return send(signal, sender, *arguments, **named)
def send_exact(signal=All, sender=Anonymous, *arguments, **named):
return send(signal, sender, *arguments, **named)
def send_robust(signal=All, sender=Anonymous, *arguments, **named):
return send(signal, sender, *arguments, **named)
|