<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> import weakref <EOL> import collections <EOL> import types <EOL> from . . import exc , util <EOL> _key_to_collection = collections . defaultdict ( dict ) <EOL> """<STR_LIT>""" <EOL> _collection_to_key = collections . defaultdict ( dict ) <EOL> """<STR_LIT>""" <EOL> def _collection_gced ( ref ) : <EOL> if not _collection_to_key or ref not in _collection_to_key : <EOL> return <EOL> listener_to_key = _collection_to_key . pop ( ref ) <EOL> for key in listener_to_key . values ( ) : <EOL> if key in _key_to_collection : <EOL> dispatch_reg = _key_to_collection [ key ] <EOL> dispatch_reg . pop ( ref ) <EOL> if not dispatch_reg : <EOL> _key_to_collection . pop ( key ) <EOL> def _stored_in_collection ( event_key , owner ) : <EOL> key = event_key . _key <EOL> dispatch_reg = _key_to_collection [ key ] <EOL> owner_ref = owner . ref <EOL> listen_ref = weakref . ref ( event_key . _listen_fn ) <EOL> if owner_ref in dispatch_reg : <EOL> return False <EOL> dispatch_reg [ owner_ref ] = listen_ref <EOL> listener_to_key = _collection_to_key [ owner_ref ] <EOL> listener_to_key [ listen_ref ] = key <EOL> return True <EOL> def _removed_from_collection ( event_key , owner ) : <EOL> key = event_key . _key <EOL> dispatch_reg = _key_to_collection [ key ] <EOL> listen_ref = weakref . ref ( event_key . _listen_fn ) <EOL> owner_ref = owner . ref <EOL> dispatch_reg . pop ( owner_ref , None ) <EOL> if not dispatch_reg : <EOL> del _key_to_collection [ key ] <EOL> if owner_ref in _collection_to_key : <EOL> listener_to_key = _collection_to_key [ owner_ref ] <EOL> listener_to_key . pop ( listen_ref ) <EOL> def _stored_in_collection_multi ( newowner , oldowner , elements ) : <EOL> if not elements : <EOL> return <EOL> oldowner = oldowner . ref <EOL> newowner = newowner . ref <EOL> old_listener_to_key = _collection_to_key [ oldowner ] <EOL> new_listener_to_key = _collection_to_key [ newowner ] <EOL> for listen_fn in elements : <EOL> listen_ref = weakref . ref ( listen_fn ) <EOL> key = old_listener_to_key [ listen_ref ] <EOL> dispatch_reg = _key_to_collection [ key ] <EOL> if newowner in dispatch_reg : <EOL> assert dispatch_reg [ newowner ] == listen_ref <EOL> else : <EOL> dispatch_reg [ newowner ] = listen_ref <EOL> new_listener_to_key [ listen_ref ] = key <EOL> def _clear ( owner , elements ) : <EOL> if not elements : <EOL> return <EOL> owner = owner . ref <EOL> listener_to_key = _collection_to_key [ owner ] <EOL> for listen_fn in elements : <EOL> listen_ref = weakref . ref ( listen_fn ) <EOL> key = listener_to_key [ listen_ref ] <EOL> dispatch_reg = _key_to_collection [ key ] <EOL> dispatch_reg . pop ( owner , None ) <EOL> if not dispatch_reg : <EOL> del _key_to_collection [ key ] <EOL> class _EventKey ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , target , identifier , <EOL> fn , dispatch_target , _fn_wrap = None ) : <EOL> self . target = target <EOL> self . identifier = identifier <EOL> self . fn = fn <EOL> if isinstance ( fn , types . MethodType ) : <EOL> self . fn_key = id ( fn . __func__ ) , id ( fn . __self__ ) <EOL> else : <EOL> self . fn_key = id ( fn ) <EOL> self . fn_wrap = _fn_wrap <EOL> self . dispatch_target = dispatch_target <EOL> @ property <EOL> def _key ( self ) : <EOL> return ( id ( self . target ) , self . identifier , self . fn_key ) <EOL> def with_wrapper ( self , fn_wrap ) : <EOL> if fn_wrap is self . _listen_fn : <EOL> return self <EOL> else : <EOL> return _EventKey ( <EOL> self . target , <EOL> self . identifier , <EOL> self . fn , <EOL> self . dispatch_target , <EOL> _fn_wrap = fn_wrap <EOL> ) <EOL> def with_dispatch_target ( self , dispatch_target ) : <EOL> if dispatch_target is self . dispatch_target : <EOL> return self <EOL> else : <EOL> return _EventKey ( <EOL> self . target , <EOL> self . identifier , <EOL> self . fn , <EOL> dispatch_target , <EOL> _fn_wrap = self . fn_wrap <EOL> ) <EOL> def listen ( self , * args , ** kw ) : <EOL> once = kw . pop ( "<STR_LIT>" , False ) <EOL> named = kw . pop ( "<STR_LIT>" , False ) <EOL> target , identifier , fn = self . dispatch_target , self . identifier , self . _listen_fn <EOL> dispatch_descriptor = getattr ( target . dispatch , identifier ) <EOL> adjusted_fn = dispatch_descriptor . _adjust_fn_spec ( fn , named ) <EOL> self = self . with_wrapper ( adjusted_fn ) <EOL> if once : <EOL> self . with_wrapper ( <EOL> util . only_once ( self . _listen_fn ) ) . listen ( * args , ** kw ) <EOL> else : <EOL> self . dispatch_target . dispatch . _listen ( self , * args , ** kw ) <EOL> def remove ( self ) : <EOL> key = self . _key <EOL> if key not in _key_to_collection : <EOL> raise exc . InvalidRequestError ( <EOL> "<STR_LIT>" % <EOL> ( self . target , self . identifier , self . fn ) <EOL> ) <EOL> dispatch_reg = _key_to_collection . pop ( key ) <EOL> for collection_ref , listener_ref in dispatch_reg . items ( ) : <EOL> collection = collection_ref ( ) <EOL> listener_fn = listener_ref ( ) <EOL> if collection is not None and listener_fn is not None : <EOL> collection . remove ( self . with_wrapper ( listener_fn ) ) <EOL> def contains ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _key in _key_to_collection <EOL> def base_listen ( self , propagate = False , insert = False , <EOL> named = False ) : <EOL> target , identifier , fn = self . dispatch_target , self . identifier , self . _listen_fn <EOL> dispatch_descriptor = getattr ( target . dispatch , identifier ) <EOL> if insert : <EOL> dispatch_descriptor . for_modify ( target . dispatch ) . insert ( self , propagate ) <EOL> else : <EOL> dispatch_descriptor . for_modify ( target . dispatch ) . append ( self , propagate ) <EOL> @ property <EOL> def _listen_fn ( self ) : <EOL> return self . fn_wrap or self . fn <EOL> def append_to_list ( self , owner , list_ ) : <EOL> if _stored_in_collection ( self , owner ) : <EOL> list_ . append ( self . _listen_fn ) <EOL> return True <EOL> else : <EOL> return False <EOL> def remove_from_list ( self , owner , list_ ) : <EOL> _removed_from_collection ( self , owner ) <EOL> list_ . remove ( self . _listen_fn ) <EOL> def prepend_to_list ( self , owner , list_ ) : <EOL> if _stored_in_collection ( self , owner ) : <EOL> list_ . insert ( <NUM_LIT:0> , self . _listen_fn ) <EOL> return True <EOL> else : <EOL> return False </s>
