File: //usr/local/lib/python3.9/site-packages/kombu/transport/__pycache__/confluentkafka.cpython-39.pyc
a
X>hm/ � @ s d Z ddlmZ ddlmZ ddlmZ ddlmZ ddl m
Z
ddl m Z m
Z
z<ddlZdd lmZmZmZmZ dd
lmZmZ d Zd ZW n ey� dZd ZZY n0 dd lmZ ee�Z d
Z G dd� de�Z G dd� dej �Z G dd� dej �Z G dd� dej!�Z!G dd� dej"�Z"dS )a confluent-kafka transport module for Kombu.
Kafka transport using confluent-kafka library.
**References**
- http://docs.confluent.io/current/clients/confluent-kafka-python
**Limitations**
The confluent-kafka transport does not support PyPy environment.
Features
========
* Type: Virtual
* Supports Direct: Yes
* Supports Topic: Yes
* Supports Fanout: No
* Supports Priority: No
* Supports TTL: No
Connection String
=================
Connection string has the following format:
.. code-block::
confluentkafka://[USER:PASSWORD@]KAFKA_ADDRESS[:PORT]
Transport Options
=================
* ``connection_wait_time_seconds`` - Time in seconds to wait for connection
to succeed. Default ``5``
* ``wait_time_seconds`` - Time in seconds to wait to receive messages.
Default ``5``
* ``security_protocol`` - Protocol used to communicate with broker.
Visit https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md for
an explanation of valid values. Default ``plaintext``
* ``sasl_mechanism`` - SASL mechanism to use for authentication.
Visit https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md for
an explanation of valid values.
* ``num_partitions`` - Number of partitions to create. Default ``1``
* ``replication_factor`` - Replication factor of partitions. Default ``1``
* ``topic_config`` - Topic configuration. Must be a dict whose key-value pairs
correspond with attributes in the
http://kafka.apache.org/documentation.html#topicconfigs.
* ``kafka_common_config`` - Configuration applied to producer, consumer and
admin client. Must be a dict whose key-value pairs correspond with attributes
in the https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md.
* ``kafka_producer_config`` - Producer configuration. Must be a dict whose
key-value pairs correspond with attributes in the
https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md.
* ``kafka_consumer_config`` - Consumer configuration. Must be a dict whose
key-value pairs correspond with attributes in the
https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md.
* ``kafka_admin_config`` - Admin client configuration. Must be a dict whose
key-value pairs correspond with attributes in the
https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md.
� )� annotations)�Empty)�virtual)�cached_property)� str_to_bytes)�dumps�loadsN)�Consumer�KafkaException�Producer�TopicPartition)� AdminClient�NewTopic� )�
get_loggeri�# c @ s e Zd ZdZdZdS )�NoBrokersAvailablez(Kafka broker is not available exception.TN)�__name__�
__module__� __qualname__�__doc__Z retriabler r r �H/usr/local/lib/python3.9/site-packages/kombu/transport/confluentkafka.pyr Z s r c s" e Zd ZdZd� fdd� Z� ZS )�MessagezMessage object.Nc s( |� d�| _t� j|fd|i|�� d S )N�topic�channel)�getr �super�__init__)�self�payloadr �kwargs�� __class__r r r c s zMessage.__init__)N)r r r r r �
__classcell__r r r r r ` s r c @ sP e Zd ZdZi Zdd� Zdd� Zdd� Zdd � Zd
d � Z dd
d�Z
ddd�Z dS )�QoSz Quality of Service guarantees.c C s | j pt| j�| j k S )z�Return true if the channel can be consumed from.
:returns: True, if this QoS object can accept a message.
:rtype: bool
�Zprefetch_count�len�_not_yet_acked�r r r r � can_consumem s zQoS.can_consumec C s | j r| j t| j� S dS d S )N� r$ r' r r r �can_consume_max_estimatev s z QoS.can_consume_max_estimatec C s || j |< d S �N�r& )r �message� delivery_tagr r r �append| s z
QoS.appendc C s
| j | S r+ r, )r r. r r r r s zQoS.getc C s4 || j vrd S | j �|�}| j�|j�}|�� d S r+ )r& �popr �
_get_consumerr �commit)r r. r- �consumerr r r �ack� s
zQoS.ackFc C sb |rT| j �|�}| j�|j�}|�� D ]*}t|j|j�}|�|g�\}|� |� q&n
| �
|� dS )z�Reject a message by delivery tag.
If requeue is True, then the last consumed message is reverted so
it'll be refetched on the next attempt.
If False, that message is consumed and ignored.
N) r& r0 r r1 r �
assignmentr � partition� committed�seekr4 )r r. Zrequeuer- r3 r5 �topic_partition�committed_offsetr r r �reject� s �z
QoS.rejectNc C s d S r+ r )r �stderrr r r �restore_unacked_once� s zQoS.restore_unacked_once)F)N) r r r r r& r( r* r/ r r4 r; r= r r r r r# h s
r# c s� e Zd ZdZeZeZdZdZdZ� fdd�Z dd� Z
dd � Z d
d � Z d d
� Z
dd� Zdd� Zdd� Zdd� Zdd� Zdd� Zedd� �Zed d � �Zed d � �Zed d!� �Zed"d#� �Zed$d%� �Z� fd&d'�Z � Z S )(�ChannelzKafka Channel.� Nc s, t � j|i |�� i | _i | _| �� | _d S r+ )r r �_kafka_consumers�_kafka_producers�_open�_client)r �argsr r r r r � s zChannel.__init__c C s t |��dd�S )z>Need to sanitize the name, celery sometimes pushes in @ signs.�@� )�str�replace)r �queuer r r �sanitize_queue_name� s zChannel.sanitize_queue_namec C sL | � |�}| j�|d�}|du rHti | j�| j�d�p8i ��}|| j|< |S )z9Create/get a producer instance for the given topic/queue.NZkafka_producer_config)rJ rA r r �
common_config�options)r rI �producerr r r �
_get_producer� s
��
zChannel._get_producerc C sf | � |�}| j�|d�}|du rbt|� d�ddd�| j�| j�d�pFi ��}|�|g� || j|< |S )z9Create/get a consumer instance for the given topic/queue.Nz-consumer-groupZearliestF)zgroup.idzauto.offset.resetzenable.auto.commitZkafka_consumer_config)rJ r@ r r rK rL � subscribe)r rI r3 r r r r1 � s
���
zChannel._get_consumerc K s4 | � |�}| �|�}|�|tt|��� |�� dS )z!Put a message on the topic/queue.N)rJ rN Zproducer r �flush)r rI r- r rM r r r �_put� s
z Channel._putc K s~ | � |�}| �|�}d}z|�| j�}W n ty: Y n0 |sFt� �|�� }|rbt�|� t� �i t|� � ��d|�
� i�S )z#Get a message from the topic/queue.Nr ) rJ r1 �poll�wait_time_seconds�
StopIterationr �error�loggerr �valuer )r rI r r3 r- rU r r r �_get� s
z Channel._getc O s6 | � |�}| j| �� | j�|� | j�|g� dS )zDelete a queue/topic.N)rJ r@ �closer0 �clientZ
delete_topics)r rI rD r r r r �_delete� s
zChannel._deletec C sp | � |�}| j�|d�}|du r$dS d}|�� D ]:}t||j�}|�|�\}}|�|g�\}|||j 7 }q0|S )z6Get the number of pending messages in the topic/queue.Nr ) rJ r@ r r5 r r6 Zget_watermark_offsetsr7 �offset) r rI r3 �sizer5 r9 �_�
end_offsetr: r r r �_size� s
z
Channel._sizec K s` | � |�}|| j�� jv r dS t|| j�dd�| j�dd�| j�di �d�}| jj|gd� dS )z(Create a new topic if it does not exist.N�num_partitionsr) �replication_factorZ topic_config)ra rb �config)Z
new_topics)rJ rZ � list_topics�topicsr rL r Z
create_topics)r rI r r r r r �
_new_queue s
�zChannel._new_queuec K s | � |�}|| j�� jv S )z Check if a topic already exists.)rJ rZ rd re )r rI r r r r �
_has_queue s
zChannel._has_queuec
C sd t i | j�| j�d�pi ��}z|j| jd� W n. tjy^ } zt|��W Y d }~n
d }~0 0 |S )NZkafka_admin_config)�timeout) r
rK rL r rd rS �confluent_kafkar
r )r rZ �er r r rB s �� z
Channel._openc C s | j d u r| �� | _ | j S r+ )rC rB r' r r r rZ ' s
zChannel.clientc C s
| j jjS r+ )�
connectionrZ Ztransport_optionsr' r r r rL - s zChannel.optionsc C s | j jS r+ )rk rZ r' r r r �conninfo1 s zChannel.conninfoc C s | j �d| j�S )NrS )rL r �default_wait_time_secondsr' r r r rS 5 s �zChannel.wait_time_secondsc C s | j �d| j�S )N� connection_wait_time_seconds)rL r �$default_connection_wait_time_secondsr' r r r rn ; s �z$Channel.connection_wait_time_secondsc C sz | j j}d|j� dt|j�p t� �i}| j�dd�}|�� dkr`|� ||j
|j | j�d�d�� |� | j�d�pri � |S )Nzbootstrap.servers�:�security_protocol� plaintextZsasl_mechanism)zsecurity.protocolz
sasl.usernamez
sasl.passwordzsasl.mechanismZkafka_common_config) rk rZ �hostname�int�port� DEFAULT_PORTrL r �lower�updateZuserid�password)r rl rc rq r r r rK B s �
�zChannel.common_configc s2 t � �� i | _| j�� D ] }|�� qi | _d S r+ )r rY rA r@ �values)r r3 r r r rY U s
z
Channel.close) r r r r r# r rm ro rC r rJ rN r1 rQ rX r[ r` rf rg rB �propertyrZ rL rl r rS rn rK rY r" r r r r r> � s<
r> c sh e Zd ZdZdddd�dd�ZeZeZdZd Z e
fZ � fd
d �Z d d
� Z
� fdd�Z� fdd�Z� ZS )� TransportzKafka Transport.F�**rG )�uri�returnc C s d S r+ r )r r~ Zinclude_password�maskr r r �as_urib s zTransport.as_uriZkafkaZconfluentkafkac s( t d u rtd��t� j|fi |�� d S )Nz,The confluent-kafka library is not installed)ri � ImportErrorr r )r rZ r r r r r p s zTransport.__init__c C s t jS r+ )ri � __version__r' r r r �driver_versionu s zTransport.driver_versionc s
t � �� S r+ )r �establish_connectionr' r r r r� x s z Transport.establish_connectionc s t � �|�S r+ )r �close_connection)r rk r r r r� { s zTransport.close_connection)Fr} )r r r r r� r> rv � default_portZ driver_typeZ driver_namer Z recoverable_connection_errorsr r� r� r� r" r r r r r| _ s � r| )#r �
__future__r rI r Zkombu.transportr Z kombu.utilsr Zkombu.utils.encodingr Zkombu.utils.jsonr r ri r r
r r Zconfluent_kafka.adminr
r ZKAFKA_CONNECTION_ERRORSZKAFKA_CHANNEL_ERRORSr� Z kombu.logr r rV rv r r r# r> r| r r r r �<module> s0 < 7 A