File: //usr/local/lib/python3.9/site-packages/kombu/transport/__pycache__/zookeeper.cpython-39.pyc
a
X>h� � @ sd d Z ddlmZ ddlZddlZddlmZ ddlmZm Z ddl
m Z m Z ddl
mZ z�ddlZdd lmZ dd
lmZ ejjejjejjejjejjejjejjejj ejj f Z ejj ejj ejj!ejjejjejjejj"ejj#ejjejj$ejj%ejj&ejj ejj'ej(fZ)W n e*�y2 dZd Z Z)Y n0 d Z+d
Z,G dd� dej-�Z-G dd� dej.�Z.dS )a� Zookeeper transport module for kombu.
Zookeeper based transport. This transport uses the built-in kazoo Zookeeper
based queue implementation.
**References**
- https://zookeeper.apache.org/doc/current/recipes.html#sc_recipes_Queues
- https://kazoo.readthedocs.io/en/latest/api/recipe/queue.html
**Limitations**
This queue does not offer reliable consumption. An entry is removed from
the queue prior to being processed. So if an error occurs, the consumer
has to re-queue the item or it will be lost.
Features
========
* Type: Virtual
* Supports Direct: Yes
* Supports Topic: Yes
* Supports Fanout: No
* Supports Priority: Yes
* Supports TTL: No
Connection String
=================
Connects to a zookeeper node as:
.. code-block::
zookeeper://SERVER:PORT/VHOST
The <vhost> becomes the base for all the other znodes. So we can use
it like a vhost.
Transport Options
=================
� )� annotationsN)�Empty)� bytes_to_str� ensure_bytes)�dumps�loads� )�virtual)� KazooClient)�Queue� i� z!Mahendra M <[email protected]>c s� e Zd ZdZdZi Z� fdd�Zdd� Zdd� Zd d
� Z d d � Z
d
d� Z dd� Z dd� Z
dd� Zdd� Zdd� Zedd� �Z� ZS )�ChannelzZookeeper Channel.Nc s4 t � j|fi |�� | jjj}d�|�d��| _d S )Nz/{}�/)�super�__init__�
connection�clientZ virtual_host�format�strip�_vhost)�selfr �kwargsZvhost�� __class__r �C/usr/local/lib/python3.9/site-packages/kombu/transport/zookeeper.pyr i s
zChannel.__init__c C s t j�| j|�S �N)�os�path�joinr )r �
queue_namer r r � _get_pathn s zChannel._get_pathc C s> | j �|d �}|d u r:t| j| �|��}|| j |< t|� |S r )�_queues�getr r r �len)r r �queuer r r �
_get_queueq s
zChannel._get_queuec K s& | � |�jtt|��| j|dd�d�S )NT)�reverse)�priority)r% �putr r Z_get_message_priority)r r$ �messager r r r �_put} s
�z Channel._putc C s, | � |�}|�� }|d u r t� �tt|��S r )r% r" r r r )r r$ �msgr r r �_get� s
z Channel._getc C s. d}| � |�}|�� }|d u r q*|d7 }q|S )Nr r )r% r" )r r$ �countr+ r r r �_purge� s
zChannel._purgec O s* | � |�r&| �|� | j�| �|�� d S r )�
_has_queuer. r �deleter )r r$ �argsr r r r �_delete� s
zChannel._deletec C s | � |�}t|�S r )r% r# �r r$ r r r �_size� s
z
Channel._sizec K s | � |�s| �|�}d S r )r/ r% )r r$ r r r r �
_new_queue� s
zChannel._new_queuec C s | j �| �|��d uS r )r �existsr r3 r r r r/ � s zChannel._has_queuec C s� | j j}g }|jr�|jD ]�}|�d�r6|td�d � }|s<qz |�dd�\}}|t|�f}W n4 ty� ||jkr�||j p~t
f}n|t
f}Y n0 |� |� q|j|j p�t
f}||vr�|� d|� d�
dd� |D ��}t|�}|�� |S )Nz zookeeper://�:r r �,c S s g | ]\}}|� d |� ��qS )r7 r )�.0�h�pr r r �
<listcomp>� � z!Channel._open.<locals>.<listcomp>)r r Zalt�
startswithr# �split�int�
ValueError�hostname�port� DEFAULT_PORT�append�insertr r
�start)r Zconninfo�hosts� host_port�hostrC Zconn_str�connr r r �_open� s.
z
Channel._openc C s | j d u r| �� | _ | j S r )�_clientrL �r r r r r � s
zChannel.client)�__name__�
__module__� __qualname__�__doc__rM r! r r r% r* r, r. r2 r4 r5 r/ rL �propertyr �
__classcell__r r r r r
c s r
c sT e Zd ZdZeZdZeZej j
e Z
ej j e
Z dZdZ� fdd�Zdd� Z� ZS ) � TransportzZookeeper Transport.r Z zookeeper�kazooc s& t d u rtd��t� j|i |�� d S )Nz"The kazoo library is not installed)rV � ImportErrorr r )r r1 r r r r r � s zTransport.__init__c C s t jS r )rV � __version__rN r r r �driver_version� s zTransport.driver_version)rO rP rQ rR r
Zpolling_intervalrD � default_portr rU Zconnection_errors�KZ_CONNECTION_ERRORSZchannel_errors�KZ_CHANNEL_ERRORSZ driver_typeZ driver_namer rY rT r r r r rU � s
�
� rU )/rR �
__future__r r �socketr$ r Zkombu.utils.encodingr r Zkombu.utils.jsonr r � r rV Z kazoo.clientr
Zkazoo.recipe.queuer �
exceptionsZSystemErrorExceptionZConnectionLossExceptionZMarshallingErrorExceptionZUnimplementedExceptionZOperationTimeoutExceptionZNoAuthExceptionZInvalidACLExceptionZAuthFailedExceptionZSessionExpiredExceptionr[ Z RuntimeInconsistencyExceptionZDataInconsistencyExceptionZBadArgumentsExceptionZApiErrorExceptionZNoNodeExceptionZNodeExistsExceptionZ NoChildrenForEphemeralsExceptionZNotEmptyExceptionZInvalidCallbackException�errorr\ rW rD �
__author__r
rU r r r r �<module> sX ) �
�f