o wvXjc#@sdZddlmZmZddlZddlZddlmZddlm Z ddl m Z ddl m Z mZddlmZdd lmZmZdd lmZd d lmZzddlZWn eyWdZYnwed ZdZdZGdddeZGdddejZGdddej Z dS)zConsul Transport. It uses Consul.io's Key/Value store to transport messages in Queues It uses python-consul for talking to Consul's HTTP API )absolute_importunicode_literalsN) defaultdict)contextmanager) ChannelError)Empty monotonic) get_logger)loadsdumps)cached_property)virtualzkombu.transport.consuli4! localhostc@seZdZdZdS) LockErrorz3An error occurred while trying to acquire the lock.N)__name__ __module__ __qualname____doc__rrS/var/www/html/myproject/venv/lib/python3.10/site-packages/kombu/transport/consul.pyr"srcseZdZdZdZdZdZdZfddZdd Z d d Z d d Z ddZ ddZ eefddZefddZddZddZddZddZddZd(d d!Zd"d#Zd$d%Zed&d'ZZS))Channelz?Consul Channel class which talks to the Consul Key/Value store.kombuN10scsztdurtdtt|j|i||jjjp|jj}|jjj p"t }t d|||j tt|_tj|t|d|_dS)NMissing python-consul libraryzHost: %s Port: %s Timeout: %shostport)consul ImportErrorsuperr__init__ connectionclientr default_porthostname DEFAULT_HOSTloggerdebugtimeoutrdictqueuesConsulint)selfargskwargsrr __class__rrr".s zChannel.__init__cCd|j|S)Nz {0}/{1}.lockformatprefixr/queuerrr _lock_key=zChannel._lock_keycCr4)Nz{0}/{1}r5r8rrr _key_prefix@r;zChannel._key_prefixcCsDz |j|d}Wn tyd}Ynw|dur||S|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)r,KeyError_renew_existing_session_create_new_session)r/r9r=rrr_get_or_create_sessionCs  zChannel._get_or_create_sessioncCs&td||jjj|d}|dS)Nz#Trying to renew existing session %s)r=ID)r(r)r$sessionrenewget)r/r=rCrrrr?Ys  zChannel._renew_existing_sessioncCs<td|j|j|jjj|j|jd}td|j||S)NzCreating session %s with TTL %s)namettlzCreated session %s with id %s)r(r) lock_name session_ttlr$rCcreate)r/r=rrrr@^szChannel._create_new_sessionc cs4|j||dz dVW||dS||w)aTry 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)r/r9rLrrr _queue_lockgs zChannel._queue_lockcCs\||}||}td|||jjj|||jdr%||j|d<dSt d||)Nz/Trying to create lock object %s with session %s)keyacquirevaluer=z Could not acquire lock on key %s) rAr:r(r)r$kvputrHr,info)r/r9rLr=lock_keyrrrrMs    zChannel._acquire_lockcCs,td|||jjj||ddS)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)rPN)r(r)r:r$rSdeleter8rrrrNs zChannel._release_lockcCs2td|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 %sr=N)r(r)r,r$rCdestroyr8rrr_destroy_sessionszChannel._destroy_sessioncKs&ddi|j|<|jjj||ddS)Nr=)rPrR)r,r$rSrTr<)r/r9_rrr _new_queueszChannel._new_queuecOs&|||j|d||dSN)rYr,pop_purge)r/r9r0rZrrr_deletes zChannel._deletecKsPd||tttdt}|jjj |t |dds&t d|dS)zaPut `message` onto `queue`. This simply writes a key to the K/V store of Consul z{0}/msg/{1}_{2}ir)rPrRcaszCannot add key {0!r} to consulN) r6r<r.roundruuiduuid4r$rSrTr r)r/r9payloadrZrPrrr_putsz Channel._putc Cs|j|tdkd||}td||j|jjj |d|j|j d\|_}z7|dur0ttd|dd |dd |jjj |dd |dd d t |dd WWdSt yhYnwWdt1suwYt) 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 rK{0}/msg/zFetching key %s with index %sTrPrecurseindexwaitNz#Removing key %s with modifyindex %srKey ModifyIndex)rPr`Value)rOrr6r<r(r)rir$rSrEr*rWr TypeError)r/r9r*rPdatarrr_gets4    z Channel._getcCs(|||jjjd||ddS)NrfT)rPrh)rYr$rSrWr6r<r8rrrr^s zChannel._purgecCszd}z&d||}td||j|jjj|d|j|jd\|_}t |}Wn t y1Ynwtd|||j|S)Nrrfz)Fetching key recursively %s with index %sTrgz$Found %s keys under %s with index %s) r6r<r(r)rir$rSrEr*lenrn)r/r9sizerProrrr_sizes$  z Channel._sizecCsdtS)Nz{0})r6socket gethostnamer/rrrrHszChannel.lock_namer\)rrrrr7rir*rIr"r:r<rAr?r@rrrOrMrNrYr[r_rerpr^rsr rH __classcell__rrr2rr&s2      rcs@eZdZdZeZeZdZdZfddZ ddZ ddZ Z S) Transportz'Consul K/V storage Transport for Kombu.rcsZtdurtdtt|j|i|tjjtjtjjf|_tjj tjtjjf|_ dS)Nr) rr r!rxr"rconnection_errorsConsulExceptionbasechannel_errors)r/r0r1r2rrr"s  zTransport.__init__cCs`|jjp|j}|jjp t}td||ztj|t |d}|j WdSt y/YdSw)Nz!Verify Consul connection to %s:%srTF) r$rr%r&r'r(r)rr-r.agentr/ ValueError)r/r#rrr$rrrverify_connections   zTransport.verify_connectioncCstjSr\)r __version__rvrrrdriver_version#szTransport.driver_version) rrrrr DEFAULT_PORTr% driver_type driver_namer"rrrwrrr2rrxs rx)!r __future__rrrbrt collectionsr contextlibrkombu.exceptionsr kombu.fiverr kombu.logr kombu.utils.jsonr r kombu.utils.objectsr rrr r(rr' Exceptionrrrxrrrrs0        T