Gemfury

trilio-internal / idna python

Repository URL to install this package:
Details
idna / lib / python2.7 / site-packages / oslo_messaging / _drivers / impl_rabbit.pyc
ó
čEYc!@sØddlZddlZddlZddlZddlZddlZddlZddlZddlZddl	Z	ddl
Z
ddlZddlZddl
Z
ddlZ
ddlZ
ddlZ
ddlmZddlmZddlmZddlmZddlZddlmZddlmZddlmZdd	lm Z dd
lm!Z"ddlm#Z#ddl$m%Z%dd
l$m&Z&ddl$m'Z'ddl$m(Z(ddl)m*Z*ddl)m+Z+dZ,ej-dde.ddddej/dddddddej/dddddddej/d dddd!dd"ej/d#dddd$dd%ej0d&dd'd(d)dd*ej/d+dd,ej1d-dd.dd/dd0ej/d1d2d3d4fdd3dd5ej/d6dd7d(d)d8e2d9d:dd;ej3d<dd=d(d)d8e2d9d:dd>ej4d?dd@gd(d)d8e2d9d:ddAej/dBddCd(d)d8e2d9d:ddDej/dEddCd(d)d8e2d9d:ddFdGe2ej/dHd2dIdJdKfddJd(d)ddLej/dMddNd(d)d8e2d9d:ddOej1dPddQddRej1dSddTd(d)ddUej1dVddWddXej1dYddZd8e2d(d)dd[ej-d\de.d(d)dd]ej1d^d_dQdd`ddaej1dbddZddcej1dddd/ddeej1dfddTddgej-dhde.d(d)ddigZ5ej6e7Z8djZ9dke:fdlYZ;dme<fdnYZ=doe*j>fdpYZ?dqe?fdrYZ@dse<fdtYZAduejBfdvYZCdS(wiÿÿÿÿN(tcfg(tlog(t
eventletutils(tnetutils(tparse(tamqp(t
amqpdriver(tbase(tcommon(tpool(t_(t_LE(t_LI(t_LW(t_utils(t
exceptionsitssltdefaulttdeprecated_nametrabbit_use_sslthelpsConnect over SSL.tssl_versionttkombu_ssl_versionsSSL version to use (valid only if SSL enabled). Valid values are TLSv1 and SSLv23. SSLv2, SSLv3, TLSv1_1, and TLSv1_2 may be available on some distributions.tssl_key_filetkombu_ssl_keyfiles)SSL key file (valid only if SSL enabled).t
ssl_cert_filetkombu_ssl_certfiles*SSL cert file (valid only if SSL enabled).tssl_ca_filetkombu_ssl_ca_certss=SSL certification authority file (valid only if SSL enabled).tkombu_reconnect_delaygð?tdeprecated_grouptDEFAULTsYHow long to wait before reconnecting in response to an AMQP consumer cancel notification.tkombu_compressionsEXPERIMENTAL: Possible values are: gzip, bz2. If not set compression will not be used. This option may not be available in future versions.t$kombu_missing_consumer_retry_timeouttkombu_reconnect_timeouti<sHow long to wait a missing client before abandoning to send it its replies. This value should not be longer than rpc_response_timeout.tkombu_failover_strategytchoicessround-robintshufflesºDetermines how the next RabbitMQ node is chosen in case the one we are currently connected to becomes unavailable. Takes effect only if more than one RabbitMQ node is provided in config.trabbit_hostt	localhosttdeprecated_for_removaltdeprecated_reasons#Replaced by [DEFAULT]/transport_urls8The RabbitMQ broker address where a single node is used.trabbit_porti(s5The RabbitMQ broker port where a single node is used.trabbit_hostss$rabbit_host:$rabbit_ports$RabbitMQ HA cluster host:port pairs.t
rabbit_useridtguestsThe RabbitMQ userid.trabbit_passwordsThe RabbitMQ password.tsecrettrabbit_login_methodtPLAINtAMQPLAINsRABBIT-CR-DEMOsThe RabbitMQ login method.trabbit_virtual_hostt/sThe RabbitMQ virtual host.trabbit_retry_intervalis1How frequently to retry connecting with RabbitMQ.trabbit_retry_backoffisDHow long to backoff for between retries when connecting to RabbitMQ.trabbit_interval_maxisGMaximum interval of RabbitMQ connection retries. Default is 30 seconds.trabbit_max_retriesisSMaximum number of RabbitMQ connection retries. Default is 0 (infinite retry count).trabbit_ha_queuessTry to use HA queues in RabbitMQ (x-ha-policy: all). If you change this option, you must wipe the RabbitMQ database. In RabbitMQ 3.0, queue mirroring is no longer controlled by the x-ha-policy argument when declaring a queue. If you just want to make sure that all queues (except those with auto-generated names) are mirrored across all nodes, run: "rabbitmqctl set_policy HA '^(?!amq\.).*' '{"ha-mode": "all"}' "trabbit_transient_queues_ttltminisÏPositive integer representing duration in seconds for queue TTL (x-expires). Queues which are unused for the duration of the TTL are automatically deleted. The parameter affects only reply and fanout queues.trabbit_qos_prefetch_countsXSpecifies the number of messages to prefetch. Setting to zero allows unlimited messages.theartbeat_timeout_thresholdsNumber of seconds after which the Rabbit broker is considered down if heartbeat's keep-alive fails (0 disable the heartbeat). EXPERIMENTALtheartbeat_ratesNHow often times during the heartbeat_timeout_threshold we check the heartbeat.tfake_rabbits<Deprecated, use rpc_backend=kombu+memory or rpc_backend=fakecCs:i}|rd|d<n|dkr6|d|d<n|S(sÔConstruct the arguments for declaring a queue.

    If the rabbit_ha_queues option is set, we try to declare a mirrored queue
    as described here:

      http://www.rabbitmq.com/ha.html

    Setting x-ha-policy to all means that the queue will be mirrored
    to all nodes in the cluster. In RabbitMQ 3.0, queue mirroring is
    no longer controlled by the x-ha-policy argument when declaring a
    queue. If you just want to make sure that all queues (except those
    with auto-generated names) are mirrored across all nodes, run:
      rabbitmqctl set_policy HA '^(?!amq\.).*' '{"ha-mode": "all"}'

    If the rabbit_queue_ttl option is > 0, then the queue is
    declared with the "Queue TTL" value as described here:

      https://www.rabbitmq.com/ttl.html

    Setting a queue TTL causes the queue to be automatically deleted
    if it is unused for the TTL duration.  This is a helpful safeguard
    to prevent queues with zero consumers from growing without bound.
    tallsx-ha-policyiiès	x-expires((R:trabbit_queue_ttltargs((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt_get_queue_argumentsÑs
t
RabbitMessagecBs#eZdZdZdZRS(cCs?tt|jtj|jtjd|||_dS(NsRabbitMessage.Init: message %s(	tsuperREt__init__t
rpc_commontdeserialize_msgtpayloadtLOGttracet_raw_message(tselftraw_message((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRGõscCs!tjd||jjdS(Ns%RabbitMessage.acknowledge: message %s(RKRLRMtack(RN((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytacknowledgeûscCs!tjd||jjdS(Ns!RabbitMessage.requeue: message %s(RKRLRMtrequeue(RN((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRRÿs(t__name__t
__module__RGRQRR(((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyREôs		tConsumercBsDeZdZedddZdZdZdZdZ	RS(sConsumer class.ic	Cs©||_||_||_||_||_||_||_||_|	|_t	|
||_
d|_d|_
tjjd|d|d|jd|j|_dS(sgInit the Consumer class with the exchange_name, routing_key,
        type, durable auto_delete
        tnamettypetdurabletauto_deleteN(t
queue_namet
exchange_nametrouting_keytexchange_auto_deletetqueue_auto_deleteRXtcallbackRWtnowaitRDtqueue_argumentstNonetqueuet_declared_ontkombutentitytExchangetexchange(RNR[RZR\RWRXR]R^R_R`R:RB((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRGs$												cCsÉtjjd|jd|jd|jd|jd|jd|jd|j	|_
y*tjd|j
|j|j
jWn;|jjk
r¸}|jd	kr²|j
jq¹nX|j|_d
S(s0Re-declare the queue after a rabbit (re)connect.RVtchannelRhRXRYR\Ras[%s] Queue.declare: %siN(ReRftQueueRZRiRhRXR^R\RaRcRKtdebugt
connection_idtdeclaret
connectiontchannel_errorstcodeRd(RNtconntexc((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRm!s"							cCsú|j|jkr"|j|ny2|jjd|jdtj|d|jWn|j	j
k
rõ}t|jj
|_|jdks±|jdkrï|jdkrï|j||jjd|jdtj|d|jqönXdS(sæActually declare the consumer on the amqp channel.  This will
        start the flow of messages from the queue.  Using the
        Connection.consume() will process the messages,
        calling the appropriate callback.
        R_tconsumer_tagR`iis	Basic.ackN(RiRdRmRctconsumet	_callbacktsixt	text_typeR`RnRotsett
_consumerstvaluest	_new_tagsRptmethod_name(RNRqttagRr((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRt=s


cCs-tjd||jjtj|dS(Ns!ConsumerBase.cancel: canceling %s(RKRLRctcancelRvRw(RNR}((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyR~bscCsyt|jjdd}|r-||}ny|jt|Wn.tk
rttjt	d|j
nXdS(sbCall callback with deserialized message.

        Messages that are processed and ack'ed.
        tmessage_to_pythons*Failed to process message ... skipping it.N(tgetattrRcRiRbR_REt	ExceptionRKt	exceptionRtreject(RNtmessagetm2p((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRufs
N(
RSRTt__doc__tFalseRbRGRmRtR~Ru(((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRUs		%	tDummyConnectionLockcBseZdZRS(cCsdS(N((RN((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytheartbeat_acquirexs(RSRTR(((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRwstConnectionLockcBsDeZdZdZdZdZdZejdZ	RS(sÑLock object to protect access to the kombu connection

    This is a lock object to protect access to the kombu connection
    object between the heartbeat thread and the driver thread.

    They are two way to acquire this lock:
        * lock.acquire()
        * lock.heartbeat_acquire()

    In both case lock.release(), release the lock.

    The goal is that the heartbeat thread always have the priority
    for acquiring the lock. This ensures we have no heartbeat
    starvation when the driver sends a lot of messages.

    So when lock.heartbeat_acquire() is called next time the lock
    is released(), the caller unconditionally acquires
    the lock, even someone else have asked for the lock before it.
    cCsgd|_t|_d|_tj|_tj|j|_	tj|j|_
tj|_
dS(Ni(t_workers_waitingRt_heartbeat_waitingRbt_lock_acquiredt	threadingtLockt_monitort	Conditiont_workers_lockst_heartbeat_lockRtfetch_current_thread_functort_get_thread_id(RN((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRGs			cCs^|jOx8|jrD|jd7_|jj|jd8_q
W|j|_WdQXdS(Ni(RRRRtwaitR(RN((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytacquires

cCsX|jIx2|jdk	r>t|_|jjt|_q
W|j|_WdQXdS(N(	RRRbtTrueRRRRR(RN((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyR¢s
	

cCsª|j|jdkr(tdn|j}|j|kr_td|j|fnd|_|jr|jjn|jdkr |j	jnWdQXdS(Ns$We can't release a not acquired locksFWe can't release lock acquired by another thread/greenthread; %s vs %si(
RRRbtRuntimeErrorRRRtnotifyRR(RNt	thread_id((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytrelease«s
		ccs%|jz	dVWd|jXdS(N(RR(RN((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt
for_heartbeatºs
	(
RSRTRRGRRRt
contextlibtcontextmanagerR(((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyR|s						t
ConnectioncBseZdZiZdZiejd6ejd6Zidd6dd6dd	6d
d6Z	x=e	D]5Z
yeee	e
ee
<Wq^ek
rq^Xq^We
dZd
ZdZedZedZdZd-d-d-edZdZdZdZdZdZd-dZejdZ dZ!dZ"dZ#dZ$dZ%d-dZ&d Z'd!Z(d-d-d"Z)d#Z*d-d-d-d$Z+d%Z,d-d-d&Z-d-d-d'Z.d-d-d(Z/d)Z0d-d-d*Z1d-d+Z2d-d,Z3RS(.sConnection object.cCsU|j}|j|_|j|_|j|_|j|_|j	|_
|j|_|j|_
|j|_|j|_|j|_|j|_|j|_|j|_|j|_|j|_|j|_|j|_|j|_|j|_|j|_|j|_|j|_|j|_|jrY|j|_|j|_|j |_ |j!|_!n|jdkrtd|_n|j
dk	r|j
}n	|j
}d|_#|jrÍt$j%t&dd||_#n|j'ræ|j(j)drt$j%t&d|j(nt*|j'dkr,t+j,|j'nx©|j'D]©}|j(j-dd}|j-dd	}|j#d
|j#r|dnd|t.j/|j0pdt.j/|j1p©d|j2|j3pÁdt4|j5pÐd|f7_#q6Wnò|j(j)dr#|j(j-dd}d
||f|_#nµt*|jdkrKt+j,|jnx|jD]}t6j7|d|j\}	}
|j#d|j#rdndt.j/|jdt.j/|jd|j2|	|
|f7_#qUWt8j9|_:i|_;d|_<t=|_>i|_?t@jAd|_Bt=|_Ct=|_DtE|_Fd|_G||_H|tIjJkrqtK|_LntM|_Lt4tNjO|_Pdt8jQjRtSjTdt8j9|jPf|_UtVjWjX|j#d|jYd|j
d|jd|jditZd6iitZd6tZd6tZd6d6|jUd6d6|j[d6|j\d6|_Wt$j]d|j^t_|jt_|jd |_`tE|_a|jL|jbWdQXd|_c|tIjJkrÒ|jdnt$j]d!|j^|jer|j`|_fn	d|_f|j#j)d"rQd#|jWj(_gd$|jW_3d%|jW_5d&|_fndS('NiRsmDeprecated: fake_rabbit option is deprecated, set rpc_backend to kombu+memory or use the fake driver instead.smemory://%s/skombu+suSelecting the kombu transport through the transport url (%s) is a experimental feature and this is not yet supported.itrabbitRs%s%s://%s:%s@%s:%s/%st;i(s%s://%stdefault_ports%samqp://%s:%s@%s:%s/%ss%s:%d:%sRtlogin_methodt	heartbeattfailover_strategyttransport_optionstconfirm_publishtauthentication_failure_closesconnection.blockedtconsumer_cancel_notifytcapabilitiestconnection_nametclient_propertiest
on_blockedton_unblockedsF[%(connection_id)s] Connecting to AMQP server on %(hostname)s:%(port)sg@s{[%(connection_id)s] Connected to AMQP server on %(hostname)s:%(port)s via [%(transport)s] client with port %(client_port)s.s	memory://gt
memory_driveriÒg©?(htoslo_messaging_rabbitR9tmax_retriesR6tinterval_startR7tinterval_steppingR8tinterval_maxR1R¤R@R4tvirtual_hostR,R+R-R/R:R;R=R>R?Rtamqp_durable_queuestamqp_auto_deleteRR"R$R!RRRRRbt_urlRKtwarningR
thostst	transportt
startswithtlentrandomR&treplaceRtquotetusernametpasswordt_parse_url_hostnamethostnametstrtportRtparse_host_porttostgetpidt_initial_pidRyt	_producerRxR{t_active_tagst	itertoolstcountt_tagst_declared_exchangest_declared_queuesRt_consume_loop_stoppedRitpurposeRHtPURPOSE_SENDRt_connection_lockRtuuidtuuid4RltpathtbasenametsystargvRVReRnR t_fetch_ssl_paramsRt_on_connection_blockedt_on_connection_unblockedRkt_get_connection_infotfloatt_heartbeat_wait_timeoutt_heartbeat_support_log_emittedtensure_connectiont_heartbeat_threadt_heartbeat_startt _heartbeat_supported_and_enabledt
_poll_timeouttpolling_interval(RNtconfturlRÔtdriver_confR¶thostR¼tadrRÅRÇ((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRGÈsò						


											
	
#	
	
	
	ttlsv1tsslv23tPROTOCOL_SSLv2tsslv2tPROTOCOL_SSLv3tsslv3tPROTOCOL_TLSv1_1ttlsv1_1tPROTOCOL_TLSv1_2ttlsv1_2cCsI|j}y|j|SWn'tk
rDttd|nXdS(NsInvalid SSL version : %s(tlowert_SSL_PROTOCOLStKeyErrorRR
(tclstversiontkey((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytvalidate_ssl_versions

cCsd|krd|S|S(sHandles hostname returned from urlparse and checks whether it's
        ipaddress. If it's ipaddress it ensures that it has brackets for IPv6.
        t:s[%s]((RNRÅ((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRÄ scCs|jrt}|jr4|j|j|d<n|jrM|j|d<n|jrf|j|d<n|jr|j|d<tj|d<n|ptSt	S(s]Handles fetching what ssl params should be used for the connection
        (if any).
        Rtkeyfiletcertfiletca_certst	cert_reqs(
RtdictRRÿRRRt
CERT_REQUIREDRR(RNt
ssl_params((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRÝ¦s						

cCstjtd|dS(Ns)The broker has blocked the connection: %s(RKterrorR(treason((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRÞ½scCstjtddS(Ns'The broker has unblocked the connection(RKtinfoR(((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRßÁscCs1|jd|jd|jj|jdS(Ntmethod(t_set_current_channelRbtensureRntconnecttset_transport_socket_timeout(RN((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRäÅs
csùtj}j|kr:tjtd|_n|dkrRj}n|dksj|dkrsd}nfd}fd}fd}	yrjj	|	dj
d|d|d	jpÙd
djdj
d
|}
|
\}}j||SWnÚtjjk
r¹}
tjddt|oR||
jdi|
d6|d6}|jjjtd|}tj|tj|n<tjk
rÏn&tk
rô}
|oí||
nXdS(sWill retry up to retry number of times.
        retry = None means use the value of rabbit_max_retries
        retry = -1 means to retry forever
        retry = 0 means no retry
        retry = N means N retries

        NOTE(sileht): Must be called within the connection lock
        sProcess forked after connection established! This can result in unpredictable behavior. See: http://docs.openstack.org/developer/oslo.messaging/transport.htmlicsðtjdjdto)|jdkrFj|n|}i|d6|d6}|jjdtj|kr¡tj	t
d|ntj	t
d|jdkrìtjd	jtj
jndS(
Ns+[%s] Received recoverable error from kombu:texc_infoiterr_strt
sleep_times
Socket closedsq[%(connection_id)s] AMQP server %(hostname)s:%(port)s closed the connection. Check login credentials: %(err_str)ss[%(connection_id)s] AMQP server on %(hostname)s:%(port)s is unreachable: %(err_str)s. Trying again in %(sleep_time)d seconds. Client port: %(client_port)ss(Delaying reconnect for %1.1f seconds ...(RKRkRlRRtupdateRàRvRwRRRLttimetsleep(RrtintervalR
(trecoverable_error_callbackRN(sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyton_errorås"	

	
cs7j|jtjtdjdS(sCallback invoked when the kombu reconnects and creates
            a new channel, we use it the reconfigure our consumers.
            s}[%(connection_id)s] Reconnected to AMQP server on %(hostname)s:%(port)s via [%(transport)s] client with port %(client_port)s.N(RRRKR
RRà(tnew_channel(RN(sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyton_reconnections
csj|dS(N(R(Ri(RRN(sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytexecute_methods
RiR²terrbackR³it
interval_stepRµt	on_revives&Received recoverable error from kombu:RRtretrys\Unable to connect to AMQP server on %(hostname)s:%(port)s after %(retry)s tries: %(err_str)sN(RÉRÊRËRKRºR
RbR²Rnt	autoretryRiR³R´RµRReRtOperationalErrorRkRRR
R
RtMessageDeliveryFailuretrpc_amqptAMQPDestinationNotFoundR(RNRRRterror_callbackttimeout_is_errortcurrent_pidRRRtautoretry_methodtretRiRrR
tmsg((RRRNsX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyR
ÌsN	)				

	
cCsÆ||jkrdS|jdk	rR|jj|jj|jj|jn||_|dk	rÂ|jtj	kr|j
|ntjj
||_x!|jD]}|j|q¨WndS(sdChange the channel to use.

        NOTE(sileht): Must be called within the connection lock
        N(RiRbRÒtclearRÑRntmaybe_close_channelRÔRHtPURPOSE_LISTENt_set_qosRet	messagingtProducerRÌRyRm(RNRtconsumer((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyR@s

	cCs,|jdkr(|jd|jtndS(s%Set QoS prefetch count on the channeliN(R=t	basic_qosR(RNRi((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyR.Vs	cCs|j|jrxS|jjD]B\}}|jdkr#tjd|jj|jj	q#q#W|j
d|jjd|_ndS(sClose/release this connection.tfanouts-[connection close] Deleting fanout queue: %s N(
t_heartbeat_stopRnRytitemsRWRKRkRcRVtdeleteRRbR(RNR1R}((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytclose]s
		

c	Cs¥|jy4x-|jjD]\}}|jd|qWWn!tjjk
ra|jnX|jj|j	j|j
jtjd|_
WdQXdS(s+Reset a connection so it can be used again.R}iN(RÖRyR5R~ReRR!RäR+RÍR{RÎRÏRÐ(RNR1R}((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytresetjs



cCsO|jdkrtS|jjr#tS|jsKtjtdt|_ntS(NisUHeartbeat support requested but it is not supported by the kombu driver or the broker(	R>RRntsupports_heartbeatsRRãRKRºR
(RN((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRçws	cCs4|j}|jr<|dkr*|}q<t||}ny|jjj}Wn3tk
r}tj	d|j
t|fn¬X|j|t
jdkr0t
jdkr0yE|dk	rÉ|dnd}|jtjtttj|Wq0tjk
r,}|d}|tjkr-q-q0XndS(Ns'[%s] Failed to get socket attribute: %stwin32tdarwinièi(R>RçRbR<RiRntsocktAttributeErrorRKRkRlRÆt
settimeoutRÛtplatformt
setsockopttsockettIPPROTO_TCPtTCP_USER_TIMEOUTtinttmathtceilRterrnotENOPROTOOPT(RNttimeouttheartbeat_timeoutR<teRRp((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRs*
			

ccs |j|dV|jdS(N(R(RNRI((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt_transport_socket_timeout±s
cCs|jjd|jdS(Ntrate(Rntheartbeat_checkR?(RN((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt_heartbeat_check·scCs\|jrOtj|_tjd|j|_t|j_	|jj
n	d|_dS(Nttarget(RçRtEventt_heartbeat_exit_eventRtThreadt_heartbeat_thread_jobRåRtdaemontstartRb(RN((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRæ¼s	cCs9|jdk	r5|jj|jjd|_ndS(N(RåRbRRRxtjoin(RN((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyR4Æs

c	Cs	xõ|jjs÷|jj»y|y<|jy|jjddWntjk
r_nXWn9t	j
jk
r}tj
td||jnXWn7tk
r×tjtdtjddtnXWdQX|jjd|jqW|jjdS(s3Thread that maintains inactive connections
        RIgü©ñÒMbP?sHA recoverable connection/channel error occurred, trying to reconnect: %ssAUnexpected error during heartbeart thread processing, retrying...RRN(RRtis_setRÖRRORntdrain_eventsRARIReRR!RKR
RRäRRºR
RkRRRâR+(RNRr((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRTÌs&

csHfd}fd}jj|d|SWdQXdS(siCreate a Consumer using the class that was passed in and
        add it to our list of consumers
        cs1ijd6|d6}tjtd|dS(NttopicRs=Failed to declare consumer for topic '%(topic)s': %(err_str)s(R\RKRR(Rrtlog_info(R1(sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt_connect_errorñscsqjjjj}|dkr`tj}|jj<jj|n|j	<S(N(
RmRÍtgetRZRbtnextRÐR{taddRy(R}(R1RN(sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt_declare_consumerös

R%N(RÖR
(RNR1R\R`((R1RNsX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytdeclare_consumerìs

c
stjdjdfdfd}fd}jj|dd|WdQXdS(	s"Consume from all queues/consumers.tdurationcSs tjd|tjdS(Ns&Timed out waiting for RPC response: %s(RKRkRHtTimeout(Rr((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt_raise_timeoutscsAt|tjs-tjj_nj|dS(N(t
isinstanceRHRcRxRyRzR{tcheck_return(Rr(RdRNttimer(sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt_recoverable_error_callbackscs$|tjtd|dS(Ns(Failed to consume message from queue: %s(RKRR(Rr(Rh(sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt_error_callbacks
cs}jjsjjdnxbjrxRjjD]A\}}|jkr;|jd|jj|q;q;Wq"Wdkrj	nt
j	}xËtrxjrÁdSj
rÚjnyjjd|dSWq®tjk
r(}j|dj	}q®jjk
rt}|jdkrn|jdkrnjjdnq®Xq®WdS(NiR}RItmaximumis	Basic.ack(Rnt	connectedtrecoverable_connection_errorsR{RyR5RttremoveRbRèR<RRÓRçRORYRARIRfRoRpR|trecoverable_channel_errors(R1R}tpoll_timeoutRr(RdRNRIRg(sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt_consumes0		
RR%N(RHt
DecayingTimerRVRÖR
(RNRIRiRp((RdRhRNRIRgsX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRts
	$
cCs
t|_dS(N(RRÓ(RN((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytstop_consumingBscCs\td|d|d|dddtdtdtd	|d
|jd|j
}|j|dS(
sCreate a 'direct' queue.
        In nova's use, this is generally a msg_id queue used for
        responses for call/multicall
        R[RZR\RWtdirectRXR]R^R_R:RBN(RURRR:R;Ra(RNRZR_R1((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytdeclare_direct_consumerEs	cCsbtd|d|p|d|ddd|jd|jd|jd	|d
|j	}|j|dS(sCreate a 'topic' consumer.R[RZR\RWRZRXR]R^R_R:N(RUR·R¸R:Ra(RNR[RZR_RZR1((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytdeclare_topic_consumerXs			cCstjj}d|}d||f}td|d|d|dddtd	td
td|d|jd
|j
}|j|dS(sCreate a 'fanout' consumer.s	%s_fanouts%s_fanout_%sR[RZR\RWR3RXR]R^R_R:RBN(	R×RØthexRURRR:R;Ra(RNRZR_tuniqueR[RZR1((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytdeclare_fanout_consumergs
	c
sWfd}tj||||}|j|j|d|d|WdQXdS(s1Send to a publisher based on the publisher class.csDijd6|d6}tjtd|tjdd|dS(NRZRs;Failed to publish message to topic '%(topic)s': %(err_str)sRR(RVRKRRRk(RrR[(Rh(sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRisRR%N(t	functoolstpartialRÖR
(RNRRhR*R\RIRRi((RhsX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt_ensure_publishing{s
cCs|jj}d}|jr^t|jjdr^|jjjr^|jjjjd}n|ji|d6|jd6|S(NR<itclient_portRl(	RnR
RbRithasattrR<tgetsocknameRRl(RNR
R|((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRàscCsº|jp|j|jksD||jj|jj|jni|d6|pWdd6|d6}tjd||j|0|j	j
|d|d|d|d	|jWd
QXd
S(sPublish a message.R*RtwhoRþsPConnection._publish: sending message %(msg)s to %(who)s with routing key %(key)sRhR\t
expirationtcompressionN(tpassiveRVRÑRiRmR_RKRLRLRÌtpublishR!(RNRhR*R\RIR[((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt_publishs

	cCsÍ|j|f}||jkrtjjd|jd|d|jd|jd|d|dt|j	d}i|d	6|d6}t
jd
||j|jj
|n|j||d|d|dS(
såPublisher that declares a default queue

        When the exchange is missing instead of silently creates an exchange
        not binded to a queue, this publisher creates a default queue
        named with the routing_key

        This is mainly used to not miss notification in case of nobody consumes
        them yet. If the future consumer bind the default queue it can retrieve
        missing messages.

        _set_current_channel is responsible to cleanup the cache.
        RiRhRXRYRVR\RaiRþs]Connection._publish_and_creates_default_queue: declare queue %(key)s on %(exchange)s exchangeRIN(RVRÒReRfRjRiRXRYRDR:RKRLRmR_R(RNRhR*R\RItqueue_indentifierRcR[((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt"_publish_and_creates_default_queueªs"			
cCs|jstdny$|j||d|d|dSWnD|jjk
r}|jdkr|tjd|jnnXdS(s7Publisher that raises exception if exchange is missing.sQ_publish_and_retry_on_missing_exchange() must be called with an passive exchange.R\RINisexchange %s doesn't exists(	RRRRnRoRpR#R$RV(RNRhR*R\RIRr((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt'_publish_and_raises_on_missing_exchangeÎs		cCsMtjjd|dddtdtdt}|j|j||d|dS(	sSend a 'direct' message.RVRWRsRXRYRR\N(ReRfRgRRR{R(RNtmsg_idR*Rh((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytdirect_sendès	c
CsYtjjd|ddd|jd|j}|j|j||d|d|d|d	S(
sSend a 'topic' message.RVRWRZRXRYR\RIRN(ReRfRgR·R¸R{R(RNR[RZR*RIRRh((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt
topic_sendós	c	CsKtjjdd|dddtdt}|j|j||d|dS(	sSend a 'fanout' message.RVs	%s_fanoutRWR3RXRYRN(ReRfRgRRR{R(RNRZR*RRh((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytfanout_sendÿs
	c	KsStjjd|ddd|jd|j}|j|j||d|d|dS(	s!Send a notify message on a topic.RVRWRZRXRYR\RN(ReRfRgR·R¸R{R(RNR[RZR*RtkwargsRh((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytnotify_sends	N(4RSRTRtpoolsRGRtPROTOCOL_TLSv1tPROTOCOL_SSLv23Rút_OPTIONAL_PROTOCOLStprotocolRR=tclassmethodRÿRÄRÝtstaticmethodRÞRßRäRbRR
RR.R7R8RçRRRRLRORæR4RTRaRtRrRtRuRxR{RàRRRRRRR(((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyR Ãsn	¾




			r			
	
	.		
		 	=				
$		tRabbitDrivercBs)eZdZdddZedZRS(sÞRabbitMQ Driver

    The ``rabbit`` driver is the default driver used in OpenStack's
    integration tests.

    The driver is aliased as ``kombu`` to support upgrading existing
    installations with older settings.

    c
Cstjdddd}|j||jtd||jtjd||jtjd|t	j
|||j}|jj
|_|jj|_|jj}|jj}|jj}tj|||||t}	tt|j|||	||dS(NRVR±ttitlesRabbitMQ driver optionstgroup(RtOptGrouptregister_groupt
register_optstrabbit_optsR#t	amqp_optsRt	base_optsRHtConfigOptsProxyRVR±R"t!missing_destination_retry_timeoutR=t
prefetch_sizetrpc_conn_pool_sizetconn_pool_min_sizet
conn_pool_ttlR	tConnectionPoolR RFRRG(
RNRêRëtdefault_exchangetallowed_remote_exmodst	opt_grouptmax_sizetmin_sizetttltconnection_pool((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRGs(	
cCsdS(N((RNRR((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pytrequire_features@sN(RSRTRRbRGRR¬(((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyRs	(DRRGRyRÎRERÉR¿RARRÛRRR×Retkombu.connectiontkombu.entitytkombu.messagingtoslo_configRtoslo_logRtloggingt
oslo_utilsRRRvtsix.moves.urllibRtoslo_messaging._driversRR#RRRRHR	toslo_messaging._i18nR
RRR
toslo_messagingRRRCtBoolOptRtStrOpttFloatOpttIntOptRtPortOpttListOptRt	getLoggerRSRKRDRREtobjectRUt	DummyLockRRR tAMQPDriverBaseR(((sX/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/impl_rabbit.pyt<module>s:			#sGÿÿÿT
trilio-internal / idna python

Products

About

Resources

Contact Gemfury
