^f$BdZddlmZddlZddlZddlmZddlmZddl m Z ddl m Z ddl mZdd lmZdd lmZmZdd lmZd d lmZ ddlZn #e$rdZYnwxYwedZdZdZGddeZGddejZGddej Z dS)aConsul Transport module for Kombu. Features ======== It uses Consul.io's Key/Value store to transport messages in Queues It uses python-consul for talking to Consul's HTTP API Features ======== * Type: Native * Supports Direct: Yes * Supports Topic: *Unreviewed* * Supports Fanout: *Unreviewed* * Supports Priority: *Unreviewed* * Supports TTL: *Unreviewed* Connection String ================= Connection string has the following format: .. code-block:: consul://CONSUL_ADDRESS[:PORT] ) annotationsN) defaultdict)contextmanager)Empty) monotonic) ChannelError) get_logger)dumpsloads)cached_property)virtualzkombu.transport.consuli4! localhostceZdZdZdS) LockErrorz3An error occurred while trying to acquire the lock.N)__name__ __module__ __qualname____doc__H/var/www/html/env/lib/python3.11/site-packages/kombu/transport/consul.pyrr9s====rrceZdZdZdZdZdZdZfdZdZ dZ d Z d Z d Z eefd Zefd ZdZdZdZdZdZddZdZdZedZxZS)Channelz?Consul Channel class which talks to the Consul Key/Value store.kombuN10scttdtj|i||jjjp |jj}|jjjpt}t d|||j tt|_tj|t#||_dS)NMissing python-consul libraryzHost: %s Port: %s Timeout: %shostport)consul ImportErrorsuper__init__ connectionclientr" default_porthostname DEFAULT_HOSTloggerdebugtimeoutrdictqueuesConsulint)selfargskwargsr"r! __class__s rr&zChannel.__init__Es >=>> >$)&)))%*Jdo.J%.>, 4dD$,OOO!$'' mCII>>> rc|jd|dS)N/z.lockprefixr3queues r _lock_keyzChannel._lock_keyTs+,,,,,,rc|jd|S)Nr8r9r;s r _key_prefixzChannel._key_prefixWs+'''''rc |j|d}n#t$rd}YnwxYw|||n|S)aGet or create consul session. Try to renew the session if it exists, otherwise create a new session in Consul. This session is used to acquire a lock inside Consul so that we achieve read-consistency between the nodes. Arguments: --------- queue (str): The name of the Queue. Returns ------- str: The ID of the session. session_idN)r0KeyError_renew_existing_session_create_new_session)r3r<rAs r_get_or_create_sessionzChannel._get_or_create_sessionZsq" U+L9JJ   JJJ ),,Z888/3/G/G/I/I Ks  %%ctd||jj|}|dS)Nz#Trying to renew existing session %s)rAID)r,r-r(sessionrenewget)r3rArHs rrCzChannel._renew_existing_sessionrsD :JGGG+%++z+BB{{4   rctd|j|j|jj|j|j}td|j||S)NzCreating session %s with TTL %s)namettlzCreated session %s with id %s)r,r- lock_name session_ttlr(rHcreate)r3rAs rrDzChannel._create_new_sessionwst 6^T%5 7 7 7[(//T%5077  4^Z 1 1 1rc#K||| dV||dS#||wxYw)aITry 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. raising (Exception): Set custom lock error class. Raises ------ LockError: if the lock cannot be acquired. Returns ------- bool: success? raisingN) _acquire_lock _release_lock)r3r<rSs r _queue_lockzChannel._queue_locksc. 5'222 & EEE   u % % % % %D  u % % % %s 6A cJ||}||}td|||jj|||jr||j|d<dSt d||)Nz/Trying to create lock object %s with session %s)keyacquirevaluerAz Could not acquire lock on key %s) rEr=r,r-r(kvputrNr0info)r3r<rSrAlock_keys rrTzChannel._acquire_locks0077 >>%(( Fz + + + ;>  (&0$(N  4 4 0:DK | , F 6AAAgiirctd|||jj||dS)zTry to release a lock. It does so by simply removing the lock key in Consul. Arguments: --------- queue (str): The name of the queue we want to release the lock from. zRemoving lock key %s)rXN)r,r-r=r(r[deleter;s rrUzChannel._release_locksR  +T^^E-B-BCCC $.."7"788888rctd|j|d|jj|j|ddS)zDestroy a previously created Consul session. Will release all locks it still might hold. Arguments: --------- queue (str): The name of the Queue. zDestroying session %srAN)r,r-r0r(rHdestroyr;s r_destroy_sessionzChannel._destroy_sessionsP  ,dk%.@.NOOO ##DK$6|$DEEEEErc ddi|j|<|jj||dS)NrA)rXrZ)r0r(r[r\r?)r3r<_s r _new_queuezChannel._new_queues>*D1 E{~!!d&6&6u&=&=T!JJJrc|||j|d||dSN)rcr0pop_purge)r3r<r4res r_deletezChannel._deletesD e$$$ t$$$ Erc Rd||ttt dzt j}|jj |t|dstd|ddS)zaPut `message` onto `queue`. This simply writes a key to the K/V store of Consul z {}/msg/{}_{}ir)rXrZcaszCannot add key z to consulN) formatr?r2roundruuiduuid4r(r[r\r r)r3r<payloadrerXs r_putz Channel._puts ##   U # # ikkD()) * * JLL   {~!!cwQ!GG DBBBBCC C D Drc||t5||d}td||j|jj|d|j|j \|_} |ttd|dd |dd |jj |dd |dd  t|dd cdddS#t$rYnwxYw dddn #1swxYwYt) zGet the first available message from the queue. Before it does so it acquires a lock on the Key/Value store so only one node reads at the same time. This is for read consistency rR/msg/zFetching key %s with index %sTrXrecurseindexwaitNz#Removing key %s with modifyindex %srKey ModifyIndex)rXrmValue) rVrr?r,r-rxr(r[rJr.r`r TypeError)r3r<r.rXdatas r_getz Channel._gets   eU  3 3  %%e,,333C LL8#tz J J J#{~11jt| 2   DJ <''M B!!WU^T!W]-CEEE %%$q'%.*.q'-*@&BBBT!WW-..%        &    )               ,gg s1A/E BD11 D>;E=D>>EEEc|||jj||ddS)NruT)rXrw)rcr(r[r`r?r;s rrjzChannel._purgesT e$$${~$$##E**111%   rcld} ||d}td||j|jj|d|j|j\|_}t|}n#t$rYnwxYwtd|||j|S)Nrruz)Fetching key recursively %s with index %sTrvz$Found %s keys under %s with index %s) r?r,r-rxr(r[rJr.lenr})r3r<sizerXr~s r_sizez Channel._sizes %%e,,333C LLDdj * * *#{~11jt| 2   DJt99DD    D   ;3  , , , sA=B BBc*tjSrh)socket gethostnamer3s rrNzChannel.lock_names$&&((rrh)rrrrr:rxr.rOr&r=r?rErCrDrrrVrTrUrcrfrkrsrrjrr rN __classcell__r6s@rrr=s}II F EGK ? ? ? ? ?---(((KKK0!!! )2&&&^&8,5     9 9 9 F F FKKK D D D<   $))_)))))rrceZdZdZeZeZdZdZe r@e j j e j e jj fzZ e j je j e jj fzZfdZdZdZxZS) Transportz'Consul K/V storage Transport for Kombu.r#cfttdtj|i|dS)Nr)r#r$r%r&)r3r4r5r6s rr&zTransport.__init__-s9 >=>> >$)&)))))rc(|jjp|j}|jjpt}t d|| tj|t|}|j dS#t$rYnwxYwdS)Nz!Verify Consul connection to %s:%sr TF) r(r"r)r*r+r,r-r#r1r2agentr3 ValueError)r3r'r"r!r(s rverify_connectionzTransport.verify_connection3s %:): )9\ 8$EEE ]3t99===F L     4    D usrs:#""""" ######%%%%%%)))))) ))))))))//////MMMM FFF , - -  >>>>> >>>W)W)W)W)W)goW)W)W)t,",",",","!,",",",","s AAA