home *** CD-ROM | disk | FTP | other *** search
- # Source Generated with Decompyle++
- # File: in.pyo (Python 2.6)
-
- from primitives.error_handling import with_traceback
- from primitives.funcs import Delegate
- from primitives.synchronization import lock
- from peak.util.addons import AddOn
- from peak.util.plugins import Hook
- import threading
-
- class OneShotHook(AddOn):
-
- def __init__(self, subject, group, impl = None):
- self.subject = subject
- self.group = group
- self.impl = impl
- self.delegate = Delegate()
- self._lock = threading.RLock()
- self.registered = False
- self.fired = False
-
-
- def check_registered(self):
- if not self.registered:
- self.registered = True
- Hook(self.group, self.impl).register(self.call_and_clear)
-
-
- check_registered = lock(check_registered)
-
- def __call__(self, func, if_not_fired = False):
- if not if_not_fired and self.fired:
- (self.delegate,) += (lambda : with_traceback(func, *a, **k))
- self.check_registered()
- return True
- return False
-
- __call__ = lock(__call__)
-
- def call_and_clear(self, *a, **k):
- self.delegate.call_and_clear(*a, **k)
- self.fired = True
-
- call_and_clear = lock(call_and_clear)
-
-