<s> import functools <EOL> import itertools <EOL> import time <EOL> import uuid <EOL> import eventlet <EOL> import greenlet <EOL> from oslo . config import cfg <EOL> from openstack_dashboard . openstack . common . gettextutils import _ <EOL> from openstack_dashboard . openstack . common import importutils <EOL> from openstack_dashboard . openstack . common import jsonutils <EOL> from openstack_dashboard . openstack . common import log as logging <EOL> from openstack_dashboard . openstack . common . rpc import amqp as rpc_amqp <EOL> from openstack_dashboard . openstack . common . rpc import common as rpc_common <EOL> qpid_codec = importutils . try_import ( "<STR_LIT>" ) <EOL> qpid_messaging = importutils . try_import ( "<STR_LIT>" ) <EOL> qpid_exceptions = importutils . try_import ( "<STR_LIT>" ) <EOL> LOG = logging . getLogger ( __name__ ) <EOL> qpid_opts = [ <EOL> cfg . StrOpt ( '<STR_LIT>' , <EOL> default = '<STR_LIT:localhost>' , <EOL> help = '<STR_LIT>' ) , <EOL> cfg . IntOpt ( '<STR_LIT>' , <EOL> default = <NUM_LIT> , <EOL> help = '<STR_LIT>' ) , <EOL> cfg . ListOpt ( '<STR_LIT>' , <EOL> default = [ '<STR_LIT>' ] , <EOL> help = '<STR_LIT>' ) , <EOL> cfg . StrOpt ( '<STR_LIT>' , <EOL> default = '<STR_LIT>' , <EOL> help = '<STR_LIT>' ) , <EOL> cfg . StrOpt ( '<STR_LIT>' , <EOL> default = '<STR_LIT>' , <EOL> help = '<STR_LIT>' , <EOL> secret = True ) , <EOL> cfg . StrOpt ( '<STR_LIT>' , <EOL> default = '<STR_LIT>' , <EOL> help = '<STR_LIT>' ) , <EOL> cfg . IntOpt ( '<STR_LIT>' , <EOL> default = <NUM_LIT> , <EOL> help = '<STR_LIT>' ) , <EOL> cfg . StrOpt ( '<STR_LIT>' , <EOL> default = '<STR_LIT>' , <EOL> help = "<STR_LIT>" ) , <EOL> cfg . BoolOpt ( '<STR_LIT>' , <EOL> default = True , <EOL> help = '<STR_LIT>' ) , <EOL> ] <EOL> cfg . CONF . register_opts ( qpid_opts ) <EOL> JSON_CONTENT_TYPE = '<STR_LIT>' <EOL> class ConsumerBase ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , session , callback , node_name , node_opts , <EOL> link_name , link_opts ) : <EOL> """<STR_LIT>""" <EOL> self . callback = callback <EOL> self . receiver = None <EOL> self . session = None <EOL> addr_opts = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT>" : True , <EOL> } , <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:name>" : link_name , <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT>" : False , <EOL> } , <EOL> } , <EOL> } <EOL> addr_opts [ "<STR_LIT>" ] [ "<STR_LIT>" ] . update ( node_opts ) <EOL> addr_opts [ "<STR_LIT>" ] [ "<STR_LIT>" ] . update ( link_opts ) <EOL> self . address = "<STR_LIT>" % ( node_name , jsonutils . dumps ( addr_opts ) ) <EOL> self . reconnect ( session ) <EOL> def reconnect ( self , session ) : <EOL> """<STR_LIT>""" <EOL> self . session = session <EOL> self . receiver = session . receiver ( self . address ) <EOL> self . receiver . capacity = <NUM_LIT:1> <EOL> def _unpack_json_msg ( self , msg ) : <EOL> """<STR_LIT>""" <EOL> if msg . content_type == JSON_CONTENT_TYPE : <EOL> msg . content = jsonutils . loads ( msg . content ) <EOL> msg . content_type = '<STR_LIT>' <EOL> def consume ( self ) : <EOL> """<STR_LIT>""" <EOL> message = self . receiver . fetch ( ) <EOL> try : <EOL> self . _unpack_json_msg ( message ) <EOL> msg = rpc_common . deserialize_msg ( message . content ) <EOL> self . callback ( msg ) <EOL> except Exception : <EOL> LOG . exception ( _ ( "<STR_LIT>" ) ) <EOL> finally : <EOL> self . session . acknowledge ( message ) <EOL> def get_receiver ( self ) : <EOL> return self . receiver <EOL> class DirectConsumer ( ConsumerBase ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , conf , session , msg_id , callback ) : <EOL> """<STR_LIT>""" <EOL> super ( DirectConsumer , self ) . __init__ ( session , callback , <EOL> "<STR_LIT>" % ( msg_id , msg_id ) , <EOL> { "<STR_LIT:type>" : "<STR_LIT>" } , <EOL> msg_id , <EOL> { "<STR_LIT>" : True } ) <EOL> class TopicConsumer ( ConsumerBase ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , conf , session , topic , callback , name = None , <EOL> exchange_name = None ) : <EOL> """<STR_LIT>""" <EOL> exchange_name = exchange_name or rpc_amqp . get_control_exchange ( conf ) <EOL> super ( TopicConsumer , self ) . __init__ ( session , callback , <EOL> "<STR_LIT>" % ( exchange_name , topic ) , <EOL> { } , name or topic , { } ) <EOL> class FanoutConsumer ( ConsumerBase ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , conf , session , topic , callback ) : <EOL> """<STR_LIT>""" <EOL> super ( FanoutConsumer , self ) . __init__ ( <EOL> session , callback , <EOL> "<STR_LIT>" % topic , <EOL> { "<STR_LIT>" : False , "<STR_LIT:type>" : "<STR_LIT>" } , <EOL> "<STR_LIT>" % ( topic , uuid . uuid4 ( ) . hex ) , <EOL> { "<STR_LIT>" : True } ) <EOL> class Publisher ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , session , node_name , node_opts = None ) : <EOL> """<STR_LIT>""" <EOL> self . sender = None <EOL> self . session = session <EOL> addr_opts = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : True , <EOL> } , <EOL> } , <EOL> } <EOL> if node_opts : <EOL> addr_opts [ "<STR_LIT>" ] [ "<STR_LIT>" ] . update ( node_opts ) <EOL> self . address = "<STR_LIT>" % ( node_name , jsonutils . dumps ( addr_opts ) ) <EOL> self . reconnect ( session ) <EOL> def reconnect ( self , session ) : <EOL> """<STR_LIT>""" <EOL> self . sender = session . sender ( self . address ) <EOL> def _pack_json_msg ( self , msg ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> msg . content = jsonutils . dumps ( msg . content ) <EOL> except AttributeError : <EOL> msg = qpid_messaging . Message ( jsonutils . dumps ( msg ) ) <EOL> msg . content_type = JSON_CONTENT_TYPE <EOL> return msg <EOL> def send ( self , msg ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> check_msg = msg <EOL> if not hasattr ( check_msg , '<STR_LIT>' ) : <EOL> check_msg = qpid_messaging . Message ( msg ) <EOL> content_type = check_msg . content_type <EOL> enc , dec = qpid_messaging . message . get_codec ( content_type ) <EOL> enc ( check_msg . content ) <EOL> except qpid_codec . CodecException : <EOL> msg = self . _pack_json_msg ( msg ) <EOL> self . sender . send ( msg ) <EOL> class DirectPublisher ( Publisher ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , conf , session , msg_id ) : <EOL> """<STR_LIT>""" <EOL> super ( DirectPublisher , self ) . __init__ ( session , msg_id , <EOL> { "<STR_LIT:type>" : "<STR_LIT>" } ) <EOL> class TopicPublisher ( Publisher ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , conf , session , topic ) : <EOL> """<STR_LIT>""" <EOL> exchange_name = rpc_amqp . get_control_exchange ( conf ) <EOL> super ( TopicPublisher , self ) . __init__ ( session , <EOL> "<STR_LIT>" % ( exchange_name , topic ) ) <EOL> class FanoutPublisher ( Publisher ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , conf , session , topic ) : <EOL> """<STR_LIT>""" <EOL> super ( FanoutPublisher , self ) . __init__ ( <EOL> session , <EOL> "<STR_LIT>" % topic , { "<STR_LIT:type>" : "<STR_LIT>" } ) <EOL> class NotifyPublisher ( Publisher ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , conf , session , topic ) : <EOL> """<STR_LIT>""" <EOL> exchange_name = rpc_amqp . get_control_exchange ( conf ) <EOL> super ( NotifyPublisher , self ) . __init__ ( session , <EOL> "<STR_LIT>" % ( exchange_name , topic ) , <EOL> { "<STR_LIT>" : True } ) <EOL> class Connection ( object ) : <EOL> """<STR_LIT>""" <EOL> pool = None <EOL> def __init__ ( self , conf , server_params = None ) : <EOL> if not qpid_messaging : <EOL> raise ImportError ( "<STR_LIT>" ) <EOL> self . session = None <EOL> self . consumers = { } <EOL> self . consumer_thread = None <EOL> self . proxy_callbacks = [ ] <EOL> self . conf = conf <EOL> if server_params and '<STR_LIT>' in server_params : <EOL> server_params [ '<STR_LIT>' ] = [ <EOL> '<STR_LIT>' % ( server_params [ '<STR_LIT>' ] , <EOL> server_params . get ( '<STR_LIT:port>' , <NUM_LIT> ) ) <EOL> ] <EOL> params = { <EOL> '<STR_LIT>' : self . conf . qpid_hosts , <EOL> '<STR_LIT:username>' : self . conf . qpid_username , <EOL> '<STR_LIT:password>' : self . conf . qpid_password , <EOL> } <EOL> params . update ( server_params or { } ) <EOL> self . brokers = params [ '<STR_LIT>' ] <EOL> self . username = params [ '<STR_LIT:username>' ] <EOL> self . password = params [ '<STR_LIT:password>' ] <EOL> self . connection_create ( self . brokers [ <NUM_LIT:0> ] ) <EOL> self . reconnect ( ) <EOL> def connection_create ( self , broker ) : <EOL> self . connection = qpid_messaging . Connection ( broker ) <EOL> self . connection . username = self . username <EOL> self . connection . password = self . password <EOL> self . connection . sasl_mechanisms = self . conf . qpid_sasl_mechanisms <EOL> self . connection . reconnect = False <EOL> self . connection . heartbeat = self . conf . qpid_heartbeat <EOL> self . connection . transport = self . conf . qpid_protocol <EOL> self . connection . tcp_nodelay = self . conf . qpid_tcp_nodelay <EOL> def _register_consumer ( self , consumer ) : <EOL> self . consumers [ str ( consumer . get_receiver ( ) ) ] = consumer <EOL> def _lookup_consumer ( self , receiver ) : <EOL> return self . consumers [ str ( receiver ) ] <EOL> def reconnect ( self ) : <EOL> """<STR_LIT>""" <EOL> attempt = <NUM_LIT:0> <EOL> delay = <NUM_LIT:1> <EOL> while True : <EOL> if self . connection . opened ( ) : <EOL> try : <EOL> self . connection . close ( ) <EOL> except qpid_exceptions . ConnectionError : <EOL> pass <EOL> broker = self . brokers [ attempt % len ( self . brokers ) ] <EOL> attempt += <NUM_LIT:1> <EOL> try : <EOL> self . connection_create ( broker ) <EOL> self . connection . open ( ) <EOL> except qpid_exceptions . ConnectionError as e : <EOL> msg_dict = dict ( e = e , delay = delay ) <EOL> msg = _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) % msg_dict <EOL> LOG . error ( msg ) <EOL> time . sleep ( delay ) <EOL> delay = min ( <NUM_LIT:2> * delay , <NUM_LIT> ) <EOL> else : <EOL> LOG . info ( _ ( '<STR_LIT>' ) , broker ) <EOL> break <EOL> self . session = self . connection . session ( ) <EOL> if self . consumers : <EOL> consumers = self . consumers <EOL> self . consumers = { } <EOL> for consumer in consumers . itervalues ( ) : <EOL> consumer . reconnect ( self . session ) <EOL> self . _register_consumer ( consumer ) <EOL> LOG . debug ( _ ( "<STR_LIT>" ) ) <EOL> def ensure ( self , error_callback , method , * args , ** kwargs ) : <EOL> while True : <EOL> try : <EOL> return method ( * args , ** kwargs ) <EOL> except ( qpid_exceptions . Empty , <EOL> qpid_exceptions . ConnectionError ) as e : <EOL> if error_callback : <EOL> error_callback ( e ) <EOL> self . reconnect ( ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cancel_consumer_thread ( ) <EOL> self . wait_on_proxy_callbacks ( ) <EOL> try : <EOL> self . connection . close ( ) <EOL> except Exception : <EOL> pass <EOL> self . connection = None <EOL> def reset ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cancel_consumer_thread ( ) <EOL> self . wait_on_proxy_callbacks ( ) <EOL> self . session . close ( ) <EOL> self . session = self . connection . session ( ) <EOL> self . consumers = { } <EOL> def declare_consumer ( self , consumer_cls , topic , callback ) : <EOL> """<STR_LIT>""" <EOL> def _connect_error ( exc ) : <EOL> log_info = { '<STR_LIT>' : topic , '<STR_LIT>' : str ( exc ) } <EOL> LOG . error ( _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) % log_info ) <EOL> def _declare_consumer ( ) : <EOL> consumer = consumer_cls ( self . conf , self . session , topic , callback ) <EOL> self . _register_consumer ( consumer ) <EOL> return consumer <EOL> return self . ensure ( _connect_error , _declare_consumer ) <EOL> def iterconsume ( self , limit = None , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> def _error_callback ( exc ) : <EOL> if isinstance ( exc , qpid_exceptions . Empty ) : <EOL> LOG . debug ( _ ( '<STR_LIT>' ) % <EOL> str ( exc ) ) <EOL> raise rpc_common . Timeout ( ) <EOL> else : <EOL> LOG . exception ( _ ( '<STR_LIT>' ) % <EOL> str ( exc ) ) <EOL> def _consume ( ) : <EOL> nxt_receiver = self . session . next_receiver ( timeout = timeout ) <EOL> try : <EOL> self . _lookup_consumer ( nxt_receiver ) . consume ( ) <EOL> except Exception : <EOL> LOG . exception ( _ ( "<STR_LIT>" ) ) <EOL> for iteration in itertools . count ( <NUM_LIT:0> ) : <EOL> if limit and iteration >= limit : <EOL> raise StopIteration <EOL> yield self . ensure ( _error_callback , _consume ) <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 wait_on_proxy_callbacks ( self ) : <EOL> """<STR_LIT>""" <EOL> for proxy_cb in self . proxy_callbacks : <EOL> proxy_cb . wait ( ) <EOL> def publisher_send ( self , cls , topic , msg ) : <EOL> """<STR_LIT>""" <EOL> def _connect_error ( exc ) : <EOL> log_info = { '<STR_LIT>' : topic , '<STR_LIT>' : str ( exc ) } <EOL> LOG . exception ( _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) % log_info ) <EOL> def _publisher_send ( ) : <EOL> publisher = cls ( self . conf , self . session , topic ) <EOL> publisher . send ( msg ) <EOL> return self . ensure ( _connect_error , _publisher_send ) <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 , queue_name = None , <EOL> exchange_name = None ) : <EOL> """<STR_LIT>""" <EOL> self . declare_consumer ( functools . partial ( TopicConsumer , <EOL> name = queue_name , <EOL> exchange_name = exchange_name , <EOL> ) , <EOL> 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 , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> qpid_message = qpid_messaging . Message ( content = msg , ttl = timeout ) <EOL> self . publisher_send ( TopicPublisher , topic , qpid_message ) <EOL> def fanout_send ( self , topic , msg ) : <EOL> """<STR_LIT>""" <EOL> self . publisher_send ( FanoutPublisher , topic , msg ) <EOL> def notify_send ( self , topic , msg , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . publisher_send ( NotifyPublisher , 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> proxy_cb = rpc_amqp . ProxyCallback ( <EOL> self . conf , proxy , <EOL> rpc_amqp . get_connection_pool ( self . conf , Connection ) ) <EOL> self . proxy_callbacks . append ( proxy_cb ) <EOL> if fanout : <EOL> consumer = FanoutConsumer ( self . conf , self . session , topic , proxy_cb ) <EOL> else : <EOL> consumer = TopicConsumer ( self . conf , self . session , topic , proxy_cb ) <EOL> self . _register_consumer ( consumer ) <EOL> return consumer <EOL> def create_worker ( self , topic , proxy , pool_name ) : <EOL> """<STR_LIT>""" <EOL> proxy_cb = rpc_amqp . ProxyCallback ( <EOL> self . conf , proxy , <EOL> rpc_amqp . get_connection_pool ( self . conf , Connection ) ) <EOL> self . proxy_callbacks . append ( proxy_cb ) <EOL> consumer = TopicConsumer ( self . conf , self . session , topic , proxy_cb , <EOL> name = pool_name ) <EOL> self . _register_consumer ( consumer ) <EOL> return consumer <EOL> def join_consumer_pool ( self , callback , pool_name , topic , <EOL> exchange_name = None ) : <EOL> """<STR_LIT>""" <EOL> callback_wrapper = rpc_amqp . CallbackWrapper ( <EOL> conf = self . conf , <EOL> callback = callback , <EOL> connection_pool = rpc_amqp . get_connection_pool ( self . conf , <EOL> Connection ) , <EOL> ) <EOL> self . proxy_callbacks . append ( callback_wrapper ) <EOL> consumer = TopicConsumer ( conf = self . conf , <EOL> session = self . session , <EOL> topic = topic , <EOL> callback = callback_wrapper , <EOL> name = pool_name , <EOL> exchange_name = exchange_name ) <EOL> self . _register_consumer ( consumer ) <EOL> return consumer <EOL> def create_connection ( conf , new = True ) : <EOL> """<STR_LIT>""" <EOL> return rpc_amqp . create_connection ( <EOL> conf , new , <EOL> rpc_amqp . get_connection_pool ( conf , Connection ) ) <EOL> def multicall ( conf , context , topic , msg , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> return rpc_amqp . multicall ( <EOL> conf , context , topic , msg , timeout , <EOL> rpc_amqp . get_connection_pool ( conf , Connection ) ) <EOL> def call ( conf , context , topic , msg , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> return rpc_amqp . call ( <EOL> conf , context , topic , msg , timeout , <EOL> rpc_amqp . get_connection_pool ( conf , Connection ) ) <EOL> def cast ( conf , context , topic , msg ) : <EOL> """<STR_LIT>""" <EOL> return rpc_amqp . cast ( <EOL> conf , context , topic , msg , <EOL> rpc_amqp . get_connection_pool ( conf , Connection ) ) <EOL> def fanout_cast ( conf , context , topic , msg ) : <EOL> """<STR_LIT>""" <EOL> return rpc_amqp . fanout_cast ( <EOL> conf , context , topic , msg , <EOL> rpc_amqp . get_connection_pool ( conf , Connection ) ) <EOL> def cast_to_server ( conf , context , server_params , topic , msg ) : <EOL> """<STR_LIT>""" <EOL> return rpc_amqp . cast_to_server ( <EOL> conf , context , server_params , topic , msg , <EOL> rpc_amqp . get_connection_pool ( conf , Connection ) ) <EOL> def fanout_cast_to_server ( conf , context , server_params , topic , msg ) : <EOL> """<STR_LIT>""" <EOL> return rpc_amqp . fanout_cast_to_server ( <EOL> conf , context , server_params , topic , msg , <EOL> rpc_amqp . get_connection_pool ( conf , Connection ) ) <EOL> def notify ( conf , context , topic , msg , envelope ) : <EOL> """<STR_LIT>""" <EOL> return rpc_amqp . notify ( conf , context , topic , msg , <EOL> rpc_amqp . get_connection_pool ( conf , Connection ) , <EOL> envelope ) <EOL> def cleanup ( ) : <EOL> return rpc_amqp . cleanup ( Connection . pool ) </s>
