^f!dZddlmZddlZddlZddlmZddlmZddl m Z ddl m Z ddl mZdd lmZmZdd lmZd d lmZ ddlZn #e$rdZYnwxYwed ZdZdZGddejZGddejZdS)aEtcd Transport module for Kombu. It uses Etcd as a store to transport messages in Queues It uses python-etcd for talking to Etcd's HTTP API Features ======== * Type: Virtual * Supports Direct: *Unreviewed* * Supports Topic: *Unreviewed* * Supports Fanout: *Unreviewed* * Supports Priority: *Unreviewed* * Supports TTL: *Unreviewed* Connection String ================= Connection string has the following format: .. code-block:: 'etcd'://SERVER:PORT ) annotationsN) defaultdict)contextmanager)Empty) ChannelError) get_logger)dumpsloads)cached_property)virtualzkombu.transport.etcdiK localhostceZdZdZdZdZdZdZdZfdZ dZ e dZ d Z d Zd Zd Zdd ZdZdZedZxZS)Channelz+Etcd Channel class which talks to the Etcd.kombuN cttdtj|i||jjjp |jj}|jjjpt}t d|||j tt|_tj|t#||_dS)NMissing python-etcd libraryzHost: %s Port: %s Timeout: %shostport)etcd ImportErrorsuper__init__ connectionclientr default_porthostname DEFAULT_HOSTloggerdebugtimeoutrdictqueuesClientint)selfargskwargsrr __class__s F/var/www/html/env/lib/python3.11/site-packages/kombu/transport/etcd.pyrzChannel.__init__>s <;<< <$)&)))%*Jdo.J%.>, 4dD$,OOO!$'' kt#d))<<< c|jd|S)zCreate and return the `queue` with the proper prefix. Arguments: --------- queue (str): The name of the queue. /)prefix)r)queues r- _key_prefixzChannel._key_prefixMs+'''''r.c#Ktj|j|}|j|_t d|j|d|j  dVt d|j| dS#t d|j| wxYw)ayTry to acquire a lock on the Queue. It does so by creating a object called 'lock' which is locked by the current session.. This way other nodes are not able to write to the lock object which means that they have to wait before the lock is released. Arguments: --------- queue (str): The name of the queue. zAcquiring lock T)blockinglock_ttlNzReleasing lock ) rLockr lock_value_uuidr"r#nameacquirer6release)r)r2locks r- _queue_lockzChannel._queue_lockVsye,,_  2ty22333 dT] ;;;  EEE LL64966 7 7 7 LLNNNNN LL64966 7 7 7 LLNNNNs (B$$8Cc ||j|<||5 |j||ddcdddS#t j$r[td|d|j ||cYcdddSwxYw#1swxYwYdS)zCreate a new `queue` if the `queue` doesn't already exist. Arguments: --------- queue (str): The name of the queue. TN)keydirvaluezQueue "z" already existsr@) r&r>rwriter3r EtcdNotFiler"r#read)r)r2_s r- _new_queuezChannel._new_queuenso# E   e $ $ E E E{((((//T)GG E E E E E E E E# E E E >u>>>???{''D,<,r3rrDr r)r)r2payloadrGr@s r-_putz Channel._puts  e $ $ F F""5))C;$$..%!! F##DS#D#D#DEEE  F F F F F F F F F F F F F F F F F F FsAA77A;>A;c||5||}td||j |j|d|j|j}|t|j d}td |dt|d}|j |d |cdddS#tttjf$r7}td t#|d |Yd}~nd}~wwxYwt#1swxYwYdS) a[Get the first available message from the queue. Before it does so it acquires a lock on the store so only one node reads at the same time. This is for read consistency Arguments: --------- queue (str): The name of the queue. timeout (int): Optional seconds to wait for a response. zFetching key %s with index %sT)r@ recursiveindexr$NzRemoving key {}r@rBrCz _get failed: :)r>r3r"r#rVrrFr$r _childrenformatr delete TypeError IndexErrorr EtcdExceptiontype)r)r2r$r@resultitem msg_contenterrors r-_getz Channel._gets  e $ $  ""5))C LL8#tz J J J D))t*dl*<<>''M'+ .55d5kBBCCC#DM22  ""tE{"333"#        $z4+=> D D D BT%[[BB5BBCCCCCCCC D''M+          s67E.B/D  E&-EE.EE..E25E2c||5||}td||j|dcdddS#1swxYwYdS)zRemove all `message`s from a `queue`. Arguments: --------- queue (str): The name of the queue. zPurging queue at key T)r@rUN)r>r3r"r#rr[)r)r2r@s r-rNzChannel._purges  e $ $ ? ?""5))C LL666 7 7 7;%%#%>> ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?sAA11A58A5c||5d} ||}td||j|j|d|j}t|j}n#t$rYnwxYwtd|||j|cdddS#1swxYwYdS)z~Return the size of the `queue`. Arguments: --------- queue (str): The name of the queue. rz)Fetching key recursively %s with index %sT)r@rUrVz$Found %s keys under %s with index %sN) r>r3r"r#rVrrFlenrYr\)r)r2sizer@r`s r-_sizez Channel._sizes:  e $ $  D &&u-- H $*...))t**&&6+,,     LL?sDJ 0 0 0                  s5CA,BC BCB&CC C cTtjdtjS)N.)socket gethostnameosgetpid)r)s r-r8zChannel.lock_values%$&&66666r.)N)__name__ __module__ __qualname____doc__r1rVr$ session_ttlr6rr3rr>rHrKrOrSrdrNrir r8 __classcell__r,s@r-rr5s55 F EGKH = = = = =(((^.EEE   FFF$    D ? ? ?077_77777r.rceZdZdZeZeZdZdZdZ e j j edgZ er*e j jejfzZe j jejfzZfdZdZd ZxZS) Transportz!Etcd storage Transport for Kombu.r python-etcddirect) exchange_typecfttdtj|i|dS)z(Create a new instance of etcd.Transport.Nr)rrrr)r)r*r+r,s r-rzTransport.__init__ s9 <;<< <$)&)))))r.c|jjp|j}|jjpt}t d|| tj|t|dS#t$rYnwxYwdS)zVerify the connection works.zVerify Etcd connection to %s:%srTF) rrrr r!r"r#rr'r( ValueError)r)rrrs r-verify_connectionzTransport.verify_connections %:): )9\ 6dCCC  KTD 2 2 2 24    D us#A)) A65A6c ddl}|jjD]4}|dr|ddcS5dS#t t f$rtdYdSwxYw)zReturn the version of the etcd library. .. note:: python-etcd has no __version__. This is a workaround. rNryz==r z'Unable to find the python-etcd version.Unknown) pip.commands.freezecommandsfreeze startswithsplitrr]r"warning)r)pipxs r-driver_versionzTransport.driver_versions   & & & &\(//11 , ,<< ..,774==++++, , ,Z(    NND E E E99 sAAA+B  B )rprqrrrsr DEFAULT_PORTr driver_type driver_namepolling_intervalr rx implementsextend frozensetrconnection_errorsr^channel_errorsrrrrurvs@r-rxrxs++GLKK"-44i ++5--J    /43E2H H    ,0B/E E *****          r.rx)rs __future__rrnrl collectionsr contextlibrr2rkombu.exceptionsr kombu.logrkombu.utils.jsonr r kombu.utils.objectsr r rrr"rr!rrxr.r-rs4#""""" ######%%%%%%)))))) ))))))))//////KKKK DDD * + +  {7{7{7{7{7go{7{7{7|99999!99999sA AA