<s> """<STR_LIT>""" <EOL> import os <EOL> import stat <EOL> from twisted . internet import defer , error , reactor , threads <EOL> from twirrdy import RRDBasicAPI , OrderedDict <EOL> from twirrdy import protocol <EOL> def issock ( path ) : <EOL> mode = os . stat ( path ) [ stat . ST_MODE ] <EOL> return stat . S_ISSOCK ( mode ) <EOL> class RRDTwistedAPI ( RRDBasicAPI ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , defer = True ) : <EOL> """<STR_LIT>""" <EOL> self . _defer = defer <EOL> self . _client = None <EOL> self . update = self . _update_direct <EOL> def open ( self , address , pidfile = None ) : <EOL> """<STR_LIT>""" <EOL> deferred = defer . Deferred ( ) <EOL> self . _client = protocol . RRDCacheClient ( deferred ) <EOL> reactor . connectUNIX ( address , self . _client , checkPID = pidfile ) <EOL> self . update = self . _update_cache <EOL> return deferred <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> def filter_done ( result ) : <EOL> if isinstance ( result . value , error . ConnectionDone ) : <EOL> return None <EOL> else : <EOL> return result <EOL> assert self . _client <EOL> self . _client . stopTrying ( ) <EOL> deferred = self . _client . sendLine ( '<STR_LIT>' ) <EOL> deferred . addErrback ( filter_done ) <EOL> self . _client = None <EOL> self . update = self . _update_direct <EOL> return deferred <EOL> def flush ( self , filename ) : <EOL> assert self . _client <EOL> filename = self . _escape_filename ( filename ) <EOL> return self . _client . sendLine ( "<STR_LIT>" % filename ) <EOL> def pending ( self , filename ) : <EOL> def parse_value ( value ) : <EOL> if value == "<STR_LIT>" : <EOL> return None <EOL> else : <EOL> return float ( value ) <EOL> def parse_line ( line ) : <EOL> fields = line . split ( '<STR_LIT::>' ) <EOL> ds_time = int ( fields . pop ( <NUM_LIT:0> ) ) <EOL> ds_values = [ parse_value ( v ) for v in fields ] <EOL> return ds_time , ds_values <EOL> def parse_result ( result ) : <EOL> lines = result . splitlines ( ) <EOL> updates = [ parse_line ( l ) for l in lines [ <NUM_LIT:1> : ] ] <EOL> updates . sort ( ) <EOL> return updates <EOL> def mask_error ( failure ) : <EOL> if ( isinstance ( failure . value , protocol . RRDCacheError ) and <EOL> failure . value . args [ <NUM_LIT:0> ] == "<STR_LIT>" ) : <EOL> return [ ] <EOL> else : <EOL> return failure <EOL> assert self . _client <EOL> filename = self . _escape_filename ( filename ) <EOL> d = self . _client . sendLine ( "<STR_LIT>" % filename ) <EOL> d . addCallbacks ( parse_result , mask_error ) <EOL> return d <EOL> def _escape_filename ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> return filename . replace ( '<STR_LIT:\\>' , '<STR_LIT>' ) . replace ( '<STR_LIT:U+0020>' , '<STR_LIT>' ) <EOL> def _update_cache ( self , filename , timestamp , values , defer = None ) : <EOL> """<STR_LIT>""" <EOL> if defer is None : <EOL> defer = self . _defer <EOL> if not defer : <EOL> super ( RRDTwistedAPI , self ) . update ( filename , timestamp , values ) <EOL> else : <EOL> assert self . _client <EOL> filename = self . _escape_filename ( filename ) <EOL> line = "<STR_LIT>" % ( filename , int ( timestamp ) , <EOL> '<STR_LIT::>' . join ( str ( v ) for v in values ) ) <EOL> return self . _client . sendLine ( line ) <EOL> def _update_direct ( self , filename , timestamp , values , defer = None ) : <EOL> """<STR_LIT>""" <EOL> if defer is None : <EOL> defer = self . _defer <EOL> doupdate = lambda : super ( RRDTwistedAPI , self ) . update ( <EOL> filename , timestamp , values ) <EOL> if not defer : <EOL> doupdate ( ) <EOL> else : <EOL> return threads . deferToThread ( doupdate ) <EOL> def create ( self , filename , ds , rra , step = <NUM_LIT> , start = None , defer = None ) : <EOL> if defer is None : <EOL> defer = self . _defer <EOL> docreate = lambda : super ( RRDTwistedAPI , self ) . create ( <EOL> filename , ds , rra , step , start ) <EOL> if not defer : <EOL> docreate ( ) <EOL> else : <EOL> return threads . deferToThread ( docreate ) <EOL> def info ( self , filename , defer = None ) : <EOL> if defer is None : <EOL> defer = self . _defer <EOL> doinfo = lambda : super ( RRDTwistedAPI , self ) . info ( filename ) <EOL> if not defer : <EOL> return doinfo ( ) <EOL> else : <EOL> if self . _client : <EOL> deferred = self . flush ( filename ) <EOL> deferred . addCallback ( lambda x : threads . deferToThread ( doinfo ) ) <EOL> return deferred <EOL> else : <EOL> return threads . deferToThread ( doinfo ) <EOL> def lastupdate ( self , filename , defer = None ) : <EOL> if defer is None : <EOL> defer = self . _defer <EOL> doinfo = lambda : super ( RRDTwistedAPI , self ) . lastupdate ( filename ) <EOL> def merge ( pending ) : <EOL> def returnmerged ( result ) : <EOL> ds_time = pending [ - <NUM_LIT:1> ] [ <NUM_LIT:0> ] <EOL> ds_values = pending [ - <NUM_LIT:1> ] [ <NUM_LIT:1> ] <EOL> ds_names = result [ <NUM_LIT:1> ] . keys ( ) <EOL> return ds_time , OrderedDict ( zip ( ds_names , ds_values ) ) <EOL> deferred = threads . deferToThread ( doinfo ) <EOL> if pending : <EOL> deferred . addCallback ( returnmerged ) <EOL> return deferred <EOL> if not defer : <EOL> return doinfo ( ) <EOL> else : <EOL> if self . _client : <EOL> deferred = self . pending ( filename ) <EOL> deferred . addCallback ( merge ) <EOL> return deferred <EOL> else : <EOL> return threads . deferToThread ( doinfo ) </s>
