<s> """<STR_LIT>""" <EOL> from twisted . python import log <EOL> from twisted . words . xish import xpath <EOL> class _MethodWrapper ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , method , * args , ** kwargs ) : <EOL> self . method = method <EOL> self . args = args <EOL> self . kwargs = kwargs <EOL> def __call__ ( self , * args , ** kwargs ) : <EOL> nargs = self . args + args <EOL> nkwargs = self . kwargs . copy ( ) <EOL> nkwargs . update ( kwargs ) <EOL> self . method ( * nargs , ** nkwargs ) <EOL> class CallbackList : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . callbacks = { } <EOL> def addCallback ( self , onetime , method , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if not method in self . callbacks : <EOL> self . callbacks [ method ] = ( _MethodWrapper ( method , * args , ** kwargs ) , <EOL> onetime ) <EOL> def removeCallback ( self , method ) : <EOL> """<STR_LIT>""" <EOL> if method in self . callbacks : <EOL> del self . callbacks [ method ] <EOL> def callback ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> for key , ( methodwrapper , onetime ) in self . callbacks . items ( ) : <EOL> try : <EOL> methodwrapper ( * args , ** kwargs ) <EOL> except : <EOL> log . err ( ) <EOL> if onetime : <EOL> del self . callbacks [ key ] <EOL> def isEmpty ( self ) : <EOL> """<STR_LIT>""" <EOL> return len ( self . callbacks ) == <NUM_LIT:0> <EOL> class EventDispatcher : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , eventprefix = "<STR_LIT>" ) : <EOL> self . prefix = eventprefix <EOL> self . _eventObservers = { } <EOL> self . _xpathObservers = { } <EOL> self . _dispatchDepth = <NUM_LIT:0> <EOL> self . _updateQueue = [ ] <EOL> def _getEventAndObservers ( self , event ) : <EOL> if isinstance ( event , xpath . XPathQuery ) : <EOL> observers = self . _xpathObservers <EOL> else : <EOL> if self . prefix == event [ : len ( self . prefix ) ] : <EOL> observers = self . _eventObservers <EOL> else : <EOL> event = xpath . internQuery ( event ) <EOL> observers = self . _xpathObservers <EOL> return event , observers <EOL> def addOnetimeObserver ( self , event , observerfn , priority = <NUM_LIT:0> , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . _addObserver ( True , event , observerfn , priority , * args , ** kwargs ) <EOL> def addObserver ( self , event , observerfn , priority = <NUM_LIT:0> , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . _addObserver ( False , event , observerfn , priority , * args , ** kwargs ) <EOL> def _addObserver ( self , onetime , event , observerfn , priority , * args , ** kwargs ) : <EOL> if self . _dispatchDepth > <NUM_LIT:0> : <EOL> self . _updateQueue . append ( lambda : self . _addObserver ( onetime , event , observerfn , priority , * args , ** kwargs ) ) <EOL> return <EOL> event , observers = self . _getEventAndObservers ( event ) <EOL> if priority not in observers : <EOL> cbl = CallbackList ( ) <EOL> observers [ priority ] = { event : cbl } <EOL> else : <EOL> priorityObservers = observers [ priority ] <EOL> if event not in priorityObservers : <EOL> cbl = CallbackList ( ) <EOL> observers [ priority ] [ event ] = cbl <EOL> else : <EOL> cbl = priorityObservers [ event ] <EOL> cbl . addCallback ( onetime , observerfn , * args , ** kwargs ) <EOL> def removeObserver ( self , event , observerfn ) : <EOL> """<STR_LIT>""" <EOL> if self . _dispatchDepth > <NUM_LIT:0> : <EOL> self . _updateQueue . append ( lambda : self . removeObserver ( event , observerfn ) ) <EOL> return <EOL> event , observers = self . _getEventAndObservers ( event ) <EOL> emptyLists = [ ] <EOL> for priority , priorityObservers in observers . iteritems ( ) : <EOL> for query , callbacklist in priorityObservers . iteritems ( ) : <EOL> if event == query : <EOL> callbacklist . removeCallback ( observerfn ) <EOL> if callbacklist . isEmpty ( ) : <EOL> emptyLists . append ( ( priority , query ) ) <EOL> for priority , query in emptyLists : <EOL> del observers [ priority ] [ query ] <EOL> def dispatch ( self , obj , event = None ) : <EOL> """<STR_LIT>""" <EOL> foundTarget = False <EOL> self . _dispatchDepth += <NUM_LIT:1> <EOL> if event != None : <EOL> observers = self . _eventObservers <EOL> match = lambda query , obj : query == event <EOL> else : <EOL> observers = self . _xpathObservers <EOL> match = lambda query , obj : query . matches ( obj ) <EOL> priorities = observers . keys ( ) <EOL> priorities . sort ( ) <EOL> priorities . reverse ( ) <EOL> emptyLists = [ ] <EOL> for priority in priorities : <EOL> for query , callbacklist in observers [ priority ] . iteritems ( ) : <EOL> if match ( query , obj ) : <EOL> callbacklist . callback ( obj ) <EOL> foundTarget = True <EOL> if callbacklist . isEmpty ( ) : <EOL> emptyLists . append ( ( priority , query ) ) <EOL> for priority , query in emptyLists : <EOL> del observers [ priority ] [ query ] <EOL> self . _dispatchDepth -= <NUM_LIT:1> <EOL> if self . _dispatchDepth == <NUM_LIT:0> : <EOL> for f in self . _updateQueue : <EOL> f ( ) <EOL> self . _updateQueue = [ ] <EOL> return foundTarget <EOL> class XmlPipe ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . source = EventDispatcher ( ) <EOL> self . sink = EventDispatcher ( ) <EOL> self . source . send = lambda obj : self . sink . dispatch ( obj ) <EOL> self . sink . send = lambda obj : self . source . dispatch ( obj ) </s>
