<s> import kombu <EOL> import kombu . entity <EOL> import kombu . messaging <EOL> import kombu . connection <EOL> import itertools <EOL> import sys <EOL> import time <EOL> import traceback <EOL> import types <EOL> import uuid <EOL> import eventlet <EOL> from eventlet import greenpool <EOL> from eventlet import pools <EOL> import greenlet <EOL> from nova import context <EOL> from nova import exception <EOL> from nova import flags <EOL> from nova . rpc . common import RemoteError , LOG <EOL> eventlet . monkey_patch ( ) <EOL> FLAGS = flags . FLAGS <EOL> class ConsumerBase ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , channel , callback , tag , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . callback = callback <EOL> self . tag = str ( tag ) <EOL> self . kwargs = kwargs <EOL> self . queue = None <EOL> self . reconnect ( channel ) <EOL> def reconnect ( self , channel ) : <EOL> """<STR_LIT>""" <EOL> self . channel = channel <EOL> self . kwargs [ '<STR_LIT>' ] = channel <EOL> self . queue = kombu . entity . Queue ( ** self . kwargs ) <EOL> self . queue . declare ( ) <EOL> def consume ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> options = { '<STR_LIT>' : self . tag } <EOL> options [ '<STR_LIT>' ] = kwargs . get ( '<STR_LIT>' , False ) <EOL> callback = kwargs . get ( '<STR_LIT>' , self . callback ) <EOL> if not callback : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> def _callback ( raw_message ) : <EOL> message = self . channel . message_to_python ( raw_message ) <EOL> callback ( message . payload ) <EOL> message . ack ( ) <EOL> self . queue . consume ( * args , callback = _callback , ** options ) <EOL> def cancel ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . queue . cancel ( self . tag ) <EOL> except KeyError , e : <EOL> if str ( e ) != "<STR_LIT>" % self . tag : <EOL> raise <EOL> self . queue = None <EOL> class DirectConsumer ( ConsumerBase ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , channel , msg_id , callback , tag , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> options = { '<STR_LIT>' : False , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True } <EOL> options . update ( kwargs ) <EOL> exchange = kombu . entity . Exchange ( <EOL> name = msg_id , <EOL> type = '<STR_LIT>' , <EOL> durable = options [ '<STR_LIT>' ] , <EOL> auto_delete = options [ '<STR_LIT>' ] ) <EOL> super ( DirectConsumer , self ) . __init__ ( <EOL> channel , <EOL> callback , <EOL> tag , <EOL> name = msg_id , <EOL> exchange = exchange , <EOL> routing_key = msg_id , <EOL> ** options ) <EOL> class TopicConsumer ( ConsumerBase ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , channel , topic , callback , tag , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> options = { '<STR_LIT>' : FLAGS . rabbit_durable_queues , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False } <EOL> options . update ( kwargs ) <EOL> exchange = kombu . entity . Exchange ( <EOL> name = FLAGS . control_exchange , <EOL> type = '<STR_LIT>' , <EOL> durable = options [ '<STR_LIT>' ] , <EOL> auto_delete = options [ '<STR_LIT>' ] ) <EOL> super ( TopicConsumer , self ) . __init__ ( <EOL> channel , <EOL> callback , <EOL> tag , <EOL> name = topic , <EOL> exchange = exchange , <EOL> routing_key = topic , <EOL> ** options ) <EOL> class FanoutConsumer ( ConsumerBase ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , channel , topic , callback , tag , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> unique = uuid . uuid4 ( ) . hex <EOL> exchange_name = '<STR_LIT>' % topic <EOL> queue_name = '<STR_LIT>' % ( topic , unique ) <EOL> options = { '<STR_LIT>' : False , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True } <EOL> options . update ( kwargs ) <EOL> exchange = kombu . entity . Exchange ( <EOL> name = exchange_name , <EOL> type = '<STR_LIT>' , <EOL> durable = options [ '<STR_LIT>' ] , <EOL> auto_delete = options [ '<STR_LIT>' ] ) <EOL> super ( FanoutConsumer , self ) . __init__ ( <EOL> channel , <EOL> callback , <EOL> tag , <EOL> name = queue_name , <EOL> exchange = exchange , <EOL> routing_key = topic , <EOL> ** options ) <EOL> class Publisher ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , channel , exchange_name , routing_key , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . exchange_name = exchange_name <EOL> self . routing_key = routing_key <EOL> self . kwargs = kwargs <EOL> self . reconnect ( channel ) <EOL> def reconnect ( self , channel ) : <EOL> """<STR_LIT>""" <EOL> self . exchange = kombu . entity . Exchange ( name = self . exchange_name , <EOL> ** self . kwargs ) <EOL> self . producer = kombu . messaging . Producer ( exchange = self . exchange , <EOL> channel = channel , routing_key = self . routing_key ) <EOL> def send ( self , msg ) : <EOL> """<STR_LIT>""" <EOL> self . producer . publish ( msg ) <EOL> class DirectPublisher ( Publisher ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , channel , msg_id , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> options = { '<STR_LIT>' : False , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True } <EOL> options . update ( kwargs ) <EOL> super ( DirectPublisher , self ) . __init__ ( channel , <EOL> msg_id , <EOL> msg_id , <EOL> type = '<STR_LIT>' , <EOL> ** options ) <EOL> class TopicPublisher ( Publisher ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , channel , topic , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> options = { '<STR_LIT>' : FLAGS . rabbit_durable_queues , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False } <EOL> options . update ( kwargs ) <EOL> super ( TopicPublisher , self ) . __init__ ( channel , <EOL> FLAGS . control_exchange , <EOL> topic , <EOL> type = '<STR_LIT>' , <EOL> ** options ) <EOL> class FanoutPublisher ( Publisher ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , channel , topic , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> options = { '<STR_LIT>' : False , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True } <EOL> options . update ( kwargs ) <EOL> super ( FanoutPublisher , self ) . __init__ ( channel , <EOL> '<STR_LIT>' % topic , <EOL> None , <EOL> type = '<STR_LIT>' , <EOL> ** options ) <EOL> class Connection ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . consumers = [ ] <EOL> self . consumer_thread = None <EOL> self . max_retries = FLAGS . rabbit_max_retries <EOL> if self . max_retries <= <NUM_LIT:0> : <EOL> self . max_retries = None <EOL> self . interval_start = FLAGS . rabbit_retry_interval <EOL> self . interval_stepping = FLAGS . rabbit_retry_backoff <EOL> self . interval_max = <NUM_LIT:30> <EOL> self . memory_transport = False <EOL> self . params = dict ( hostname = FLAGS . rabbit_host , <EOL> port = FLAGS . rabbit_port , <EOL> userid = FLAGS . rabbit_userid , <EOL> password = FLAGS . rabbit_password , <EOL> virtual_host = FLAGS . rabbit_virtual_host ) <EOL> if FLAGS . fake_rabbit : <EOL> self . params [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . memory_transport = True <EOL> else : <EOL> self . memory_transport = False <EOL> self . connection = None <EOL> self . reconnect ( ) <EOL> def reconnect ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . connection : <EOL> try : <EOL> self . connection . close ( ) <EOL> except self . connection . connection_errors : <EOL> pass <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> self . connection = kombu . connection . BrokerConnection ( ** self . params ) <EOL> if self . memory_transport : <EOL> self . connection . transport . polling_interval = <NUM_LIT:0.0> <EOL> self . consumer_num = itertools . count ( <NUM_LIT:1> ) <EOL> try : <EOL> self . connection . ensure_connection ( errback = self . connect_error , <EOL> max_retries = self . max_retries , <EOL> interval_start = self . interval_start , <EOL> interval_step = self . interval_stepping , <EOL> interval_max = self . interval_max ) <EOL> except self . connection . connection_errors , e : <EOL> err_str = str ( e ) <EOL> max_retries = self . max_retries <EOL> LOG . error ( _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % locals ( ) ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> LOG . info ( _ ( '<STR_LIT>' % <EOL> self . params ) ) <EOL> self . channel = self . connection . channel ( ) <EOL> if self . memory_transport : <EOL> self . channel . _new_queue ( '<STR_LIT>' ) <EOL> for consumer in self . consumers : <EOL> consumer . reconnect ( self . channel ) <EOL> if self . consumers : <EOL> LOG . debug ( _ ( "<STR_LIT>" ) ) <EOL> def get_channel ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . channel <EOL> def connect_error ( self , exc , interval ) : <EOL> """<STR_LIT>""" <EOL> info = self . params . copy ( ) <EOL> info [ '<STR_LIT>' ] = interval <EOL> info [ '<STR_LIT:e>' ] = exc <EOL> LOG . error ( _ ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) % info ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cancel_consumer_thread ( ) <EOL> self . connection . release ( ) <EOL> self . connection = None <EOL> def reset ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cancel_consumer_thread ( ) <EOL> self . channel . close ( ) <EOL> self . channel = self . connection . channel ( ) <EOL> if self . memory_transport : <EOL> self . channel . _new_queue ( '<STR_LIT>' ) <EOL> self . consumers = [ ] <EOL> def declare_consumer ( self , consumer_cls , topic , callback ) : <EOL> """<STR_LIT>""" <EOL> consumer = consumer_cls ( self . channel , topic , callback , <EOL> self . consumer_num . next ( ) ) <EOL> self . consumers . append ( consumer ) <EOL> return consumer <EOL> def iterconsume ( self , limit = None ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> try : <EOL> queues_head = self . consumers [ : - <NUM_LIT:1> ] <EOL> queues_tail = self . consumers [ - <NUM_LIT:1> ] <EOL> for queue in queues_head : <EOL> queue . consume ( nowait = True ) <EOL> queues_tail . consume ( nowait = False ) <EOL> for iteration in itertools . count ( <NUM_LIT:0> ) : <EOL> if limit and iteration >= limit : <EOL> raise StopIteration <EOL> yield self . connection . drain_events ( ) <EOL> except self . connection . connection_errors , e : <EOL> LOG . exception ( _ ( '<STR_LIT>' <EOL> '<STR_LIT:%s>' % str ( e ) ) ) <EOL> self . reconnect ( ) <EOL> def cancel_consumer_thread ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . consumer_thread is not None : <EOL> self . consumer_thread . kill ( ) <EOL> try : <EOL> self . consumer_thread . wait ( ) <EOL> except greenlet . GreenletExit : <EOL> pass <EOL> self . consumer_thread = None <EOL> def publisher_send ( self , cls , topic , msg ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> publisher = None <EOL> try : <EOL> publisher = cls ( self . channel , topic ) <EOL> publisher . send ( msg ) <EOL> return <EOL> except self . connection . connection_errors , e : <EOL> LOG . exception ( _ ( '<STR_LIT>' % str ( e ) ) ) <EOL> try : <EOL> self . reconnect ( ) <EOL> if publisher : <EOL> publisher . reconnect ( self . channel ) <EOL> except self . connection . connection_errors , e : <EOL> pass <EOL> def declare_direct_consumer ( self , topic , callback ) : <EOL> """<STR_LIT>""" <EOL> self . declare_consumer ( DirectConsumer , topic , callback ) <EOL> def declare_topic_consumer ( self , topic , callback = None ) : <EOL> """<STR_LIT>""" <EOL> self . declare_consumer ( TopicConsumer , topic , callback ) <EOL> def declare_fanout_consumer ( self , topic , callback ) : <EOL> """<STR_LIT>""" <EOL> self . declare_consumer ( FanoutConsumer , topic , callback ) <EOL> def direct_send ( self , msg_id , msg ) : <EOL> """<STR_LIT>""" <EOL> self . publisher_send ( DirectPublisher , msg_id , msg ) <EOL> def topic_send ( self , topic , msg ) : <EOL> """<STR_LIT>""" <EOL> self . publisher_send ( TopicPublisher , topic , msg ) <EOL> def fanout_send ( self , topic , msg ) : <EOL> """<STR_LIT>""" <EOL> self . publisher_send ( FanoutPublisher , topic , msg ) <EOL> def consume ( self , limit = None ) : <EOL> """<STR_LIT>""" <EOL> it = self . iterconsume ( limit = limit ) <EOL> while True : <EOL> try : <EOL> it . next ( ) <EOL> except StopIteration : <EOL> return <EOL> def consume_in_thread ( self ) : <EOL> """<STR_LIT>""" <EOL> def _consumer_thread ( ) : <EOL> try : <EOL> self . consume ( ) <EOL> except greenlet . GreenletExit : <EOL> return <EOL> if self . consumer_thread is None : <EOL> self . consumer_thread = eventlet . spawn ( _consumer_thread ) <EOL> return self . consumer_thread <EOL> def create_consumer ( self , topic , proxy , fanout = False ) : <EOL> """<STR_LIT>""" <EOL> if fanout : <EOL> self . declare_fanout_consumer ( topic , ProxyCallback ( proxy ) ) <EOL> else : <EOL> self . declare_topic_consumer ( topic , ProxyCallback ( proxy ) ) <EOL> class Pool ( pools . Pool ) : <EOL> """<STR_LIT>""" <EOL> def create ( self ) : <EOL> LOG . debug ( '<STR_LIT>' ) <EOL> return Connection ( ) <EOL> ConnectionPool = Pool ( <EOL> max_size = FLAGS . rpc_conn_pool_size , <EOL> order_as_stack = True ) <EOL> class ConnectionContext ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , pooled = True ) : <EOL> """<STR_LIT>""" <EOL> self . connection = None <EOL> if pooled : <EOL> self . connection = ConnectionPool . get ( ) <EOL> else : <EOL> self . connection = Connection ( ) <EOL> self . pooled = pooled <EOL> def __enter__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return self <EOL> def _done ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . connection : <EOL> if self . pooled : <EOL> self . connection . reset ( ) <EOL> ConnectionPool . put ( self . connection ) <EOL> else : <EOL> try : <EOL> self . connection . close ( ) <EOL> except Exception : <EOL> pass <EOL> self . connection = None <EOL> def __exit__ ( self , t , v , tb ) : <EOL> """<STR_LIT>""" <EOL> self . _done ( ) <EOL> def __del__ ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _done ( ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _done ( ) <EOL> def __getattr__ ( self , key ) : <EOL> """<STR_LIT>""" <EOL> if self . connection : <EOL> return getattr ( self . connection , key ) <EOL> else : <EOL> raise exception . InvalidRPCConnectionReuse ( ) <EOL> class ProxyCallback ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , proxy ) : <EOL> self . proxy = proxy <EOL> self . pool = greenpool . GreenPool ( FLAGS . rpc_thread_pool_size ) <EOL> def __call__ ( self , message_data ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( _ ( '<STR_LIT>' ) % message_data ) <EOL> ctxt = _unpack_context ( message_data ) <EOL> method = message_data . get ( '<STR_LIT>' ) <EOL> args = message_data . get ( '<STR_LIT:args>' , { } ) <EOL> if not method : <EOL> LOG . warn ( _ ( '<STR_LIT>' ) % message_data ) <EOL> ctxt . reply ( _ ( '<STR_LIT>' ) % message_data ) <EOL> return <EOL> self . pool . spawn_n ( self . _process_data , ctxt , method , args ) <EOL> @ exception . wrap_exception ( ) <EOL> def _process_data ( self , ctxt , method , args ) : <EOL> """<STR_LIT>""" <EOL> node_func = getattr ( self . proxy , str ( method ) ) <EOL> node_args = dict ( ( str ( k ) , v ) for k , v in args . iteritems ( ) ) <EOL> try : <EOL> rval = node_func ( context = ctxt , ** node_args ) <EOL> if isinstance ( rval , types . GeneratorType ) : <EOL> for x in rval : <EOL> ctxt . reply ( x , None ) <EOL> else : <EOL> ctxt . reply ( rval , None ) <EOL> ctxt . reply ( None , None ) <EOL> except Exception as e : <EOL> LOG . exception ( '<STR_LIT>' ) <EOL> ctxt . reply ( None , sys . exc_info ( ) ) <EOL> return <EOL> def _unpack_context ( msg ) : <EOL> """<STR_LIT>""" <EOL> context_dict = { } <EOL> for key in list ( msg . keys ( ) ) : <EOL> key = str ( key ) <EOL> if key . startswith ( '<STR_LIT>' ) : <EOL> value = msg . pop ( key ) <EOL> context_dict [ key [ <NUM_LIT:9> : ] ] = value <EOL> context_dict [ '<STR_LIT>' ] = msg . pop ( '<STR_LIT>' , None ) <EOL> LOG . debug ( _ ( '<STR_LIT>' ) , context_dict ) <EOL> return RpcContext . from_dict ( context_dict ) <EOL> def _pack_context ( msg , context ) : <EOL> """<STR_LIT>""" <EOL> context_d = dict ( [ ( '<STR_LIT>' % key , value ) <EOL> for ( key , value ) in context . to_dict ( ) . iteritems ( ) ] ) <EOL> msg . update ( context_d ) <EOL> class RpcContext ( context . RequestContext ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> msg_id = kwargs . pop ( '<STR_LIT>' , None ) <EOL> self . msg_id = msg_id <EOL> super ( RpcContext , self ) . __init__ ( * args , ** kwargs ) <EOL> def reply ( self , * args , ** kwargs ) : <EOL> if self . msg_id : <EOL> msg_reply ( self . msg_id , * args , ** kwargs ) <EOL> class MulticallWaiter ( object ) : <EOL> def __init__ ( self , connection ) : <EOL> self . _connection = connection <EOL> self . _iterator = connection . iterconsume ( ) <EOL> self . _result = None <EOL> self . _done = False <EOL> def done ( self ) : <EOL> self . _done = True <EOL> self . _connection . close ( ) <EOL> def __call__ ( self , data ) : <EOL> """<STR_LIT>""" <EOL> if data [ '<STR_LIT>' ] : <EOL> self . _result = RemoteError ( * data [ '<STR_LIT>' ] ) <EOL> else : <EOL> self . _result = data [ '<STR_LIT:result>' ] <EOL> def __iter__ ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _done : <EOL> raise StopIteration <EOL> while True : <EOL> self . _iterator . next ( ) <EOL> result = self . _result <EOL> if isinstance ( result , Exception ) : <EOL> self . done ( ) <EOL> raise result <EOL> if result == None : <EOL> self . done ( ) <EOL> raise StopIteration <EOL> yield result <EOL> def create_connection ( new = True ) : <EOL> """<STR_LIT>""" <EOL> return ConnectionContext ( pooled = not new ) <EOL> def multicall ( context , topic , msg ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( _ ( '<STR_LIT>' ) , topic ) <EOL> msg_id = uuid . uuid4 ( ) . hex <EOL> msg . update ( { '<STR_LIT>' : msg_id } ) <EOL> LOG . debug ( _ ( '<STR_LIT>' ) % ( msg_id ) ) <EOL> _pack_context ( msg , context ) <EOL> conn = ConnectionContext ( ) <EOL> wait_msg = MulticallWaiter ( conn ) <EOL> conn . declare_direct_consumer ( msg_id , wait_msg ) <EOL> conn . topic_send ( topic , msg ) <EOL> return wait_msg <EOL> def call ( context , topic , msg ) : <EOL> """<STR_LIT>""" <EOL> rv = multicall ( context , topic , msg ) <EOL> rv = list ( rv ) <EOL> if not rv : <EOL> return <EOL> return rv [ - <NUM_LIT:1> ] <EOL> def cast ( context , topic , msg ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( _ ( '<STR_LIT>' ) , topic ) <EOL> _pack_context ( msg , context ) <EOL> with ConnectionContext ( ) as conn : <EOL> conn . topic_send ( topic , msg ) <EOL> def fanout_cast ( context , topic , msg ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( _ ( '<STR_LIT>' ) ) <EOL> _pack_context ( msg , context ) <EOL> with ConnectionContext ( ) as conn : <EOL> conn . fanout_send ( topic , msg ) <EOL> def msg_reply ( msg_id , reply = None , failure = None ) : <EOL> """<STR_LIT>""" <EOL> with ConnectionContext ( ) as conn : <EOL> if failure : <EOL> message = str ( failure [ <NUM_LIT:1> ] ) <EOL> tb = traceback . format_exception ( * failure ) <EOL> LOG . error ( _ ( "<STR_LIT>" ) , message ) <EOL> LOG . error ( tb ) <EOL> failure = ( failure [ <NUM_LIT:0> ] . __name__ , str ( failure [ <NUM_LIT:1> ] ) , tb ) <EOL> try : <EOL> msg = { '<STR_LIT:result>' : reply , '<STR_LIT>' : failure } <EOL> except TypeError : <EOL> msg = { '<STR_LIT:result>' : dict ( ( k , repr ( v ) ) <EOL> for k , v in reply . __dict__ . iteritems ( ) ) , <EOL> '<STR_LIT>' : failure } <EOL> conn . direct_send ( msg_id , msg ) </s>
