<s> import textwrap <EOL> import logging <EOL> import six <EOL> import semantic_version <EOL> from lymph . core . components import Component , Componentized , ComponentizedBase <EOL> from lymph . core . decorators import rpc , RPCBase <EOL> from lymph . core . events import TaskHandler , EventHandler <EOL> from lymph . core . monitoring import metrics <EOL> from lymph . exceptions import RemoteError , EventHandlerTimeout , Timeout , Nack <EOL> from lymph . utils import hash_id <EOL> from lymph . core . versioning import serialize_version <EOL> import gevent <EOL> from gevent . event import AsyncResult <EOL> logger = logging . getLogger ( __name__ ) <EOL> REQUEST_TIMEOUT = <NUM_LIT:3> <EOL> class AsyncResultWrapper ( object ) : <EOL> def __init__ ( self , container , handler , async_result ) : <EOL> self . container = container <EOL> self . handler = handler <EOL> self . result = async_result <EOL> def get ( self , timeout = REQUEST_TIMEOUT ) : <EOL> try : <EOL> return self . result . get ( timeout = timeout ) <EOL> except gevent . Timeout : <EOL> self . container . unsubscribe ( self . handler ) <EOL> raise EventHandlerTimeout <EOL> class InterfaceBase ( ComponentizedBase ) : <EOL> def __new__ ( cls , clsname , bases , attrs ) : <EOL> methods = { } <EOL> event_handlers = { } <EOL> is_worker = False <EOL> for base in bases : <EOL> if isinstance ( base , InterfaceBase ) : <EOL> methods . update ( base . methods ) <EOL> is_worker |= base . worker <EOL> for name , value in six . iteritems ( attrs ) : <EOL> if isinstance ( value , RPCBase ) : <EOL> methods [ name ] = value <EOL> if issubclass ( getattr ( value , '<STR_LIT>' , object ) , EventHandler ) : <EOL> event_handlers [ name ] = value <EOL> if issubclass ( getattr ( value , '<STR_LIT>' , object ) , TaskHandler ) : <EOL> is_worker = True <EOL> new_cls = super ( InterfaceBase , cls ) . __new__ ( cls , clsname , bases , attrs ) <EOL> new_cls . methods = methods <EOL> new_cls . event_handlers = event_handlers <EOL> new_cls . worker = is_worker <EOL> return new_cls <EOL> class DeferredReply ( AsyncResult ) : <EOL> def __init__ ( self , subject ) : <EOL> super ( DeferredReply , self ) . __init__ ( ) <EOL> self . subject = subject <EOL> class ProxyMethod ( object ) : <EOL> def __init__ ( self , proxy , subject ) : <EOL> self . proxy = proxy <EOL> self . subject = subject <EOL> def __call__ ( self , ** kwargs ) : <EOL> return self . proxy . _call ( self . subject , ** kwargs ) <EOL> def defer ( self , * args , ** kwargs ) : <EOL> result = DeferredReply ( self . subject ) <EOL> self . proxy . spawn ( self , * args , ** kwargs ) . link ( result ) <EOL> return result <EOL> class Proxy ( Component ) : <EOL> def __init__ ( self , container , address , timeout = REQUEST_TIMEOUT , namespace = '<STR_LIT>' , version = None , error_map = None ) : <EOL> super ( Proxy , self ) . __init__ ( ) <EOL> self . _container = container <EOL> self . _address = address <EOL> self . _method_cache = { } <EOL> self . _timeout = timeout <EOL> self . _namespace = namespace or address <EOL> if version and not isinstance ( version , semantic_version . Version ) : <EOL> version = semantic_version . Version . coerce ( version ) <EOL> self . _version = version <EOL> self . _error_map = error_map or { } <EOL> def on_start ( self ) : <EOL> super ( Proxy , self ) . on_start ( ) <EOL> self . timeout_counts = self . metrics . add ( metrics . Counter ( '<STR_LIT>' , { '<STR_LIT:address>' : self . _address } ) ) <EOL> self . exception_counts = self . metrics . add ( metrics . TaggedCounter ( '<STR_LIT>' , { '<STR_LIT:address>' : self . _address } ) ) <EOL> def _call ( self , __name , ** kwargs ) : <EOL> channel = self . _container . send_request ( self . _address , __name , kwargs , version = self . _version ) <EOL> try : <EOL> return channel . get ( timeout = self . _timeout ) . body <EOL> except RemoteError as e : <EOL> error_type = str ( e . __class__ ) <EOL> self . exception_counts . incr ( name = e . __class__ . __name__ ) <EOL> if error_type in self . _error_map : <EOL> raise self . _error_map [ error_type ] ( ) <EOL> raise <EOL> except Timeout : <EOL> self . timeout_counts += <NUM_LIT:1> <EOL> raise <EOL> except Nack : <EOL> self . exception_counts . incr ( name = '<STR_LIT>' ) <EOL> raise <EOL> def __getattr__ ( self , name ) : <EOL> try : <EOL> return self . _method_cache [ name ] <EOL> except KeyError : <EOL> method = ProxyMethod ( self , '<STR_LIT>' % ( self . _namespace , name ) ) <EOL> self . _method_cache [ name ] = method <EOL> return method <EOL> @ six . add_metaclass ( InterfaceBase ) <EOL> class Interface ( Componentized ) : <EOL> def __init__ ( self , container , name = None , builtin = False , version = None ) : <EOL> super ( Interface , self ) . __init__ ( ) <EOL> self . container = container <EOL> self . name = name or self . __class__ . __name__ <EOL> self . base_name = self . name <EOL> self . builtin = builtin <EOL> self . version = version <EOL> if container . worker and not builtin : <EOL> self . name = '<STR_LIT>' % self . name <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' % ( self . name , self . version ) <EOL> @ property <EOL> def id ( self ) : <EOL> return hash_id ( self . container . identity , self . name , self . version ) <EOL> def should_register ( self ) : <EOL> return True <EOL> def should_install ( self ) : <EOL> return not self . container . worker or self . worker or self . builtin <EOL> def apply_config ( self , config ) : <EOL> self . config = config <EOL> def get_description ( self ) : <EOL> return { } <EOL> def handle_request ( self , func_name , channel ) : <EOL> method = self . methods [ func_name ] <EOL> channel . add_header ( '<STR_LIT:version>' , serialize_version ( self . version ) ) <EOL> method . rpc_call ( self , channel , ** channel . request . body ) <EOL> def request ( self , address , subject , body , timeout = REQUEST_TIMEOUT , version = None ) : <EOL> channel = self . container . send_request ( address , subject , body , version = version ) <EOL> return channel . get ( timeout = timeout ) <EOL> def emit ( self , event_type , payload , delay = <NUM_LIT:0> ) : <EOL> self . container . emit_event ( event_type , payload , delay = delay ) <EOL> def proxy ( self , address , ** kwargs ) : <EOL> proxy = Proxy ( self . container , address , ** kwargs ) <EOL> self . add_component ( proxy ) <EOL> return proxy <EOL> def subscribe ( self , * event_types , ** kwargs ) : <EOL> def decorator ( func ) : <EOL> from lymph . core . events import EventHandler <EOL> handler = EventHandler ( self , func , event_types , ** kwargs ) <EOL> self . container . subscribe ( handler ) <EOL> return handler <EOL> return decorator <EOL> def get_next_event ( self , * event_types , ** kwargs ) : <EOL> kwargs [ '<STR_LIT>' ] = True <EOL> result = AsyncResult ( ) <EOL> @ self . subscribe ( * event_types , ** kwargs ) <EOL> def handler ( interface , event ) : <EOL> result . set ( event . body ) <EOL> return AsyncResultWrapper ( self . container , handler , result ) <EOL> class DefaultInterface ( Interface ) : <EOL> @ rpc ( ) <EOL> def ping ( self , payload = None ) : <EOL> return payload <EOL> @ rpc ( ) <EOL> def status ( self ) : <EOL> return { <EOL> '<STR_LIT>' : self . container . endpoint , <EOL> '<STR_LIT>' : self . container . identity , <EOL> } <EOL> @ rpc ( ) <EOL> def inspect ( self ) : <EOL> """<STR_LIT>""" <EOL> methods = [ ] <EOL> for interface in list ( self . container . iter_interfaces ( ) ) : <EOL> for name , func in six . iteritems ( interface . methods ) : <EOL> methods . append ( { <EOL> '<STR_LIT:name>' : '<STR_LIT>' % ( interface . name , name ) , <EOL> '<STR_LIT:version>' : str ( interface . version ) , <EOL> '<STR_LIT>' : list ( func . args . args ) , <EOL> '<STR_LIT>' : textwrap . dedent ( func . __doc__ or '<STR_LIT>' ) . strip ( ) , <EOL> } ) <EOL> return { <EOL> '<STR_LIT>' : methods , <EOL> } <EOL> @ rpc ( ) <EOL> def change_loglevel ( self , qualname , loglevel , period = <NUM_LIT> ) : <EOL> curr_logger = logging . getLogger ( qualname ) <EOL> def reset ( ) : <EOL> logger . info ( "<STR_LIT>" , qualname , self . container . loglevel ) <EOL> curr_logger . setLevel ( self . container . loglevel ) <EOL> gevent . spawn_later ( period , reset ) <EOL> curr_logger . setLevel ( loglevel ) <EOL> logger . info ( "<STR_LIT>" , qualname , loglevel ) <EOL> @ rpc ( ) <EOL> def get_metrics ( self ) : <EOL> return list ( self . container . metrics ) </s>
