ryu/build/lib.linux-armv7l-2.7/ryu/services/protocols/bgp/signals/base.py

   1 import logging
   2 LOG = logging.getLogger('bgpspeaker.signals.base')
   3
   4
   5 class SignalBus(object):
   6     def __init__(self):
   7         self._listeners = {}
   8
   9     def emit_signal(self, identifier, data):
  10         identifier = _to_tuple(identifier)
  11         LOG.debug('SIGNAL: %s emitted with data: %s ', identifier, data)
  12         for func, filter_func in self._listeners.get(identifier, []):
  13             if not filter_func or filter_func(data):
  14                 func(identifier, data)
  15
  16     def register_listener(self, identifier, func, filter_func=None):
  17         identifier = _to_tuple(identifier)
  18         substrings = (identifier[:i] for i in range(1, len(identifier) + 1))
  19         for partial_id in substrings:
  20             self._listeners.setdefault(
  21                 partial_id,
  22                 []
  23             ).append((func, filter_func))
  24
  25     def unregister_all(self):
  26         self._listeners = {}
  27
  28
  29 def _to_tuple(tuple_or_not):
  30     if not isinstance(tuple_or_not, tuple):
  31         return (tuple_or_not, )
  32     else:
  33         return tuple_or_not