<s> from logging import getLogger <EOL> from json import JSONDecoder , JSONEncoder <EOL> from sys import exc_info <EOL> from traceback import format_exception <EOL> from twisted . web import server , resource <EOL> from distributed_frontera . worker . utils import listen_tcp <EOL> logger = getLogger ( "<STR_LIT>" ) <EOL> def jsonrpc_error ( id , code , message , data = None ) : <EOL> """<STR_LIT>""" <EOL> return { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:error>' : { <EOL> '<STR_LIT:code>' : code , <EOL> '<STR_LIT:message>' : message , <EOL> '<STR_LIT:data>' : data , <EOL> } , <EOL> '<STR_LIT:id>' : id , <EOL> } <EOL> def jsonrpc_result ( id , result ) : <EOL> """<STR_LIT>""" <EOL> return { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:result>' : result , <EOL> '<STR_LIT:id>' : id , <EOL> } <EOL> class JsonRpcError ( Exception ) : <EOL> def __init__ ( self , code , message ) : <EOL> self . code = code <EOL> self . message = message <EOL> def __call__ ( self , id ) : <EOL> return jsonrpc_error ( id , self . code , self . message ) <EOL> class JsonResource ( resource . Resource ) : <EOL> json_encoder = JSONEncoder ( ) <EOL> json_decoder = JSONDecoder ( ) <EOL> def render ( self , txrequest ) : <EOL> r = resource . Resource . render ( self , txrequest ) <EOL> return self . render_object ( r , txrequest ) <EOL> def render_object ( self , obj , txrequest ) : <EOL> r = self . json_encoder . encode ( obj ) + "<STR_LIT:\n>" <EOL> txrequest . setHeader ( '<STR_LIT:Content-Type>' , '<STR_LIT:application/json>' ) <EOL> txrequest . setHeader ( '<STR_LIT>' , '<STR_LIT:*>' ) <EOL> txrequest . setHeader ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> txrequest . setHeader ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> txrequest . setHeader ( '<STR_LIT>' , len ( r ) ) <EOL> return r <EOL> def parse_jsonrpc ( self , txrequest ) : <EOL> if isinstance ( txrequest . content , file ) : <EOL> data = txrequest . content . read ( ) <EOL> else : <EOL> data = txrequest . content . getvalue ( ) <EOL> return self . json_decoder . decode ( data ) <EOL> class StatusResource ( JsonResource ) : <EOL> ws_name = '<STR_LIT:status>' <EOL> def __init__ ( self , worker ) : <EOL> self . worker = worker <EOL> JsonResource . __init__ ( self ) <EOL> def render_GET ( self , txrequest ) : <EOL> return { <EOL> '<STR_LIT>' : self . worker . slot . is_finishing , <EOL> '<STR_LIT>' : self . worker . slot . disable_new_batches , <EOL> '<STR_LIT>' : self . worker . stats <EOL> } <EOL> class JsonRpcResource ( JsonResource ) : <EOL> ws_name = '<STR_LIT>' <EOL> def __init__ ( self ) : <EOL> JsonResource . __init__ ( self ) <EOL> def render_POST ( self , txrequest ) : <EOL> jrequest = self . parse_jsonrpc ( txrequest ) <EOL> method = jrequest [ '<STR_LIT>' ] <EOL> try : <EOL> try : <EOL> return self . process_request ( method , jrequest ) <EOL> except Exception , err : <EOL> if isinstance ( err , JsonRpcError ) : <EOL> raise err <EOL> trace_lines = format_exception ( * exc_info ( ) ) <EOL> raise JsonRpcError ( <NUM_LIT> , "<STR_LIT>" % ( str ( "<STR_LIT>" ) . join ( trace_lines ) ) ) <EOL> except JsonRpcError , err : <EOL> return err ( jrequest [ '<STR_LIT:id>' ] ) <EOL> class WorkerJsonRpcResource ( JsonRpcResource ) : <EOL> def __init__ ( self , worker ) : <EOL> self . worker = worker <EOL> JsonRpcResource . __init__ ( self ) <EOL> def process_request ( self , method , jrequest ) : <EOL> if method == '<STR_LIT>' : <EOL> self . worker . disable_new_batches ( ) <EOL> return jsonrpc_result ( jrequest [ '<STR_LIT:id>' ] , "<STR_LIT:success>" ) <EOL> if method == '<STR_LIT>' : <EOL> self . worker . enable_new_batches ( ) <EOL> return jsonrpc_result ( jrequest [ '<STR_LIT:id>' ] , "<STR_LIT:success>" ) <EOL> raise JsonRpcError ( <NUM_LIT> , "<STR_LIT>" ) <EOL> class RootResource ( JsonResource ) : <EOL> def render_GET ( self , txrequest ) : <EOL> return { '<STR_LIT>' : self . children . keys ( ) } <EOL> def getChild ( self , name , txrequest ) : <EOL> if name == '<STR_LIT>' : <EOL> return self <EOL> return JsonResource . getChild ( self , name , txrequest ) <EOL> class JsonRpcService ( server . Site ) : <EOL> def __init__ ( self , root , settings ) : <EOL> logfile = settings . get ( '<STR_LIT>' ) <EOL> self . portrange = settings . get ( '<STR_LIT>' , [ <NUM_LIT> , <NUM_LIT> ] ) <EOL> self . host = settings . get ( '<STR_LIT>' , '<STR_LIT:127.0.0.1>' ) <EOL> server . Site . __init__ ( self , root , logPath = logfile ) <EOL> self . noisy = False <EOL> def start_listening ( self ) : <EOL> self . port = listen_tcp ( self . portrange , self . host , self ) <EOL> h = self . port . getHost ( ) <EOL> logger . info ( '<STR_LIT>' . format ( host = h . host , port = h . port ) ) <EOL> def stop_listening ( self ) : <EOL> self . port . stopListening ( ) <EOL> class WorkerJsonRpcService ( JsonRpcService ) : <EOL> def __init__ ( self , worker , settings ) : <EOL> root = RootResource ( ) <EOL> root . putChild ( '<STR_LIT:status>' , StatusResource ( worker ) ) <EOL> root . putChild ( '<STR_LIT>' , WorkerJsonRpcResource ( worker ) ) <EOL> JsonRpcService . __init__ ( self , root , settings ) <EOL> self . worker = worker <EOL> def start_listening ( self ) : <EOL> JsonRpcService . start_listening ( self ) <EOL> address = self . port . getHost ( ) <EOL> self . worker . set_process_info ( "<STR_LIT>" % ( address . host , address . port ) ) </s>
