<s> from __future__ import absolute_import , unicode_literals <EOL> import sys <EOL> import uuid <EOL> from functools import partial <EOL> from logging import getLogger <EOL> from eventlet . event import Event <EOL> from eventlet . queue import Empty <EOL> from kombu import Connection , Exchange , Queue <EOL> from kombu . pools import producers <EOL> import kombu . serialization <EOL> from nameko . constants import ( <EOL> AMQP_URI_CONFIG_KEY , DEFAULT_RETRY_POLICY , RPC_EXCHANGE_CONFIG_KEY , <EOL> SERIALIZER_CONFIG_KEY , DEFAULT_SERIALIZER ) <EOL> from nameko . exceptions import ( <EOL> ContainerBeingKilled , deserialize , MalformedRequest , MethodNotFound , <EOL> RpcConnectionError , serialize , UnknownService , UnserializableValueError ) <EOL> from nameko . extensions import ( <EOL> DependencyProvider , Entrypoint , ProviderCollector , SharedExtension ) <EOL> from nameko . messaging import HeaderDecoder , HeaderEncoder , QueueConsumer <EOL> _log = getLogger ( __name__ ) <EOL> RPC_QUEUE_TEMPLATE = '<STR_LIT>' <EOL> RPC_REPLY_QUEUE_TEMPLATE = '<STR_LIT>' <EOL> def get_rpc_exchange ( config ) : <EOL> exchange_name = config . get ( RPC_EXCHANGE_CONFIG_KEY , '<STR_LIT>' ) <EOL> exchange = Exchange ( exchange_name , durable = True , type = "<STR_LIT>" ) <EOL> return exchange <EOL> class RpcConsumer ( SharedExtension , ProviderCollector ) : <EOL> queue_consumer = QueueConsumer ( ) <EOL> def __init__ ( self ) : <EOL> self . _unregistering_providers = set ( ) <EOL> self . _unregistered_from_queue_consumer = Event ( ) <EOL> self . queue = None <EOL> super ( RpcConsumer , self ) . __init__ ( ) <EOL> def setup ( self ) : <EOL> if self . queue is None : <EOL> service_name = self . container . service_name <EOL> queue_name = RPC_QUEUE_TEMPLATE . format ( service_name ) <EOL> routing_key = '<STR_LIT>' . format ( service_name ) <EOL> exchange = get_rpc_exchange ( self . container . config ) <EOL> self . queue = Queue ( <EOL> queue_name , <EOL> exchange = exchange , <EOL> routing_key = routing_key , <EOL> durable = True ) <EOL> self . queue_consumer . register_provider ( self ) <EOL> self . _registered = True <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . _providers_registered : <EOL> self . queue_consumer . unregister_provider ( self ) <EOL> self . _unregistered_from_queue_consumer . send ( True ) <EOL> def unregister_provider ( self , provider ) : <EOL> """<STR_LIT>""" <EOL> self . _unregistering_providers . add ( provider ) <EOL> remaining_providers = self . _providers - self . _unregistering_providers <EOL> if not remaining_providers : <EOL> _log . debug ( '<STR_LIT>' , self ) <EOL> self . queue_consumer . unregister_provider ( self ) <EOL> _log . debug ( '<STR_LIT>' , self ) <EOL> self . _unregistered_from_queue_consumer . send ( True ) <EOL> _log . debug ( '<STR_LIT>' , self ) <EOL> self . _unregistered_from_queue_consumer . wait ( ) <EOL> super ( RpcConsumer , self ) . unregister_provider ( provider ) <EOL> def get_provider_for_method ( self , routing_key ) : <EOL> service_name = self . container . service_name <EOL> for provider in self . _providers : <EOL> key = '<STR_LIT>' . format ( service_name , provider . method_name ) <EOL> if key == routing_key : <EOL> return provider <EOL> else : <EOL> method_name = routing_key . split ( "<STR_LIT:.>" ) [ - <NUM_LIT:1> ] <EOL> raise MethodNotFound ( method_name ) <EOL> def handle_message ( self , body , message ) : <EOL> routing_key = message . delivery_info [ '<STR_LIT>' ] <EOL> try : <EOL> provider = self . get_provider_for_method ( routing_key ) <EOL> provider . handle_message ( body , message ) <EOL> except Exception : <EOL> exc_info = sys . exc_info ( ) <EOL> self . handle_result ( message , None , exc_info ) <EOL> def handle_result ( self , message , result , exc_info ) : <EOL> responder = Responder ( self . container . config , message ) <EOL> result , exc_info = responder . send_response ( result , exc_info ) <EOL> self . queue_consumer . ack_message ( message ) <EOL> return result , exc_info <EOL> def requeue_message ( self , message ) : <EOL> self . queue_consumer . requeue_message ( message ) <EOL> class Rpc ( Entrypoint , HeaderDecoder ) : <EOL> rpc_consumer = RpcConsumer ( ) <EOL> def __init__ ( self , expected_exceptions = ( ) , sensitive_variables = ( ) ) : <EOL> """<STR_LIT>""" <EOL> self . expected_exceptions = expected_exceptions <EOL> self . sensitive_variables = sensitive_variables <EOL> def setup ( self ) : <EOL> self . rpc_consumer . register_provider ( self ) <EOL> def stop ( self ) : <EOL> self . rpc_consumer . unregister_provider ( self ) <EOL> def handle_message ( self , body , message ) : <EOL> try : <EOL> args = body [ '<STR_LIT:args>' ] <EOL> kwargs = body [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> raise MalformedRequest ( '<STR_LIT>' ) <EOL> self . check_signature ( args , kwargs ) <EOL> worker_ctx_cls = self . container . worker_ctx_cls <EOL> context_data = self . unpack_message_headers ( worker_ctx_cls , message ) <EOL> handle_result = partial ( self . handle_result , message ) <EOL> try : <EOL> self . container . spawn_worker ( self , args , kwargs , <EOL> context_data = context_data , <EOL> handle_result = handle_result ) <EOL> except ContainerBeingKilled : <EOL> self . rpc_consumer . requeue_message ( message ) <EOL> def handle_result ( self , message , worker_ctx , result , exc_info ) : <EOL> result , exc_info = self . rpc_consumer . handle_result ( <EOL> message , result , exc_info ) <EOL> return result , exc_info <EOL> rpc = Rpc . decorator <EOL> class Responder ( object ) : <EOL> def __init__ ( self , config , message ) : <EOL> self . config = config <EOL> self . message = message <EOL> def send_response ( self , result , exc_info , ** kwargs ) : <EOL> error = None <EOL> if exc_info is not None : <EOL> error = serialize ( exc_info [ <NUM_LIT:1> ] ) <EOL> serializer = self . config . get ( <EOL> SERIALIZER_CONFIG_KEY , DEFAULT_SERIALIZER ) <EOL> try : <EOL> kombu . serialization . dumps ( result , serializer ) <EOL> except Exception : <EOL> exc_info = sys . exc_info ( ) <EOL> error = serialize ( UnserializableValueError ( result ) ) <EOL> result = None <EOL> conn = Connection ( self . config [ AMQP_URI_CONFIG_KEY ] ) <EOL> exchange = get_rpc_exchange ( self . config ) <EOL> retry = kwargs . pop ( '<STR_LIT>' , True ) <EOL> retry_policy = kwargs . pop ( '<STR_LIT>' , DEFAULT_RETRY_POLICY ) <EOL> with producers [ conn ] . acquire ( block = True ) as producer : <EOL> routing_key = self . message . properties [ '<STR_LIT>' ] <EOL> correlation_id = self . message . properties . get ( '<STR_LIT>' ) <EOL> msg = { '<STR_LIT:result>' : result , '<STR_LIT:error>' : error } <EOL> _log . debug ( '<STR_LIT>' , routing_key , correlation_id ) <EOL> producer . publish ( <EOL> msg , retry = retry , retry_policy = retry_policy , <EOL> exchange = exchange , routing_key = routing_key , <EOL> serializer = serializer , <EOL> correlation_id = correlation_id , ** kwargs ) <EOL> return result , exc_info <EOL> class ReplyListener ( SharedExtension ) : <EOL> queue_consumer = QueueConsumer ( ) <EOL> def __init__ ( self , ** kwargs ) : <EOL> self . _reply_events = { } <EOL> super ( ReplyListener , self ) . __init__ ( ** kwargs ) <EOL> def setup ( self ) : <EOL> service_uuid = uuid . uuid4 ( ) <EOL> service_name = self . container . service_name <EOL> queue_name = RPC_REPLY_QUEUE_TEMPLATE . format ( <EOL> service_name , service_uuid ) <EOL> self . routing_key = str ( service_uuid ) <EOL> exchange = get_rpc_exchange ( self . container . config ) <EOL> self . queue = Queue ( <EOL> queue_name , <EOL> exchange = exchange , <EOL> routing_key = self . routing_key , <EOL> auto_delete = True , <EOL> exclusive = True , <EOL> ) <EOL> self . queue_consumer . register_provider ( self ) <EOL> def stop ( self ) : <EOL> self . queue_consumer . unregister_provider ( self ) <EOL> super ( ReplyListener , self ) . stop ( ) <EOL> def get_reply_event ( self , correlation_id ) : <EOL> reply_event = Event ( ) <EOL> self . _reply_events [ correlation_id ] = reply_event <EOL> return reply_event <EOL> def on_consume_ready ( self ) : <EOL> for event in self . _reply_events . values ( ) : <EOL> event . send_exception ( <EOL> RpcConnectionError ( '<STR_LIT>' ) <EOL> ) <EOL> self . _reply_events . clear ( ) <EOL> def handle_message ( self , body , message ) : <EOL> self . queue_consumer . ack_message ( message ) <EOL> correlation_id = message . properties . get ( '<STR_LIT>' ) <EOL> client_event = self . _reply_events . pop ( correlation_id , None ) <EOL> if client_event is not None : <EOL> client_event . send ( body ) <EOL> else : <EOL> _log . debug ( "<STR_LIT>" , correlation_id ) <EOL> class RpcProxy ( DependencyProvider ) : <EOL> rpc_reply_listener = ReplyListener ( ) <EOL> def __init__ ( self , target_service ) : <EOL> self . target_service = target_service <EOL> def get_dependency ( self , worker_ctx ) : <EOL> return ServiceProxy ( worker_ctx , self . target_service , <EOL> self . rpc_reply_listener ) <EOL> class ServiceProxy ( object ) : <EOL> def __init__ ( self , worker_ctx , service_name , reply_listener ) : <EOL> self . worker_ctx = worker_ctx <EOL> self . service_name = service_name <EOL> self . reply_listener = reply_listener <EOL> def __getattr__ ( self , name ) : <EOL> return MethodProxy ( <EOL> self . worker_ctx , self . service_name , name , self . reply_listener ) <EOL> class RpcReply ( object ) : <EOL> resp_body = None <EOL> def __init__ ( self , reply_event ) : <EOL> self . reply_event = reply_event <EOL> def result ( self ) : <EOL> _log . debug ( '<STR_LIT>' , self ) <EOL> if self . resp_body is None : <EOL> self . resp_body = self . reply_event . wait ( ) <EOL> _log . debug ( '<STR_LIT>' , self , self . resp_body ) <EOL> error = self . resp_body . get ( '<STR_LIT:error>' ) <EOL> if error : <EOL> raise deserialize ( error ) <EOL> return self . resp_body [ '<STR_LIT:result>' ] <EOL> class MethodProxy ( HeaderEncoder ) : <EOL> def __init__ ( self , worker_ctx , service_name , method_name , reply_listener ) : <EOL> self . worker_ctx = worker_ctx <EOL> self . service_name = service_name <EOL> self . method_name = method_name <EOL> self . reply_listener = reply_listener <EOL> def __call__ ( self , * args , ** kwargs ) : <EOL> reply = self . _call ( * args , ** kwargs ) <EOL> return reply . result ( ) <EOL> def async ( self , * args , ** kwargs ) : <EOL> reply = self . _call ( * args , ** kwargs ) <EOL> return reply <EOL> def _call ( self , * args , ** kwargs ) : <EOL> _log . debug ( '<STR_LIT>' , self ) <EOL> worker_ctx = self . worker_ctx <EOL> container = worker_ctx . container <EOL> msg = { '<STR_LIT:args>' : args , '<STR_LIT>' : kwargs } <EOL> conn = Connection ( <EOL> container . config [ AMQP_URI_CONFIG_KEY ] , <EOL> transport_options = { '<STR_LIT>' : True } , <EOL> ) <EOL> serializer = container . config . get ( <EOL> SERIALIZER_CONFIG_KEY , DEFAULT_SERIALIZER ) <EOL> routing_key = '<STR_LIT>' . format ( self . service_name , self . method_name ) <EOL> exchange = get_rpc_exchange ( container . config ) <EOL> with producers [ conn ] . acquire ( block = True ) as producer : <EOL> headers = self . get_message_headers ( worker_ctx ) <EOL> correlation_id = str ( uuid . uuid4 ( ) ) <EOL> reply_listener = self . reply_listener <EOL> reply_to_routing_key = reply_listener . routing_key <EOL> reply_event = reply_listener . get_reply_event ( correlation_id ) <EOL> producer . publish ( <EOL> msg , <EOL> exchange = exchange , <EOL> routing_key = routing_key , <EOL> mandatory = True , <EOL> serializer = serializer , <EOL> reply_to = reply_to_routing_key , <EOL> headers = headers , <EOL> correlation_id = correlation_id , <EOL> retry = True , <EOL> retry_policy = DEFAULT_RETRY_POLICY <EOL> ) <EOL> try : <EOL> producer . channel . returned_messages . get_nowait ( ) <EOL> except Empty : <EOL> pass <EOL> else : <EOL> raise UnknownService ( self . service_name ) <EOL> return RpcReply ( reply_event ) <EOL> def __repr__ ( self ) : <EOL> service_name = self . service_name <EOL> method_name = self . method_name <EOL> return '<STR_LIT>' . format ( service_name , method_name ) </s>
