ó
čEYc@s£
dg
Zd
dl
Z
d
dlZd
dlZd
dlZd
dlZd
dlmZ
d
dlm	Z	
d
dl
Z
d
dlmZ

d
dlmZ
d
dlmZ
d
dlmZ
d
d	lmZ
d
d
lmZ
d
dlmZ
e
jeƒ
ZdZd
Zdef
d„ƒYZdejf
d„ƒYZdef
d„ƒYZdej f
d„ƒYZ!def
d„ƒYZ"def
d„ƒYZ#dej$f
d„ƒYZ%dS(tAMQPDriverBaseiÿÿÿÿN(
t	timeutils(
tmoves(
tamqp(
tbase(
tcommon(
t
_(
t_LE(
t_LI(
t_LWgü©ñÒMbP?gð?tMessageOperationsHandlerc
BsDeZ
dZd
„Zd„Zd„Zd„Zd„Zd„ZRS(sªQueue used by message operations to ensure that all tasks are
    serialized and run in the same thread, since underlying drivers like kombu
    are not thread safe.
    cCsqd
|
tt
|ƒ
ƒ
f|_tjjƒ|_tjƒ|_	tj
d|jƒ
|_t
|j_d|_dS(Ns%s (%s)ttarget(thextidtnameRtqueuetQueuet_taskst	threadingtEventt	_shutdowntThreadt_process_in_backgroundt_shutdown_threadtTruetdaemontNonet	_executor(tselfR((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyt__init__2s


	

c


Cs|jj
ƒ
dS(
N(Rtset(
R((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pytstopBs
c


Cs|jj
ƒ
d
S(s_Run all pending tasks queued by do() in an thread during the
        shutdown process.
        N(Rtstart(
R((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pytprocess_in_backgroundEsc

Cs1x*|jj
ƒs,|jƒ
tjtƒ

qWdS(
N(Rtis_settprocessttimetsleeptACK_REQUEUE_EVERY_SECONDS_MIN(
R((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRKs



c
Csdx]tr_y|j
jd
tƒ
\}
}Wntjjk
rB


Pn
Xz|
ƒ
Wd|jƒ
XqWdS(s%Run all pending tasks queued by do().tblockN(RRtgettFalseRRtEmptyR(Rttasktevent((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyR#Ps	





cCsg|jdkrtd
ƒ
‚
nE|jdkr7|
ƒ
n,tjƒ}|jj|
|fƒ

|jƒ
dS(s@Put the task in the queue and waits until the task is completed.s(Unexpected error, no executor is setupedtblockingN(RRtRuntimeErrorRRRtputtwait(RR+R,((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pytdo]s




(	t__name__t
__module__t__doc__RRR!RR#R1(((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyR
-s
					
tAMQPIncomingMessagecBsAeZ
d„Zddd
„Zddd„Zd„Zd„ZRS(c		Csott
|ƒj||ƒ
|
|_||_||_||_||_||_t	j
ƒ|_|jjƒ
dS(
N(
tsuperR5Rtlistenert	unique_idtmsg_idtreply_qt_obsolete_reply_queuest_message_operations_handlerR
t	StopWatcht	stopwatchR (	RR7tctxttmessageR8R9R:tobsolete_reply_queuestmessage_operations_handler((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRms
		
	
	
	
	

cCsÑ|jj
|j|jƒsdS|r7tj|ƒ
}ni|d
6|d6td6|jd6}tj|ƒ

|tj	}t
jdi|jd6|d6|jd6|jj
ƒd	6ƒ
|
j|jtj|ƒ
ƒ
dS(
Ntresulttfailuretendingt_msg_idsTsending reply msg_id: %(msg_id)s reply queue: %(reply_q)s time elapsed: %(elapsed)ssR9R8R:telapsed(R;t
reply_q_validR:R9t
rpc_commontserialize_remote_exceptionRtrpc_amqpt_add_unique_idt	UNIQUE_IDtLOGtdebugR>RGtdirect_sendt
serialize_msg(RtconntreplyRDtmsgR8((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyt_send_replyzs 









	






cCsO
|js
dS|j
j|j|jƒs,dS|jjj}tjd
|ƒ
}|j	ƒ
xñt
rJ
y<|jjjtjƒ
}|j
||
|ƒ
WdQXdSWqZtjk
rF



|jƒdkrõtjdi|jd6|jd6ƒ
tjdƒ

qG
|j
j|j|jƒ
tjtdƒ
i|jd6|jd6|d
6ƒ
dSqZXqZWdS(NtdurationisUThe reply %(msg_id)s cannot be sent  %(reply_q)s reply queue don't exist, retrying...R9R:gÐ?smThe reply %(msg_id)s cannot be sent  %(reply_q)s reply queue don't exist after %(duration)s sec abandoning...(R9R;RHR:R7tdrivert!missing_destination_retry_timeoutRIt
DecayingTimerR Rt_get_connectiontPURPOSE_SENDRURKtAMQPDestinationNotFoundtcheck_returnRNROR$R%taddtinfoR(RRSRDRVttimerRR((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRS‘s8
	




	







	










c

Cs0|jj
|jjƒ

|jjj|jƒ

dS(
N(R<R1R@tacknowledgeR7tmsg_id_cacheR^R8(
R((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRa»s

c

Cs|jj
|jjƒ

dS(
N(R<R1R@trequeue(
R((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRc¿sN(R2R3RRRURSRaRc(((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyR5ks
	
*	tObsoleteReplyQueuesCachec
Bs>eZ
dZd
ZdZd„Zd„Zd„Zd„ZRS(sLCache of reply queue id that doesn't exists anymore.

    NOTE(sileht): In case of a broker restart/failover
    a reply queue can be unreachable for short period
    the IncomingMessage.send_reply will block for 60 seconds
    in this case or until rabbit recovers.

    But in case of the reply queue is unreachable because the
    rpc client is really gone, we can have a ton of reply to send
    waiting 60 seconds.
    This leads to a starvation of connection of the pool
    The rpc server take to much time to send reply, other rpc client will
    raise TimeoutError because their don't receive their replies in time.

    This object cache stores already known gone client to not wait 60 seconds
    and hold a connection of the pool.
    Keeping 200 last gone rpc client for 1 minute is enough
    and doesn't hold to much memory.
    iÈi<c

Cs.tj
ƒ|_tj|j|jƒ|_dS(
N(RtRLockt_lockt
cachetoolstTTLCachetSIZEtTTLt_cache(
R((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRás

cCs'|
|jkr#|j
|
|ƒ
tStS(
N(Rkt
_no_reply_logR)R(RR:R9((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRHås



cCs;|j
|j
ji
||
6ƒ

WdQX|j|
|ƒ
dS(
N(RfRktupdateRl(RR:R9((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyR^ës



cCs(tj
td
ƒ
i|
d6|d6ƒ
dS(Ns8%(reply_queue)s doesn't exists, drop reply to %(msg_id)streply_queueR9(RNtwarningR	(RR:R9((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRlðs



(	R2R3R4RiRjRRHR^Rl(((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRdÉs
			tAMQPListenercBsAeZ
d„Zd
„Zejdd„
ƒ
Zd„Zd„Z	RS(cCs‰tt
|ƒj|
jƒ

|
|_||_tjƒ|_g|_	t
jƒ|_t
jƒ|_
tƒ|_td
ƒ
|_t|_dS(NRp(R6RpRt
prefetch_sizeRWRRRKt_MsgIdCacheRbtincomingRRRt_shutoffRdR;R
R<R&t_current_timeout(RRWRR((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRøs

	
	

	





c
Cs¢tj
|
ƒ
}|jj|
ƒ
}|jrQtjd
i|jd6|jd6ƒ
ntjd|ƒ
|jj	t
||jƒ|
||j|j|j|j
ƒƒ

dS(Ns6received message msg_id: %(msg_id)s reply to %(queue)sRR9s#received message with unique_id: %s(RKtunpack_contextRbtcheck_duplicate_messageR9RNROR:RstappendR5tto_dictR;R<(RR@R?R8((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyt__call__
s 


	
	




	





cCs!
tj
d
|
ƒ
jƒ}xÏ|jjƒsé|jjƒ
|jrP|jjdƒ
S|j	dt
ƒ
}|dkrz|j}n|dk
rŠdSy#|j
jdt|j|ƒƒ

Wn-tjk
rÜ


t|jdtƒ|_qXt|_qW|jjƒ
|jr
|jjdƒ
S|jjƒ
dS(NRVitreturn_nonettimeouti(R
R=R RR"R<R#RstpoptleftoverRRRuRRtconsumetminRItTimeouttmaxtACK_REQUEUE_EVERY_SECONDS_MAXR&RtR(RR|R>tleft((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pytpoll
s(

	





#






	
c


Cs8|jj
ƒ
|jjƒ
|jjƒ
|jjƒ
dS(
N(RRRRtstop_consumingRtR0R<R!(
R((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyR9
s





c


Cs|jj
ƒ
|jjƒ
dS(
N(R<RRRtclose(
R((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pytcleanupD
s
N(
R2R3RRzRtbatch_poll_helperRR…RRˆ(((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRpös	
	
	tReplyWaitersc
Bs>eZ
eƒZd„Zd
„Zd„Zd„Zd„ZRS(c

Csi|_d
|_
dS(Ni
(t_queuest_wrn_threshold(
R((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRQ
s
	
cCsRy!|j|
j
d
td|ƒSWn*tjjk
rM


tjd|
ƒ
‚
n
XdS(NR'R|s.Timed out waiting for a reply to message ID %s(R‹R(RRRR*toslo_messagingtMessagingTimeout(RR9R|((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyR(U
s

!



cCsi|jj
|
ƒ
}|sXtjtd
ƒ
|
ƒ
tjdit|jƒ
d6|d6ƒ
n
|j|ƒ

dS(Ns*No calling threads waiting for msg_id : %ss) queues: %(queues)s, message: %(message)stqueuesR@(R‹R(RNR_RROtlenR/(RR9tmessage_dataR((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyR/]
s



	
!cCstj
jƒ|j|
<t|jƒ
}||jkr{tjtd
ƒ
i|d6|jd6|jdd6ƒ
|jd9_ndS(NsœNumber of call queues is %(queues_length)s, greater than warning threshold: %(old_threshold)s. There could be a leak. Increasing threshold to: %(threshold)st
queues_lengtht
old_thresholdit	threshold(	RRRR‹RRŒRNRoR	(RR9R’((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyR^f
s








cCs|j|
=dS(
N(
R‹(RR9((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pytremoves
s
(	R2R3tobjecttWAKE_UPRR(R/R^R•(((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRŠM
s						
tReplyWaitercBs_eZ
d„Zd
„Zd„Zd„Zd„Zd„Zed„ƒ
Z	d„Z
d„ZRS(	cCs„||_||_
tjƒ|_tƒ|_|jj|
|ƒ
tj	ƒ|_
tjd
|jƒ
|_
t|j
_|j
jƒ
dS(NR(RRtallowed_remote_exmodsRKRrRbRŠtwaiterstdeclare_direct_consumerRRt_thread_exit_eventRR…t_threadRRR (RR:RRR™((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRx
s
	
	




c

Cs@|jr<|j
jƒ
|jjƒ
|jjƒ
d|_ndS(
N(RRœRRRR†tjoinR(
R((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyR…
s

	






c
CsŠt}
x}|j
jƒs…y|jjd
|
ƒ

WnJtjk
rX


t|
dtƒ}
q	t	k
r{


t
jtdƒ
ƒ

q	Xt}
q	WdS(NR|is/Failed to process incoming message, retrying...(
R&RœR"RRRRIRR‚Rƒt	ExceptionRNt	exceptionR(Rtcurrent_timeout((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyR…Œ
s









cCsU|
jƒ
|
j
d
dƒ}|
jdƒ
r>tjd|ƒ
n|jj||
ƒ
dS(NRFREsreceived reply msg_id: %s(RaR}RR(RNRORšR/(RR@tincoming_msg_id((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRz›
s





cCs|jj
|
ƒ

dS(
N(RšR^(RR9((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pytlisten¦
s
cCs|jj
|
ƒ

dS(
N(RšR•(RR9((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pytunlisten©
s
c

Cstj
td
ƒ
|ƒ
‚
dS(Ns/Timed out waiting for a reply to message ID %s.(RRŽR(
R9((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyt_raise_timeout_exception¬
s
cCsj|jj
|
ƒ

|
d
r<|
d
}tj||jƒ}n|
jddƒ}|
jdtƒ}||fS(NRDRCRE(RbRwRItdeserialize_remote_exceptionR™R(RR)(RtdataRDRCRE((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyt_process_reply±
s







cCs»tj
d
|ƒ
}|jƒ
d}t}xŒ|s¶|j|j|
ƒ}y|jj|
d|ƒ

}Wn$t	j
jk
rˆ


|j|
ƒ

n
X|j|ƒ
\}}|dk	r+|}q+q+W|S(NRVR|(
RIRYR RR)R]R¥RšR(RRR*R¨(RR9R|R`tfinal_replyRER@RS((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyR0½
s




	







(R2R3RRR…RzR£R¤tstaticmethodR¥R¨R0(((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyR˜w
s
	
						cBseZ
dZddd
„Zd„Zejd„
Zd„Z	dde
edd„Zdddd„Z
dd„
Zd„Zd	„Zd
„ZRS(icCs_tt
|ƒj|
|||ƒ
||_||_tjƒ|_d|_	d|_
d|_dS(
N(R6RRt_default_exchanget_connection_poolRtLockt
_reply_q_lockRt_reply_qt
_reply_q_connt_waiter(Rtconfturltconnection_pooltdefault_exchangeR™((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRÛ
s
		
	
	
c
Cs|
jp|j
S(
N(texchangeR«(RR((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyt
_get_exchangeé
s
cCstj
|jd
|
ƒ

S(Ntpurpose(RItConnectionContextR¬(RR¸((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRZì
s

c
Cs||jj
|j
dk	r |j
Sd
tjƒj}
|jtjƒ
}t	|
||j
ƒ|_|
|_
||_WdQX|j
S(Ntreply_(
R®R¯Rtuuidtuuid4RRZRItPURPOSE_LISTENR˜t_allowed_remote_exmodsR±R°(RR:RR((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyt_get_reply_qð
s



	
	
c	Csg|}	|rLtj
ƒj}
|	ji
|
d
6ƒ

|	ji
|jƒd6ƒ

ntj|	ƒ

|	tj}tj|	|ƒ
|rŽt	j
|	ƒ
}	n|r±|jj|
ƒ

d|
}n
d|}z‹
|j
t	jƒ
<
}
|r3
|j|
ƒ
}|di|d6|
jd67}tj|ƒ

|
j||
j|	d|ƒ

nÕ|
jr}
|d	i
|
jd67}tj|ƒ

|
j|
j|	d|ƒ

n‹|
j}|j|
ƒ
}|
jr·
d
|
j|
jf}n|di|d6|d67}tj|ƒ

|
jd|d|d
|	d|d|ƒ
WdQX|rE|jj|
|ƒ}t|tƒrA|‚
n|SWd|rb|jj|
ƒ

nXdS(NRFR¯sCALL msg_id: %s sCAST unique_id: %s s0NOTIFY exchange '%(exchange)s' topic '%(topic)s'R¶ttopictretrysFANOUT topic '%(topic)s's%s.%ss)exchange '%(exchange)s' topic '%(topic)s't
exchange_nameRTR|(R»R¼RRmR¿RKRLRMtpack_contextRIRQR±R£RZR[R·RÀRNROtnotify_sendtfanouttfanout_sendtservert
topic_sendR0t
isinstanceRŸR¤(RRR?R@twait_for_replyR|tenvelopetnotifyRÁRTR9R8tlog_msgRRR¶RÀRC((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyt_send
s^





















	
	



	

	











	

cCs|j|
||||d
|ƒ
S(NRÁ(
RÎ(RRR?R@RÊR|RÁ((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pytsend?s
c
Cs+|j|
||d
|dkdt
d|ƒS(NRËg@RÌRÁ(RÎR(RRR?R@tversionRÁ((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pytsend_notificationDs

cCs¤|jt
jƒ
}t||ƒ}|jd
|j|
ƒ
d|
jd|ƒ
|jd
|j|
ƒ
dd|
j|
jfd|ƒ
|j|
j|ƒ
t	j
|||ƒS(NRÂRÀtcallbacks%s.%s(RZRIR½Rptdeclare_topic_consumerR·RÀRÇtdeclare_fanout_consumerRtPollStyleListenerAdapter(RRt
batch_sizet
batch_timeoutRRR7((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyR£Hs

	


	




c	
Cs|jt
jƒ
}||j_t||ƒ}xL|
D]D\}}|jd
|j|ƒ
dd|j|fd|d|ƒ
q4Wt	j
|||ƒS(NRÂRÀs%s.%sRÒt
queue_name(RZRIR½t
connectiontrabbit_qos_prefetch_countRpRÓR·RÀRRÕ(	Rttargets_and_prioritiestpoolRÖR×RRR7Rtpriority((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pytlisten_for_notificationsYs

	




c

Cs}|jr|jj
ƒ
nd|_|jL
|jdk	rs|jjƒ
|jjƒ
d|_d|_d|_nWdQXdS(
N(	R¬temptyRR®R¯R±RR°R‡(
R((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRˆls
	

	






	
	
N(R2R3RXRRR·RIR[RZR¿RR)RÎRÏRÑR£RÞRˆ(((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyRØ
s

		
<
		(&t__all__tloggingRR$R»Rgt
oslo_utilsR
tsixRRtoslo_messaging._driversRRKRRRItoslo_messaging._i18nRRRR	t	getLoggerR2RNR&RƒR–R
tRpcIncomingMessageR5RdtPollStyleListenerRpRŠR˜t
BaseDriverR(((sW/home/tvault/.virtenv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.pyt<module>s2	












>^-W*a