o
    —¨ÊhÃ  ã                   @  sb  d Z ddlmZ ddlZddlZddlmZ ddlmZm	Z	 ddl
mZmZ ddlmZ z]ddlZdd	lmZ dd
lmZ ejjejjejjejjejjejjejjejjejjf	Zejjejj ejj!ejjejjejjejj"ejj#ejjejj$ejj%ejj&ejjejj'ej(fZ)W n e*y˜   dZd ZZ)Y nw dZ+dZ,G dd„ dej-ƒZ-G dd„ dej.ƒZ.dS )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 <mahendra.m@gmail.com>c                      s„   e Zd ZdZdZi Z‡ fdd„Zdd„ Zdd„ Zd	d
„ Z	dd„ Z
dd„ Zdd„ Zdd„ Zdd„ Zdd„ Zdd„ Zedd„ ƒZ‡  ZS )ÚChannelzZookeeper Channel.Nc                   s4   t ƒ j|fi |¤Ž | jjj}d | d¡¡| _d S )Nz/{}ú/)ÚsuperÚ__init__Ú
connectionÚclientÚvirtual_hostÚformatÚstripÚ_vhost)Úselfr   ÚkwargsÚvhost©Ú	__class__r   úK/var/www/html/env/lib/python3.10/site-packages/kombu/transport/zookeeper.pyr   i   s   
zChannel.__init__c                 C  s   t j | j|¡S ©N)ÚosÚpathÚjoinr   )r   Ú
queue_namer   r   r   Ú	_get_pathn   s   zChannel._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   
zChannel._get_queuec                 K  s&   |   |¡jtt|ƒƒ| j|dd�d�S )NT)Úreverse)Úpriority)r'   Úputr   r   Ú_get_message_priority)r   r&   Úmessager   r   r   r   Ú_put}   s   

þzChannel._putc                 C  s,   |   |¡}| ¡ }|d u rtƒ ‚tt|ƒƒS r   )r'   r$   r   r   r   )r   r&   Úmsgr   r   r   Ú_getƒ   s
   
zChannel._getc                 C  s0   d}|   |¡}	 | ¡ }|d u r	 |S |d7 }q)Nr   Tr   )r'   r$   )r   r&   Úcountr.   r   r   r   Ú_purgeŒ   s   
þüzChannel._purgec                 O  s.   |   |¡r|  |¡ | j |  |¡¡ d S d S r   )Ú
_has_queuer1   r   Údeleter"   )r   r&   Úargsr   r   r   r   Ú_delete˜   s   

þzChannel._deletec                 C  s   |   |¡}t|ƒS r   )r'   r%   ©r   r&   r   r   r   Ú_size�   s   
zChannel._sizec                 K  s   |   |¡s|  |¡}d S d S r   )r2   r'   )r   r&   r   r   r   r   Ú
_new_queue¡   s   
ÿzChannel._new_queuec                 C  s   | j  |  |¡¡d uS r   )r   Úexistsr"   r6   r   r   r   r2   ¥   s   zChannel._has_queuec              	   C  sê   | j j}g }|jrO|jD ]B}| d¡r|tdƒd … }|sqz| dd¡\}}|t|ƒf}W n tyH   ||jkrB||j	p?t
f}n|t
f}Y nw | |¡ q|j|j	pUt
f}||vra| d|¡ d dd„ |D ƒ¡}t|ƒ}| ¡  |S )Nzzookeeper://ú:r   r   ú,c                 S  s   g | ]\}}|› d |› �‘qS )r:   r   )Ú.0ÚhÚpr   r   r   Ú
<listcomp>½   s    z!Channel._open.<locals>.<listcomp>)r   r   ÚaltÚ
startswithr%   ÚsplitÚintÚ
ValueErrorÚhostnameÚportÚDEFAULT_PORTÚappendÚinsertr    r
   Ústart)r   ÚconninfoÚhostsÚ	host_portÚhostrF   Úconn_strÚconnr   r   r   Ú_open¨   s2   


€üzChannel._openc                 C  s   | j d u r
|  ¡ | _ | j S r   )Ú_clientrQ   ©r   r   r   r   r   Â   s   

zChannel.client)Ú__name__Ú
__module__Ú__qualname__Ú__doc__rR   r#   r   r"   r'   r-   r/   r1   r5   r7   r8   r2   rQ   Úpropertyr   Ú__classcell__r   r   r   r   r   c   s"    	r   c                      sT   e Zd ZdZeZdZeZej	j
e Z
ej	je ZdZdZ‡ fdd„Zdd„ Z‡  ZS )	Ú	TransportzZookeeper Transport.r   Ú	zookeeperÚkazooc                   s&   t d u rtdƒ‚tƒ j|i |¤Ž d S )Nz"The kazoo library is not installed)r\   ÚImportErrorr   r   )r   r4   r   r   r   r   r   Ø   s   zTransport.__init__c                 C  s   t jS r   )r\   Ú__version__rS   r   r   r   Údriver_versionÞ   s   zTransport.driver_version)rT   rU   rV   rW   r   Úpolling_intervalrG   Údefault_portr	   rZ   Úconnection_errorsÚKZ_CONNECTION_ERRORSÚchannel_errorsÚKZ_CHANNEL_ERRORSÚdriver_typeÚdriver_namer   r_   rY   r   r   r   r   rZ   É   s    
ÿ
ÿrZ   )/rW   Ú
__future__r   r   Úsocketr&   r   Úkombu.utils.encodingr   r   Úkombu.utils.jsonr   r   Ú r	   r\   Úkazoo.clientr
   Úkazoo.recipe.queuer   Ú
exceptionsÚSystemErrorExceptionÚConnectionLossExceptionÚMarshallingErrorExceptionÚUnimplementedExceptionÚOperationTimeoutExceptionÚNoAuthExceptionÚInvalidACLExceptionÚAuthFailedExceptionÚSessionExpiredExceptionrc   ÚRuntimeInconsistencyExceptionÚDataInconsistencyExceptionÚBadArgumentsExceptionÚApiErrorExceptionÚNoNodeExceptionÚNodeExistsExceptionÚ NoChildrenForEphemeralsExceptionÚNotEmptyExceptionÚInvalidCallbackExceptionÚerrorre   r]   rG   Ú
__author__r   rZ   r   r   r   r   Ú<module>   s\   )÷ñþf