O'^c@@s`dZddlmZmZddlZddlZddlZddlZddlmZddl m Z ddl m Z ddl mZddlmZdd lmZmZdd lmZmZmZdd lmZmZmZmZdd lmZdd lm Z m!Z!ddl"m#Z#ddl$m%Z%ddl&m'Z'ej(ddkr_dndZ)dZ*dZ+de,fdYZ-de.fdYZ/de0fdYZ1de,fdYZ2de,fd YZ3d!e#j4fd"YZ4d#e,fd$YZ5d%e5e#j6fd&YZ7d'e#j8fd(YZ8d)e#j9fd*YZ9dS(+u kombu.transport.virtual ======================= Virtual transport implementation. Emulates the AMQ API for non-AMQ transports. i(tabsolute_importtunicode_literalsN(tarray(tcount(tFinalize(tsleep(tqueue_declare_ok_t(t ResourceErrort ChannelError(tEmptytitemst monotonic(temergency_dump_statetkwdicttsaytuuid(t OrderedDict(t str_to_bytest bytes_to_str(tbasei(t FairCycle(tSTANDARD_EXCHANGE_TYPESiuHtHulMessage could not be delivered: No queues bound to exchange {exchange!r} using binding key {routing_key!r}. ukCannot redeclare exchange {0!r} in vhost {1!r} with different type, durable, autodelete or arguments value.tBase64cB@seZdZdZRS(cC@sttjt|S(N(Rtbase64t b64encodeR(tselfts((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pytencode2scC@stjt|S(N(Rt b64decodeR(RR((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pytdecode5s(t__name__t __module__RR(((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyR0s tNotEquivalentErrorcB@seZdZRS(uAEntity declaration is not equivalent to the previous declaration.(RR t__doc__(((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyR!9stUndeliverableWarningcB@seZdZRS(u.The message could not be delivered to a queue.(RR R"(((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyR#>st BrokerStatecB@s,eZdZdZdddZdZRS(cC@s:|dkrin||_|dkr-in||_dS(N(tNonet exchangestbindings(RR&R'((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt__init__KscC@s|jj|jjdS(N(R&tclearR'(R((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyR)Os N(RR R%R&R'R(R)(((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyR$CstQoScB@seZdZdZd Zd ZeZddZ dZ dZ dZ dZ dZdZed Zd Zd Zd ZRS(uQuality of Service guarantees. Only supports `prefetch_count` at this point. :param channel: AMQ Channel. :keyword prefetch_count: Initial prefetch count (defaults to 0). icC@sy||_|pd|_t|_t|j_t|_|jj|_ |jj |_ t ||j dd|_dS(Nit exitpriorityi(tchanneltprefetch_countRt _deliveredtFalsetrestoredtsett_dirtytaddt _quick_ackt __setitem__t _quick_appendRtrestore_unacked_oncet _on_collect(RR,R-((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyR(ns    cC@s0|j}| p/t|jt|j|kS(uReturn true if the channel can be consumed from. Used to ensure the client adhers to currently active prefetch limits. (R-tlenR.R2(Rtpcount((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt can_consume{s cC@s:|j}|r6t|t|jt|jdSdS(uzReturns the maximum number of messages allowed to be returned. Returns an estimated number of messages that a consumer may be allowed to consume at once from the broker. This is used for services where bulk 'get message' calls are preferred to many individual 'get message' calls - like SQS. returns: An integer > 0 iN(R-tmaxR9R.R2(RR:((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pytcan_consume_max_estimates cC@s*|jr|jn|j||dS(u&Append message to transactional state.N(R2t_flushR6(Rtmessaget delivery_tag((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pytappends  cC@s |j|S(N(R.(RR@((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pytgetscC@sR|j}|j}x9y|j}Wntk r9PnX|j|dqWdS(u'Flush dirty (acked/rejected) tags from.N(R2R.tpoptKeyErrorR%(Rtdirtyt deliveredt dirty_tag((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyR>s   cC@s|j|dS(u8Acknowledge message and remove from transactional state.N(R4(RR@((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pytackscC@s1|r |jj|j|n|j|dS(u4Remove from transactional state and requeue message.N(R,t_restore_at_beginningR.R4(RR@trequeue((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pytrejectscC@s|j|j}g}|jj}|j}xi|ry|\}}Wntk r^PnXy||Wq1tk r}|j||fq1Xq1W|j|S(u$Restore all unacknowledged messages.( R>R.R,t_restoretpopitemRDt BaseExceptionRAR)(RRFterrorstrestoret pop_messaget_R?texc((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pytrestore_unackeds        cC@s|jj|j|j}|j s7|jj r;dSt|ddr^| sZt dSzp|rt dt |j|j }|rt t|\}}t dt ||t|qnWdt|_XdS(uyRestores all unacknowledged messages at shutdown/gc collect. Will only be done once for each instance. Nurestoredu*Restoring {0!r} unacknowledged message(s).u#UNABLE TO RESTORE {0} MESSAGES: {1}(R8tcancelR>R.trestore_at_shutdownR,t do_restoretgetattrR%tAssertionErrorRR9RTtlisttzipR tTrueR0(Rtstatet unrestoredROtmessages((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyR7s&     cO@sdS(uRestore any pending unackwnowledged messages for visibility_timeout style implementations. Optional: Currently only used by the Redis transport. N((Rtargstkwargs((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pytrestore_visiblesN(RR R"R-R%R.R2R\RVR(R;R=RARBR>RHR/RKRTR7Rb(((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyR*Ts        tMessagecB@seZdZdZRS(cK@s||_|d}|jd}|rF|j||jd}n|ji|d6|dd6|jdd6|jdd6|jd d 6|d6|jd d 6d d 6tt|j|t|dS( Nu propertiesubodyu body_encodingu delivery_tagu content-typeu content_typeucontent-encodingucontent_encodinguheadersu delivery_infouutf-8u postencode(t_rawRBt decode_bodytupdatetsuperRcR(R (RR,tpayloadRat propertiestbody((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyR(s     cC@s||j}|jj|j|jd\}}t|j}|jddi|d6|d6|j d6|j d6|d6S(Nu body_encodingu compressionubodyu propertiesu content-typeucontent-encodinguheaders( RiR,t encode_bodyRjRBtdicttheadersRCR%t content_typetcontent_encoding(RtpropsRjRRRm((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt serializables   (RR R(Rq(((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyRcs tAbstractChannelcB@s\eZdZd dZdZdZdZdZdZ dZ d dZ RS( uThis is an abstract class defining the channel methods you'd usually want to implement in a virtual channel. Do not subclass directly, but rather inherit from :class:`Channel` instead. cC@stddS(uGet next message from `queue`.u$Virtual channels must implement _getN(tNotImplementedError(Rtqueuettimeout((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt_getscC@stddS(uPut `message` onto `queue`.u$Virtual channels must implement _putN(Rs(RRtR?((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt_putscC@stddS(u!Remove all messages from `queue`.u&Virtual channels must implement _purgeN(Rs(RRt((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt_purge#scC@sdS(u<Return the number of messages in `queue` as an :class:`int`.i((RRt((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt_size'scO@s|j|dS(u{Delete `queue`. This just purges the queue, if you need to do more you can override this method. N(Rx(RRtR`Ra((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt_delete+scK@sdS(uCreate new queue. Your transport can override this method if it needs to do something whenever a new queue is declared. N((RRtRa((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt _new_queue4scK@stS(uVerify that queue exists. Should return :const:`True` if the queue exists or :const:`False` otherwise. (R\(RRtRa((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt _has_queue=scC@s |jS(u-Poll a list of queues for available messages.(RB(RtcycleRu((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt_pollFsN( RR R"R%RvRwRxRyRzR{R|R~(((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyRrs     tChannelcB@s^eZdZeZeZeZeeZ e Z ie d6Z dZedZd0Zd1ZdZd0de e d0e e dZe e dZd0e d Ze e d Zd Zd d e d0d Zd d e d0dZd0d d0dZd0d d0dZdZdZdZ dZ!dZ"dZ#e dZ$dZ%e dZ&e dZ'dde dZ(dZ)dZ*ddZ+d0d Z,d!Z-d"Z.d0d#Z/d$Z0d0d0d0d0d0d%Z1ed&Z2d'Z3d0d(Z4d0d)Z5d*Z6d+Z7d,Z8e9d-Z:e9d.Z;e9d/Z<RS(2u^Virtual channel. :param connection: The transport instance this channel is part of. ubase64iu body_encodingudeadletter_queuec @s|_t_d_i_g_d_t_ t fdt j D_ yjj j_Wn>tk rtdjtjjjjdnXjjj}x=jD]2}yt|||Wqtk rqXqWdS(Nc3@s'|]\}}||fVqdS(N((t.0ttyptcls(R(sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pys }su1No free channel ids, current={0}, channel_max={1}ii (ii (t connectionR1t _consumersR%t_cyclet _tag_to_queuet_active_queuest_qosR/tclosedRlR texchange_typest_avail_channel_idsRCt channel_idt IndexErrorRtformatR9tchannelst channel_maxtclientttransport_optionstfrom_transport_optionstsetattrRD(RRRattoptstopt_name((RsB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyR(rs,       %  udirectc C@s!|p d}|pd|}|rk||jjkrgtdj||jjjpRddddnd Sye|jj|}|j|j||||||st t j||jjjpdnWnJt k ri|d 6|d 6|d 6|pid 6gd6|jj|s(R]R&(R((RsB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt list_bindingss cK@s |j|S(u%Remove all ready messages from queue.(Rx(RRtRa((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt queue_purgescC@stS(N(R(R((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt_next_delivery_tagscK@s|j|d|j\|d<}|d}|jd|d|j|djd|d||r|j|j||||S|j|||S(uPublish message.ubodyu propertiest body_encodingR@u delivery_infoRR(RkRRfRRtdeliverRw(RR?RRRaRRp((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt basic_publishs    c @s`|j|<jj|fd}|jj|<jj|jdS(uConsume from `queue`c@s;j|}s1jj||jn|S(N(RctqosRAR@(t raw_messageR?(tcallbacktno_ackR(sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt _callbacksN(RRRARt _callbacksRR3t _reset_cycle(RRtRRt consumer_tagRaR((RRRsB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt basic_consumes  cC@s||jkr|jj||j|jj|d}y|jj|Wntk renX|jj j|dndS(u Cancel consumer by consumer tag.N( RtremoveRRRCR%Rt ValueErrorRR(RRRt((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt basic_cancels  cK@sZyB|j||j|}|s=|jj||jn|SWntk rUnXdS(u+Get message by direct access (synchronous).N(RcRvRRAR@R (RRtRRaR?((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt basic_get*s cC@s|jj|dS(uAcknowledge message.N(RRH(RR@((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt basic_ack4scC@s#|r|jjStddS(uRecover unacked messages.u'Does not support recover(requeue=False)N(RRTRs(RRJ((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt basic_recover8s cC@s|jj|d|dS(uReject message.RJN(RRK(RR@RJ((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt basic_reject>sicC@s||j_dS(u\Change QoS settings for this channel. Only `prefetch_count` is supported. N(RR-(Rt prefetch_sizeR-t apply_global((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt basic_qosBscC@st|jjS(N(RZR]R&(R((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt get_exchangesKscC@s|jj|dS(u%Get table of bindings for `exchange`.utable(R]R&(RR((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyRNscC@s=y|jj|d}Wntk r1|}nX|j|S(u.Get the exchange type instance for `exchange`.utype(R]R&RDR(RRtdefaultR((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyRRs   cC@s|dkr|j}ny.|j|j|j||||}Wntk r_g}nX| r|dk rtjtt j d|d||j ||g}n|S(uFind all queues matching `routing_key` for the given `exchange`. Must return the string `default` if no queues matched. RRN( R%tdeadletter_queueRtlookupRRDtwarningstwarnR#tUNDELIVERABLE_FMTRR{(RRRRtR((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt_lookupZs       cC@sX|j}|j}t|dg|jjD]*\}}}td|d|d|^qS(NRRR(R,RRl(Rtqtetr((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt get_bindingsscC@s|jjdS(N(R,R(R((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyRs(RR R(RR(((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyRs  t TransportcB@seZdZeZeZeZeZd Z d Z d Z d Z dZdZdZdZdZdZdZd dZd d Zed ZRS( uPVirtual transport. :param client: :class:`~kombu.Connection` instance g?icK@s||_g|_g|_i|_|j|j|jt|_|jj d}|dk rl||_ nt t t|jdd|_dS(Nupolling_intervalii(RRt_avail_channelsRtCyclet_drain_channelR R}RRBR%tpolling_intervalRt ARRAY_TYPE_HtrangeRR(RRRaR((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyR(s      cC@sIy|jjSWn1tk rD|j|}|jj||SXdS(N(RRCRRRRA(RRR,((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pytcreate_channels  cC@sPz?|jj|jy|jj|Wntk r=nXWdd|_XdS(N(RRARRRRR%R(RR,((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyRs cC@s|jj|j||S(N(RRAR(R((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pytestablish_connection%scC@so|jjx[|j|jfD]G}x>|rfy|j}Wnttfk rXq)X|jq)Wq WdS(N(R}RRRRCRRD(RRtlR,((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pytclose_connection,s  c C@sd}t}|jj}|j}x}y|d|\}}Wn\tk r|rxt||krxtjn|d7}|dk rt|qq'XPq'W|\} } | s| |j krt dj | | n|j | | dS(NiRuiu.Message for queue {0!r} without consumers: {1}( R R}RBRR tsocketRuR%RRRDR( RRRutloopt time_startRBRtitemR,R?Rt((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyR7s(       cC@s|jd|S(NRu(R(RR,Ru((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyRQscC@si|jd6dd6S(Nuportu localhostuhostname(t default_port(R((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pytdefault_connection_paramsTsN(RR R"RRRRR$R]R%R}RRRRRR(RRRRRRRR(((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyRs&     (:R"t __future__RRRRtsysRRt itertoolsRtmultiprocessing.utilRttimeRt amqp.protocolRtkombu.exceptionsRRt kombu.fiveR R R t kombu.utilsR R RRtkombu.utils.compatRtkombu.utils.encodingRRtkombu.transportRt schedulingRRRt version_infoRRRtobjectRt ExceptionR!t UserWarningR#R$R*RcRrt StdChannelRRR(((sB/tmp/pip-unpacked-wheel-UAnTfW/kombu/transport/virtual/__init__.pyt s@    " $9