Файловый менеджер - Редактировать - /opt/imh-python/lib/python3.9/site-packages/kombu/transport/__pycache__/consul.cpython-39.pyc
Ðазад
a o�h�$ � @ s� d Z ddlmZ ddlZddlZddlmZ ddlmZ ddl m Z ddl m Z ddl mZ dd lmZ dd lmZmZ dd lmZ d d lmZ z ddlZW n ey� dZY n0 ed�ZdZdZ G dd� de �Z G dd� dej �Z G dd� dej �Z dS )a� Consul 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)�dumps�loads)�cached_property� )�virtualzkombu.transport.consuli4! � localhostc @ s e Zd ZdZdS )� LockErrorz3An error occurred while trying to acquire the lock.N)�__name__� __module__� __qualname__�__doc__� r r ��/root/rpmbuild/BUILDROOT/imh-python39-modules-3.9.7-92.el8.x86_64/opt/imh-python/lib/python3.9/site-packages/kombu/transport/consul.pyr 9 s r c s� e Zd ZdZdZdZdZdZ� fdd�Zdd � Z d d � Z d d � Z dd� Z dd� Z eefdd��Zefdd�Zdd� Zdd� Zdd� Zd d � Zd d � Zd(d d!�Zd"d#� Zd$d%� Zed&d'� �Z� Z S ))�Channelz?Consul Channel class which talks to the Consul Key/Value store.ZkombuNZ10s� c sv t d u rtd��t� j|i |�� | jjjp2| jj}| jjjp@t }t � d||| j � t t�| _t j|t|�d�| _d S )N� Missing python-consul libraryz Host: %s Port: %s Timeout: %s��host�port)�consul� ImportError�super�__init__� connection�clientr � default_port�hostname� DEFAULT_HOST�logger�debug�timeoutr �dict�queues�Consul�int)�self�args�kwargsr r �� __class__r r r E s zChannel.__init__c C s | j � d|� d�S )N�/z.lock��prefix�r, �queuer r r � _lock_keyT s zChannel._lock_keyc C s | j � d|� �S )Nr1 r2 r4 r r r � _key_prefixW s zChannel._key_prefixc C sD z| j | d }W n ty( d}Y n0 |dur<| �|�S | �� S )a� Get 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, r5 r8 r r r �_get_or_create_sessionZ s � �z Channel._get_or_create_sessionc C s& t �d|� | jjj|d�}|�d�S )Nz#Trying to renew existing session %s)r8 ZID)r% r& r! �sessionZrenew�get)r, r8 r= r r r r: r s z Channel._renew_existing_sessionc C s< t �d| j| j� | jjj| j| jd�}t �d| j|� |S )Nz Creating session %s with TTL %s)�name�ttlz Created session %s with id %s)r% r& � lock_name� session_ttlr! r= �create)r, r8 r r r r; w s ���zChannel._create_new_sessionc c s4 | j ||d� zdV W | �|� n | �|� 0 dS )aI Try 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, r5 rE r r r � _queue_lock� s zChannel._queue_lockc C s` | � |�}| �|�}t�d||� | jjj||| jd�rJ|| j| d< d S t� d|� |� �d S )Nz/Trying to create lock object %s with session %s)�key�acquire�valuer8 z Could not acquire lock on key %s) r< r6 r% r& r! �kv�putrA r) �info)r, r5 rE r8 Zlock_keyr r r rF � s � � zChannel._acquire_lockc C s, t �d| �|�� | jjj| �|�d� dS )z�Try 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)rI N)r% r& r6 r! rL �deleter4 r r r rG � s zChannel._release_lockc C s2 t �d| j| d � | jj�| j| d � dS )z�Destroy a previously created Consul session. Will release all locks it still might hold. Arguments: --------- queue (str): The name of the Queue. zDestroying session %sr8 N)r% r& r) r! r= �destroyr4 r r r �_destroy_session� s zChannel._destroy_sessionc K s& dd i| j |< | jjj| �|�d d�S )Nr8 )rI rK )r) r! rL rM r7 )r, r5 �_r r r � _new_queue� s zChannel._new_queuec O s& | � |� | j�|d � | �|� d S �N)rQ r) �pop�_purge)r, r5 r- rR r r r �_delete� s zChannel._deletec K sR d� | �|�ttt� d ��t�� �}| jjj |t |�dd�sNt d|�d���dS )zaPut `message` onto `queue`. This simply writes a key to the K/V store of Consul z {}/msg/{}_{}i� r )rI rK �caszCannot add key z to consulN) �formatr7 r+ �roundr �uuid�uuid4r! rL rM r r )r, r5 �payloadrR rI r r r �_put� s �z Channel._putc C s� | j |td��� | �|�� d�}t�d|| j� | jjj|d| j| j d�\| _}zn|du r`t� �t�d|d d |d d � | jjj |d d |d d d � t |d d �W W d � S t y� Y n0 W d � n1 s�0 Y t� �dS ) z�Get 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 rD �/msg/z Fetching key %s with index %sT�rI �recurse�index�waitNz#Removing key %s with modifyindex %sr �KeyZ ModifyIndex)rI rX �Value) rH r r7 r% r& rb r! rL r> r' rO r � TypeError)r, r5 r' rI �datar r r �_get� s( � � � $z Channel._getc C s( | � |� | jjj| �|�� d�dd�S )Nr_ T)rI ra )rQ r! rL rO r7 r4 r r r rV � s �zChannel._purgec C sz d}zL| � |�� d�}t�d|| j� | jjj|d| j| jd�\| _}t|�}W n t yb Y n0 t�d||| j� |S )Nr r_ z)Fetching key recursively %s with index %sTr` z$Found %s keys under %s with index %s) r7 r% r& rb r! rL r> r' �lenrf )r, r5 �sizerI rg r r r �_size s" �� �z Channel._sizec C s t �� � S rT )�socket� gethostname�r, r r r rA s zChannel.lock_name)N) r r r r r3 rb r' rB r r6 r7 r< r: r; r r rH rF rG rQ rS rW r^ rh rV rk r rA � __classcell__r r r/ r r = s0 r c sp e Zd ZdZeZeZdZdZe rLe j j e j e jj f Z e j je j e jj f Z� fdd�Zdd� Zdd� Z� ZS ) � Transportz'Consul K/V storage Transport for Kombu.r c s& t d u rtd��t� j|i |�� d S )Nr )r r r r )r, r- r. r/ r r r - s zTransport.__init__c C sb |j jp | j}|j jpt}t�d||� z"tj|t |�d�}|j � � W dS t y\ Y n0 dS )Nz!Verify Consul connection to %s:%sr TF) r! r r" r# r$ r% r& r r* r+ Zagentr, � ValueError)r, r r r r! r r r �verify_connection3 s zTransport.verify_connectionc C s t jS rT )r � __version__rn r r r �driver_versionB s zTransport.driver_version)r r r r r � DEFAULT_PORTr" Z driver_typeZ driver_namer r rp Zconnection_errorsZConsulException�baseZchannel_errorsr rr rt ro r r r/ r rp s" �� �� rp )!r � __future__r rl r[ � collectionsr � contextlibr r5 r �timer Zkombu.exceptionsr Z kombu.logr Zkombu.utils.jsonr r Zkombu.utils.objectsr � r r r r% ru r$ � Exceptionr r rp r r r r �<module> s. [
| ver. 1.1 | |
.
| PHP 8.3.30 | Ð“ÐµÐ½ÐµÑ€Ð°Ñ†Ð¸Ñ Ñтраницы: 0 |
proxy
|
phpinfo
|
ÐаÑтройка