UV.j!dZddlmZddlZddlZddlmZddlmZddl m Z ddl m Z ddl mZdd lmZmZdd lmZd d lmZ ddlZed ZdZdZGddej6ZGddej8Zy#e$rdZY@wxYw)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 ct tdt| |i||jj j xs|jj}|jj jxst}tjd|||jtt|_tj |t#||_y)NMissing python-etcd libraryzHost: %s Port: %s Timeout: %shostport)etcd ImportErrorsuper__init__ connectionclientr default_porthostname DEFAULT_HOSTloggerdebugtimeoutrdictqueuesClientint)selfargskwargsrr __class__s VC:\xampp\htdocs\tradingbinance\backend\.venv\Lib\site-packages\kombu/transport/etcd.pyrzChannel.__init__>s <;< < $)&)%%**Jdoo.J.J%%..>, 4dD$,,O!$' kkt#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++aw''r.c#Ktj|j|}|j|_t j d|j|jd|j dt j d|j|jy#t j d|j|jwxYww)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_lockVsyye,__  tyyk23 dT]] ;   LL?499+6 7 LLN LL?499+6 7 LLNsA1C"4B+83C"+4CC"c ||j|<|j|5 |jj|j |ddcdddS#t j $rPtjd|d|jj|j |cYcdddSwxYw#1swYyxYw)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_queuens# E   e $ E{{((((/T)G% $## E wug-=>?{{''D,<,r3rrDr r)r)r2payloadrGr@s r-_putz Channel._putsm  e $""5)C;;$$.%!#_SG8#DEE !% $ $s AA$$A-c|j|5|j|}tjd||j |j j |d|j|j}| t|jd}tjdj|dt|d}|j j|d |cdddS#tttj f$r7}tjd t#|d |Yd}~td}~wwxYw#1swYyxYw) 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#tzz J D))t**dll*<>'M''+ .55d5kBC#DM2  ""tE{"3"#% $$z4+=+=> D }T%[M5'BCC'M D%% $s03E B#C33E$E5E EE  Ec|j|5|j|}tjd||jj |dcdddS#1swYyxYw)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._purgesY  e $""5)C LL06 7;;%%#%>% $ $s AA""A+c|j|5d} |j|}tjd||j|j j |d|j}t|j}tjd||j|cdddS#t$rY8wxYw#1swYyxYw)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?sDJJ 0% $  % $s/CA/B2#C2 B>;C=B>>CC cXtjdtjS)N.)socket gethostnameosgetpid)r)s r-r8zChannel.lock_values#$$&'q 66r.)N)__name__ __module__ __qualname____doc__r1rVr$ session_ttlr6rr3rr>rHrKrOrSrdrNrir r8 __classcell__r,s@r-rr5sw5 F EGKH =(.E F$ D ?077r.rc0eZdZdZeZeZdZdZdZ e jjjedgZ erHe jj ej"fzZe jj$ej"fzZfdZdZd ZxZS) Transportz!Etcd storage Transport for Kombu.r python-etcddirect) exchange_typecFt tdt| |i|y)z(Create a new instance of etcd.Transport.Nr)rrrr)r)r*r+r,s r-rzTransport.__init__ s& <;< < $)&)r.c|jjxs |j}|jjxst}t j d|| tj|t|y#t$rYywxYw)zVerify the connection works.zVerify Etcd connection to %s:%srTF) rrrr r!r"r#rr'r( ValueError)r)rrrs r-verify_connectionzTransport.verify_connectionst  %%:):):  ))9\ 6dC  KKTD 2   s A<< BBc ddl}|jjjD])}|jds|j ddcSy#t t f$rtjdYywxYw)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_versionsl   &\\((//1<< .774=++2Z(  NND E s=A>)rprqrrrsr DEFAULT_PORTr driver_type driver_namepolling_intervalr rx implementsextend frozensetrconnection_errorsr^channel_errorsrrrrurvs@r-rxrxs+GLKK""--44 +5-J    / /43E3E2H H     , ,0B0B/E E *  r.rx)rs __future__rrnrl collectionsr contextlibrr2rkombu.exceptionsr kombu.logrkombu.utils.jsonr r kombu.utils.objectsr r rrr"rr!rrxr.r-rs4# #%) )/ * +  {7goo{7|9!!9O DsA??B B