Файловый менеджер - Редактировать - /opt/imh-python/lib/python3.9/site-packages/kombu/transport/__pycache__/confluentkafka.cpython-39.pyc
Ðазад
a o�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 ��/root/rpmbuild/BUILDROOT/imh-python39-modules-3.9.7-92.el8.x86_64/opt/imh-python/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_protocolZ 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> ru � 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 ru r r r# r> r{ r r r r �<module> s0 < 7 A
| ver. 1.1 | |
.
| PHP 8.3.30 | Ð“ÐµÐ½ÐµÑ€Ð°Ñ†Ð¸Ñ Ñтраницы: 0 |
proxy
|
phpinfo
|
ÐаÑтройка