File: //usr/local/lib/python3.9/site-packages/kombu/transport/__pycache__/consul.cpython-39.pyc
a
X>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 �@/usr/local/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)�nameZttlz 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 rD 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�putr@ r) �info)r, r5 rD r8 Zlock_keyr r r rE � 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)rH N)r% r& r6 r! rK �deleter4 r r r rF � 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 )rH rJ )r) r! rK rL r7 )r, r5 �_r r r �
_new_queue� s zChannel._new_queuec O s& | � |� | j�|d � | �|� d S �N)rP r) �pop�_purge)r, r5 r- rQ 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 )rH rJ �caszCannot add key z
to consulN) �formatr7 r+ �roundr �uuid�uuid4r! rK rL r r )r, r5 �payloadrQ rH 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
rC �/msg/z Fetching key %s with index %sT�rH �recurse�index�waitNz#Removing key %s with modifyindex %sr �KeyZ ModifyIndex)rH rW �Value)
rG r r7 r% r& ra r! rK r> r' rN r
� TypeError)r, r5 r' rH �datar r r �_get� s( � �
� $z Channel._getc C s( | � |� | jjj| �|�� d�dd�S )Nr^ T)rH r` )rP r! rK rN r7 r4 r r r rU � 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& ra r! rK r> r' �lenre )r, r5 �sizerH rf r r r �_size s" �� �z
Channel._sizec C s
t �� � S rS )�socket� gethostname�r, r r r r@ s zChannel.lock_name)N) r r r r r3 ra r' rA r r6 r7 r< r: r; r r rG rE rF rP rR rV r] rg rU rj r r@ �
__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 rS )r � __version__rm r r r �driver_versionB s zTransport.driver_version)r r r r r � DEFAULT_PORTr" Z driver_typeZ driver_namer r
ro Zconnection_errorsZConsulException�baseZchannel_errorsr rq rs rn r r r/ r ro s"
��
�� ro )!r �
__future__r rk rZ � collectionsr �
contextlibr r5 r �timer Zkombu.exceptionsr Z kombu.logr Zkombu.utils.jsonr r
Zkombu.utils.objectsr � r
r r r% rt r$ � Exceptionr r ro r r r r �<module> s.
[