<s> """<STR_LIT>""" <EOL> import sys <EOL> import weakref <EOL> import itertools <EOL> import socket <EOL> import time <EOL> import gc <EOL> from threading import Lock , RLock , Event <EOL> from rpyc . lib . compat import pickle , next , is_py3k , maxint , select_error <EOL> from rpyc . lib . colls import WeakValueDict , RefCountingColl <EOL> from rpyc . core import consts , brine , vinegar , netref <EOL> from rpyc . core . async import AsyncResult <EOL> class PingError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> DEFAULT_CONFIG = dict ( <EOL> allow_safe_attrs = True , <EOL> allow_exposed_attrs = True , <EOL> allow_public_attrs = False , <EOL> allow_all_attrs = False , <EOL> safe_attrs = set ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , ] ) , <EOL> exposed_prefix = "<STR_LIT>" , <EOL> allow_getattr = True , <EOL> allow_setattr = False , <EOL> allow_delattr = False , <EOL> include_local_traceback = True , <EOL> instantiate_custom_exceptions = False , <EOL> import_custom_exceptions = False , <EOL> instantiate_oldstyle_exceptions = False , <EOL> propagate_SystemExit_locally = False , <EOL> propagate_KeyboardInterrupt_locally = True , <EOL> log_exceptions = True , <EOL> allow_pickle = False , <EOL> connid = None , <EOL> credentials = None , <EOL> endpoints = None , <EOL> logger = None , <EOL> ) <EOL> """<STR_LIT>""" <EOL> _connection_id_generator = itertools . count ( <NUM_LIT:1> ) <EOL> class Connection ( object ) : <EOL> """<STR_LIT>""" <EOL> SYNC_REQUEST_TIMEOUT = <NUM_LIT:30> <EOL> def __init__ ( self , service , channel , config = { } , _lazy = False ) : <EOL> self . _closed = True <EOL> self . _config = DEFAULT_CONFIG . copy ( ) <EOL> self . _config . update ( config ) <EOL> if self . _config [ "<STR_LIT>" ] is None : <EOL> self . _config [ "<STR_LIT>" ] = "<STR_LIT>" % ( next ( _connection_id_generator ) , ) <EOL> self . _channel = channel <EOL> self . _seqcounter = itertools . count ( ) <EOL> self . _recvlock = Lock ( ) <EOL> self . _sendlock = Lock ( ) <EOL> self . _sync_replies = { } <EOL> self . _sync_lock = RLock ( ) <EOL> self . _sync_event = Event ( ) <EOL> self . _async_callbacks = { } <EOL> self . _local_objects = RefCountingColl ( ) <EOL> self . _last_traceback = None <EOL> self . _proxy_cache = WeakValueDict ( ) <EOL> self . _netref_classes_cache = { } <EOL> self . _remote_root = None <EOL> self . _local_root = service ( weakref . proxy ( self ) ) <EOL> if not _lazy : <EOL> self . _init_service ( ) <EOL> self . _closed = False <EOL> def _init_service ( self ) : <EOL> self . _local_root . on_connect ( ) <EOL> def __del__ ( self ) : <EOL> self . close ( ) <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , t , v , tb ) : <EOL> self . close ( ) <EOL> def __repr__ ( self ) : <EOL> a , b = object . __repr__ ( self ) . split ( "<STR_LIT>" ) <EOL> return "<STR_LIT>" % ( a , self . _config [ "<STR_LIT>" ] , b ) <EOL> def _cleanup ( self , _anyway = True ) : <EOL> if self . _closed and not _anyway : <EOL> return <EOL> self . _closed = True <EOL> self . _channel . close ( ) <EOL> self . _local_root . on_disconnect ( ) <EOL> self . _sync_replies . clear ( ) <EOL> self . _async_callbacks . clear ( ) <EOL> self . _local_objects . clear ( ) <EOL> self . _proxy_cache . clear ( ) <EOL> self . _netref_classes_cache . clear ( ) <EOL> self . _last_traceback = None <EOL> self . _remote_root = None <EOL> self . _local_root = None <EOL> def close ( self , _catchall = True ) : <EOL> """<STR_LIT>""" <EOL> if self . _closed : <EOL> return <EOL> self . _closed = True <EOL> try : <EOL> self . _async_request ( consts . HANDLE_CLOSE ) <EOL> except EOFError : <EOL> pass <EOL> except Exception : <EOL> if not _catchall : <EOL> raise <EOL> finally : <EOL> self . _cleanup ( _anyway = True ) <EOL> @ property <EOL> def closed ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _closed <EOL> def fileno ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _channel . fileno ( ) <EOL> def ping ( self , data = None , timeout = <NUM_LIT:3> ) : <EOL> """<STR_LIT>""" <EOL> if data is None : <EOL> data = "<STR_LIT>" * <NUM_LIT:20> <EOL> res = self . async_request ( consts . HANDLE_PING , data , timeout = timeout ) <EOL> if res . value != data : <EOL> raise PingError ( "<STR_LIT>" ) <EOL> def _get_seq_id ( self ) : <EOL> return next ( self . _seqcounter ) <EOL> def _send ( self , msg , seq , args ) : <EOL> data = brine . dump ( ( msg , seq , args ) ) <EOL> is_gc_enabled = gc . isenabled ( ) <EOL> gc . disable ( ) <EOL> self . _sendlock . acquire ( ) <EOL> try : <EOL> self . _channel . send ( data ) <EOL> finally : <EOL> self . _sendlock . release ( ) <EOL> if is_gc_enabled : <EOL> gc . enable ( ) <EOL> def _send_request ( self , seq , handler , args ) : <EOL> self . _send ( consts . MSG_REQUEST , seq , ( handler , self . _box ( args ) ) ) <EOL> def _send_reply ( self , seq , obj ) : <EOL> self . _send ( consts . MSG_REPLY , seq , self . _box ( obj ) ) <EOL> def _send_exception ( self , seq , exctype , excval , exctb ) : <EOL> exc = vinegar . dump ( exctype , excval , exctb , <EOL> include_local_traceback = self . _config [ "<STR_LIT>" ] ) <EOL> self . _send ( consts . MSG_EXCEPTION , seq , exc ) <EOL> def _box ( self , obj ) : <EOL> """<STR_LIT>""" <EOL> if brine . dumpable ( obj ) : <EOL> return consts . LABEL_VALUE , obj <EOL> if type ( obj ) is tuple : <EOL> return consts . LABEL_TUPLE , tuple ( self . _box ( item ) for item in obj ) <EOL> elif isinstance ( obj , netref . BaseNetref ) and obj . ____conn__ ( ) is self : <EOL> return consts . LABEL_LOCAL_REF , obj . ____oid__ <EOL> else : <EOL> self . _local_objects . add ( obj ) <EOL> try : <EOL> cls = obj . __class__ <EOL> except Exception : <EOL> cls = type ( obj ) <EOL> if not isinstance ( cls , type ) : <EOL> cls = type ( obj ) <EOL> return consts . LABEL_REMOTE_REF , ( id ( obj ) , cls . __name__ , cls . __module__ ) <EOL> def _unbox ( self , package ) : <EOL> """<STR_LIT>""" <EOL> label , value = package <EOL> if label == consts . LABEL_VALUE : <EOL> return value <EOL> if label == consts . LABEL_TUPLE : <EOL> return tuple ( self . _unbox ( item ) for item in value ) <EOL> if label == consts . LABEL_LOCAL_REF : <EOL> return self . _local_objects [ value ] <EOL> if label == consts . LABEL_REMOTE_REF : <EOL> oid , clsname , modname = value <EOL> if oid in self . _proxy_cache : <EOL> return self . _proxy_cache [ oid ] <EOL> proxy = self . _netref_factory ( oid , clsname , modname ) <EOL> self . _proxy_cache [ oid ] = proxy <EOL> return proxy <EOL> raise ValueError ( "<STR_LIT>" % ( label , ) ) <EOL> def _netref_factory ( self , oid , clsname , modname ) : <EOL> typeinfo = ( clsname , modname ) <EOL> if typeinfo in self . _netref_classes_cache : <EOL> cls = self . _netref_classes_cache [ typeinfo ] <EOL> elif typeinfo in netref . builtin_classes_cache : <EOL> cls = netref . builtin_classes_cache [ typeinfo ] <EOL> else : <EOL> info = self . sync_request ( consts . HANDLE_INSPECT , oid ) <EOL> cls = netref . class_factory ( clsname , modname , info ) <EOL> self . _netref_classes_cache [ typeinfo ] = cls <EOL> return cls ( weakref . ref ( self ) , oid ) <EOL> def _dispatch_request ( self , seq , raw_args ) : <EOL> logger = self . _config [ "<STR_LIT>" ] <EOL> try : <EOL> handler , args = raw_args <EOL> if logger : <EOL> logger . debug ( "<STR_LIT>" , handler , seq ) <EOL> args = self . _unbox ( args ) <EOL> res = self . _HANDLERS [ handler ] ( self , * args ) <EOL> except : <EOL> t , v , tb = sys . exc_info ( ) <EOL> self . _last_traceback = tb <EOL> if logger and t is not StopIteration : <EOL> logger . debug ( "<STR_LIT>" , exc_info = True ) <EOL> if t is SystemExit and self . _config [ "<STR_LIT>" ] : <EOL> raise <EOL> if t is KeyboardInterrupt and self . _config [ "<STR_LIT>" ] : <EOL> raise <EOL> self . _send_exception ( seq , t , v , tb ) <EOL> else : <EOL> self . _send_reply ( seq , res ) <EOL> def _dispatch_reply ( self , seq , raw ) : <EOL> obj = self . _unbox ( raw ) <EOL> if seq in self . _async_callbacks : <EOL> self . _async_callbacks . pop ( seq ) ( False , obj ) <EOL> else : <EOL> self . _sync_replies [ seq ] = ( False , obj ) <EOL> def _dispatch_exception ( self , seq , raw ) : <EOL> obj = vinegar . load ( raw , <EOL> import_custom_exceptions = self . _config [ "<STR_LIT>" ] , <EOL> instantiate_custom_exceptions = self . _config [ "<STR_LIT>" ] , <EOL> instantiate_oldstyle_exceptions = self . _config [ "<STR_LIT>" ] ) <EOL> if seq in self . _async_callbacks : <EOL> self . _async_callbacks . pop ( seq ) ( True , obj ) <EOL> else : <EOL> self . _sync_replies [ seq ] = ( True , obj ) <EOL> def _recv ( self , timeout , wait_for_lock ) : <EOL> if not self . _recvlock . acquire ( wait_for_lock ) : <EOL> return None <EOL> try : <EOL> if self . _channel . poll ( timeout ) : <EOL> data = self . _channel . recv ( ) <EOL> else : <EOL> data = None <EOL> except EOFError : <EOL> self . close ( ) <EOL> raise <EOL> finally : <EOL> self . _recvlock . release ( ) <EOL> return data <EOL> def _dispatch ( self , data ) : <EOL> msg , seq , args = brine . load ( data ) <EOL> if msg == consts . MSG_REQUEST : <EOL> self . _dispatch_request ( seq , args ) <EOL> elif msg == consts . MSG_REPLY : <EOL> self . _dispatch_reply ( seq , args ) <EOL> elif msg == consts . MSG_EXCEPTION : <EOL> self . _dispatch_exception ( seq , args ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" % ( msg , ) ) <EOL> def poll ( self , timeout = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> data = self . _recv ( timeout , wait_for_lock = False ) <EOL> if not data : <EOL> return False <EOL> self . _dispatch ( data ) <EOL> return True <EOL> def serve ( self , timeout = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> data = self . _recv ( timeout , wait_for_lock = True ) <EOL> if not data : <EOL> return False <EOL> self . _dispatch ( data ) <EOL> return True <EOL> def serve_all ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> while True : <EOL> self . serve ( <NUM_LIT:0.1> ) <EOL> except ( socket . error , select_error , IOError ) : <EOL> if not self . closed : <EOL> raise <EOL> except EOFError : <EOL> pass <EOL> finally : <EOL> self . close ( ) <EOL> def poll_all ( self , timeout = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> at_least_once = False <EOL> t0 = time . time ( ) <EOL> duration = timeout <EOL> try : <EOL> while True : <EOL> if self . poll ( duration ) : <EOL> at_least_once = True <EOL> if timeout is not None : <EOL> duration = t0 + timeout - time . time ( ) <EOL> if duration < <NUM_LIT:0> : <EOL> break <EOL> except EOFError : <EOL> pass <EOL> return at_least_once <EOL> def sync_request ( self , handler , * args ) : <EOL> """<STR_LIT>""" <EOL> seq = self . _get_seq_id ( ) <EOL> self . _send_request ( seq , handler , args ) <EOL> start_time = time . time ( ) <EOL> while seq not in self . _sync_replies : <EOL> remaining_time = self . SYNC_REQUEST_TIMEOUT - ( time . time ( ) - start_time ) <EOL> if remaining_time < <NUM_LIT:0> : <EOL> raise socket . timeout <EOL> if self . _sync_lock . acquire ( False ) : <EOL> self . _sync_event . clear ( ) <EOL> try : <EOL> self . serve ( remaining_time ) <EOL> finally : <EOL> self . _sync_lock . release ( ) <EOL> self . _sync_event . set ( ) <EOL> else : <EOL> self . _sync_event . wait ( remaining_time ) <EOL> isexc , obj = self . _sync_replies . pop ( seq ) <EOL> if isexc : <EOL> raise obj <EOL> else : <EOL> return obj <EOL> def _async_request ( self , handler , args = ( ) , callback = ( lambda a , b : None ) ) : <EOL> seq = self . _get_seq_id ( ) <EOL> self . _async_callbacks [ seq ] = callback <EOL> try : <EOL> self . _send_request ( seq , handler , args ) <EOL> except : <EOL> if seq in self . _async_callbacks : <EOL> del self . _async_callbacks [ seq ] <EOL> raise <EOL> def async_request ( self , handler , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> timeout = kwargs . pop ( "<STR_LIT>" , None ) <EOL> if kwargs : <EOL> raise TypeError ( "<STR_LIT>" % ( list ( kwargs . keys ( ) ) , ) ) <EOL> res = AsyncResult ( weakref . proxy ( self ) ) <EOL> self . _async_request ( handler , args , res ) <EOL> if timeout is not None : <EOL> res . set_expiry ( timeout ) <EOL> return res <EOL> @ property <EOL> def root ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _remote_root is None : <EOL> self . _remote_root = self . sync_request ( consts . HANDLE_GETROOT ) <EOL> return self . _remote_root <EOL> def _check_attr ( self , obj , name ) : <EOL> if self . _config [ "<STR_LIT>" ] : <EOL> if name . startswith ( self . _config [ "<STR_LIT>" ] ) : <EOL> name2 = name <EOL> else : <EOL> name2 = self . _config [ "<STR_LIT>" ] + name <EOL> if hasattr ( obj , name2 ) : <EOL> return name2 <EOL> if self . _config [ "<STR_LIT>" ] : <EOL> return name <EOL> if self . _config [ "<STR_LIT>" ] and name in self . _config [ "<STR_LIT>" ] : <EOL> return name <EOL> if self . _config [ "<STR_LIT>" ] and not name . startswith ( "<STR_LIT:_>" ) : <EOL> return name <EOL> return False <EOL> def _access_attr ( self , oid , name , args , overrider , param , default ) : <EOL> if is_py3k : <EOL> if type ( name ) is bytes : <EOL> name = str ( name , "<STR_LIT:utf8>" ) <EOL> elif type ( name ) is not str : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> else : <EOL> if type ( name ) not in ( str , unicode ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> name = str ( name ) <EOL> obj = self . _local_objects [ oid ] <EOL> accessor = getattr ( type ( obj ) , overrider , None ) <EOL> if accessor is None : <EOL> name2 = self . _check_attr ( obj , name ) <EOL> if not self . _config [ param ] or not name2 : <EOL> raise AttributeError ( "<STR_LIT>" % ( name , ) ) <EOL> accessor = default <EOL> name = name2 <EOL> return accessor ( obj , name , * args ) <EOL> def _handle_ping ( self , data ) : <EOL> return data <EOL> def _handle_close ( self ) : <EOL> self . _cleanup ( ) <EOL> def _handle_getroot ( self ) : <EOL> return self . _local_root <EOL> def _handle_del ( self , oid ) : <EOL> self . _local_objects . decref ( oid ) <EOL> def _handle_repr ( self , oid ) : <EOL> return repr ( self . _local_objects [ oid ] ) <EOL> def _handle_str ( self , oid ) : <EOL> return str ( self . _local_objects [ oid ] ) <EOL> def _handle_cmp ( self , oid , other ) : <EOL> obj = self . _local_objects [ oid ] <EOL> try : <EOL> return type ( obj ) . __cmp__ ( obj , other ) <EOL> except ( AttributeError , TypeError ) : <EOL> return NotImplemented <EOL> def _handle_hash ( self , oid ) : <EOL> return hash ( self . _local_objects [ oid ] ) <EOL> def _handle_call ( self , oid , args , kwargs = ( ) ) : <EOL> return self . _local_objects [ oid ] ( * args , ** dict ( kwargs ) ) <EOL> def _handle_dir ( self , oid ) : <EOL> return tuple ( dir ( self . _local_objects [ oid ] ) ) <EOL> def _handle_inspect ( self , oid ) : <EOL> return tuple ( netref . inspect_methods ( self . _local_objects [ oid ] ) ) <EOL> def _handle_getattr ( self , oid , name ) : <EOL> return self . _access_attr ( oid , name , ( ) , "<STR_LIT>" , "<STR_LIT>" , getattr ) <EOL> def _handle_delattr ( self , oid , name ) : <EOL> return self . _access_attr ( oid , name , ( ) , "<STR_LIT>" , "<STR_LIT>" , delattr ) <EOL> def _handle_setattr ( self , oid , name , value ) : <EOL> return self . _access_attr ( oid , name , ( value , ) , "<STR_LIT>" , "<STR_LIT>" , setattr ) <EOL> def _handle_callattr ( self , oid , name , args , kwargs ) : <EOL> return self . _handle_getattr ( oid , name ) ( * args , ** dict ( kwargs ) ) <EOL> def _handle_pickle ( self , oid , proto ) : <EOL> if not self . _config [ "<STR_LIT>" ] : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return pickle . dumps ( self . _local_objects [ oid ] , proto ) <EOL> def _handle_buffiter ( self , oid , count ) : <EOL> items = [ ] <EOL> obj = self . _local_objects [ oid ] <EOL> i = <NUM_LIT:0> <EOL> try : <EOL> while i < count : <EOL> items . append ( next ( obj ) ) <EOL> i += <NUM_LIT:1> <EOL> except StopIteration : <EOL> pass <EOL> return tuple ( items ) <EOL> def _handle_oldslicing ( self , oid , attempt , fallback , start , stop , args ) : <EOL> try : <EOL> getitem = self . _handle_getattr ( oid , attempt ) <EOL> return getitem ( slice ( start , stop ) , * args ) <EOL> except Exception : <EOL> if stop is None : <EOL> stop = maxint <EOL> getslice = self . _handle_getattr ( oid , fallback ) <EOL> return getslice ( start , stop , * args ) <EOL> _HANDLERS = { } <EOL> for name , obj in dict ( locals ( ) ) . items ( ) : <EOL> if name . startswith ( "<STR_LIT>" ) : <EOL> name2 = "<STR_LIT>" + name [ <NUM_LIT:8> : ] . upper ( ) <EOL> if hasattr ( consts , name2 ) : <EOL> _HANDLERS [ getattr ( consts , name2 ) ] = obj <EOL> else : <EOL> raise NameError ( "<STR_LIT>" , name ) <EOL> del name , name2 , obj </s>
