Файловый менеджер - Редактировать - /opt/imh-python/lib/python3.9/site-packages/kombu/transport/__pycache__/gcpubsub.cpython-39.pyc
Ðазад
a o�h�k � @ s d Z ddlmZ ddlZddlZddlZddlZddlmZm Z m Z m Z ddl m Z ddlmZ ddlmZ ddlmZ dd lmZmZ dd lmZmZ dd lmZ dd lmZ dd l m Z m Z m Z ddl!m"Z" ddl#m$Z$ ddl%m&Z& ddl'm(Z(m)Z) ddl'm*Z+ ddl,m*Z- ddl.m*Z/ ddl0m1Z2 ddl3m4Z4 ddl5m6Z6 ddl7m8Z8m9Z9 ddl:m;Z;m<Z< ddl=m>Z> ddl?m@Z@ e6d�ZAeBejC�h d � ZDeEd �eEd �id d � eDD ��ZFG d!d"� d"�ZGG d#d$� d$�ZHejIG d%d&� d&��ZJG d'd(� d(e@jK�ZKG d)d*� d*e@jL�ZLdS )+ap GCP Pub/Sub transport module for kombu. More information about GCP Pub/Sub: https://cloud.google.com/pubsub Features ======== * Type: Virtual * Supports Direct: Yes * Supports Topic: No * Supports Fanout: Yes * Supports Priority: No * Supports TTL: No Connection String ================= Connection string has the following formats: .. code-block:: gcpubsub://projects/project-name Transport Options ================= * ``queue_name_prefix``: (str) Prefix for queue names. * ``ack_deadline_seconds``: (int) The maximum time after receiving a message and acknowledging it before pub/sub redelivers the message. * ``expiration_seconds``: (int) Subscriptions without any subscriber activity or changes made to their properties are removed after this period. Examples of subscriber activities include open connections, active pulls, or successful pushes. * ``wait_time_seconds``: (int) The maximum time to wait for new messages. Defaults to 10. * ``retry_timeout_seconds``: (int) The maximum time to wait before retrying. * ``bulk_max_messages``: (int) The maximum number of messages to pull in bulk. Defaults to 32. � )� annotationsN)�FIRST_COMPLETED�Future�ThreadPoolExecutor�wait)�suppress)�getpid)�Empty)�Lock)� monotonic�sleep)� NAMESPACE_OID�uuid3)� gethostname��timeout)� AlreadyExists�DeadlineExceeded�PermissionDenied)�Retry)� monitoring_v3)�query)�PublisherClient�SubscriberClient)� exceptions)� gapic_version)�TRANSIENT_DELIVERY_MODE)� get_logger)� bytes_to_str�safe_str)�dumps�loads)�cached_property� )�virtualzkombu.transport.gcpubsub> �_�.�-r& r' c C s i | ]}t |�t d ��qS )r% )�ord)�.0�c� r+ ��/root/rpmbuild/BUILDROOT/imh-python39-modules-3.9.7-92.el8.x86_64/opt/imh-python/lib/python3.9/site-packages/kombu/transport/gcpubsub.py� <dictcomp>S � r- c @ sP e Zd ZdZdd� Zdd� Zdd�dd �Zdd d �Zd d� Zdd� Z dd� Z dS )� UnackedIdszThreadsafe list of ack_ids.c C s g | _ t� | _d S �N)�_listr �_lock��selfr+ r+ r, �__init__Z s zUnackedIds.__init__c C s | j �|� d S r0 )r1 �append�r4 �valr+ r+ r, r6 ^ s zUnackedIds.append�list)�valsc C s | j �|� d S r0 )r1 �extend)r4 r: r+ r+ r, r; b s zUnackedIds.extend���c C s6 | j � | j�|�W d � S 1 s(0 Y d S r0 )r2 r1 �pop)r4 �indexr+ r+ r, r= f s zUnackedIds.popc C s^ | j �D tt�� | j�|� W d � n1 s20 Y W d � n1 sP0 Y d S r0 )r2 r � ValueErrorr1 �remover7 r+ r+ r, r@ j s zUnackedIds.removec C s4 | j � t| j�W d � S 1 s&0 Y d S r0 )r2 �lenr1 r3 r+ r+ r, �__len__n s zUnackedIds.__len__c C s | j | S r0 )r1 )r4 �itemr+ r+ r, � __getitem__r s zUnackedIds.__getitem__N)r< ) �__name__� __module__� __qualname__�__doc__r5 r6 r; r= r@ rB rD r+ r+ r+ r, r/ W s r/ c @ s6 e Zd ZdZd dd�Zddd�Zddd �Zd d � Zd S )� AtomicCounterzIThreadsafe counter. Returns the value after inc/dec operations. r c C s || _ t� | _d S r0 )�_valuer r2 )r4 �initialr+ r+ r, r5 } s zAtomicCounter.__init__r# c C s> | j �$ | j|7 _| jW d � S 1 s00 Y d S r0 �r2 rJ �r4 �nr+ r+ r, �inc� s zAtomicCounter.incc C s> | j �$ | j|8 _| jW d � S 1 s00 Y d S r0 rL rM r+ r+ r, �dec� s zAtomicCounter.decc C s0 | j � | jW d � S 1 s"0 Y d S r0 rL r3 r+ r+ r, �get� s zAtomicCounter.getN)r )r# )r# )rE rF rG rH r5 rO rP rQ r+ r+ r+ r, rI w s rI c @ sF e Zd ZU dZded< ded< ded< ded< ejed�Zded < d S ) �QueueDescriptorzPub/Sub queue descriptor.�str�name� topic_path�subscription_id�subscription_path)�default_factoryr/ � unacked_idsN) rE rF rG rH �__annotations__� dataclasses�fieldr/ rY r+ r+ r+ r, rR � s rR c s* e Zd ZU dZdZdZdZdZdZdZ dZ dZ e � Z d Zd ed < e�� Ze� Zi Zd ed < e � Zded< � fdd�Zefddd�dd�Zdd� Zdbddddd�dd�Zdd d �d d �Z dcddddddd �d!d"�Z d#d$� Z d%d&� Z d'd(� Z dddd)d*�d+d,�Z!d-d.�d/d0�Z"dd)d*�d1d2�Z#dd3�d4d5�Z$de� fd6d7� Z%ddd8�d9d:�Z&df� fd;d<� Z'd=dd>�d?d@�Z(ddA�dBdC�Z)dDdE� Z*ddA�dFdG�Z+e,dHdI� �Z-e,dJdK� �Z.e,dLdM� �Z/e0dNdO� �Z1e0dPdQ� �Z2e,dRdS� �Z3e,dTdU� �Z4e,dVdW� �Z5e,dXdY� �Z6e,dZd[� �Z7e,d\d]� �Z8� fd^d_�Z9e:d`da� �Z;� Z<S )g�ChannelzGCP Pub/Sub channel.TF� �� i�Q i, � Nzthreading.Thread�_unacked_extenderzdict[str, QueueDescriptor]� _queue_cachezset[str]�_tmp_subscriptionsc sr t � j|i |�� t� | _t�d| jj� t� | jj�| _ | j � � dkrnt j| jdd�t_| j�� tj�� d S )Nznew GCP pub/sub channel: %sr# T)�target�daemon)�superr5 r �pool�logger�info�conninfo�hostname� Transport� parse_uri� project_id� _n_channelsrO � threading�Thread�_extend_unacked_deadliner] ra �_stop_extender�clear�start)r4 �args�kwargs�� __class__r+ r, r5 � s � zChannel.__init__rS )rT �returnc C s( |� | j�s| j| }tt|���|�S )z7Format AMQP queue name into a valid Pub/Sub queue name.)� startswith�queue_name_prefixrS r � translate)r4 rT �tabler+ r+ r, � entity_name� s zChannel.entity_namec C s | � |�j}| �|�}t�d||||� i }|dkr\dd|� d�i}| j�| j|�}| j}nn|dkr�t t t � � dt � � ��� } |� d| � �} | j�| j| �}| j �|� | j�|� d }ntd |� d ���| �| j||�} | j| |||d � t|| ||d �} | | j|< d S )Nz9binding queue: %s to %s exchange: %s with routing_key: %s�direct�filterzattributes.routing_key="�"�fanoutr& r' iX zexchange type z not implemented)rU rW � filter_args� msg_retention)rT rU rV rW )�typeof�typer rh �debug� subscriberrW rn �expiration_secondsr r r r rc �add�_fanout_exchanges�NotImplementedError� _create_topic�_create_subscriptionrR rb ) r4 �exchange� routing_key�pattern�queue� exchange_typer� rW �message_retention_duration�uidZ uniq_sub_nameZexchange_topic�qdescr+ r+ r, � _queue_bind� sZ � ��� ����zChannel._queue_bind�int)rn �topic_idr� rz c C sx | j �||�}| �|�r(t�d|� |S z8t�d|� d|i}|rP|� d�|d< | j j|d� W n tyr Y n0 |S )Nztopic: %s existszcreating topic: %srT �sr� ��request)� publisherrU �_is_topic_existsrh r� Z create_topicr )r4 rn r� r� rU r� r+ r+ r, r� s �� zChannel._create_topic�bool)rU rz c C s8 | j jdd| j� �id�}|D ]}|j|kr dS q dS )N�projectz projects/r� TF)r� Z list_topicsrn rT )r4 rU Ztopics�tr+ r+ r, r� s � zChannel._is_topic_exists)rn r� rU rW r� rz c C s� |p| j �| j|�}|p$| j�||�}zTt�d|||� |p@| j}| j j||| j d| j� d�i|� d�d�|ppi �d� W n t y� Y n0 |S )Nz0creating subscription: %s, topic: %s, filter: %s�ttlr� )rT Ztopic�ack_deadline_secondsZexpiration_policyr� r� ) r� rW rn r� rU rh r� r� Zcreate_subscriptionr� r )r4 rn r� rU rW r� r� r+ r+ r, r� # s8 � �� ���� z Channel._create_subscriptionc O sP | � |�}t�d|� | j�|�}|s*dS | jjd|jid� | j�|d� dS )zDelete a queue by name.zdeleting queue: %sN� subscriptionr� ) r rh ri rb rQ r� �delete_subscriptionrW r= )r4 r� rv rw r� r+ r+ r, �_deleteK s �zChannel._deletec K sV | � |�}| j| }| �|�}t�d||j|� t|�}| jj|j|� d�|d� dS )z Put a message onto the queue.z8putting message to queue: %s, topic: %s, routing_key: %s�utf-8)r� N) r rb �_get_routing_keyrh r� rU r r� �publish�encode)r4 r� �messagerw r� r� �encoded_messager+ r+ r, �_putW s ��z Channel._putc K sV | � ||� | j�| j|�}t�d||� t|�}| jj||�d�t | j d�d� dS )z#Put a message onto fanout exchange.z-putting msg to fanout exchange: %s, topic: %sr� ��deadline)�retryN) �_lookupr� rU rn rh r� r r� r� r �retry_timeout_seconds)r4 r� r� r� rw rU r� r+ r+ r, � _put_fanouti s � �zChannel._put_fanout�float)r� r c C s� | � |�}| j| }z,| jj|jdd�t| jd�|p8| jd�}W n tyX t � �Y n0 t |j �dkrnt � �|j d }|j }t |jj�}|d d }t�d|||d � | �|d �r�t�d |� | �|g|j� n$|||jj|jd �|d < |j�|� |S ) z(Retrieves a single message from a queue.r# �r� � max_messagesr� �r� r� r r � properties� delivery_infoz-queue:%s got message, ack_id: %s, payload: %sz auto acking message ack_id: %s�r� �ack_id� message_idrW �gcpubsub_message)r rb r� �pullrW r r� �wait_time_secondsr r rA �received_messagesr� r! r� �datarh r� � _is_auto_ack�_do_ackr� rY r6 ) r4 r� r r� �responser� r� �payloadr� r+ r+ r, �_gety sD � � � � z Channel._get�dict)�payload_propertiesc C s& |d d }|d }|t kp$|| jv S )Nr� r� � delivery_mode)r r� )r4 r� r� r� r+ r+ r, r� � s �zChannel._is_auto_ackc C s: | � |�}| j| }| �� }|s&t� �z,| jj|j|d�t| jd�|pJ| j d�}W n t yj t� �Y n0 |j }t |�dkr�t� �g }g } t �dt |�|� |D ]n} | j} tt| jj��} | d d } || | jj|jd�| d < | �| d �r�|�| � n |j�| � | �| � q�|�r2t �d |� | �||j� || fS ) z(Retrieves bulk of messages from a queue.r� r� r� r z#batching %d messages from queue: %sr� r� r� r� zauto acking ack_ids: %s)r rb �_get_max_messages_estimater r� r� rW r r� r� r r� rA rh r� r� r! r r� r� r� r� r6 rY r� )r4 r� r Zprefixed_queuer� r� r� r� Z auto_ack_idsZ ret_payloadsr� r� r� r� r+ r+ r, � _get_bulk� sT � � � � zChannel._get_bulk)rz c C s | j �� }| j}|d u r |S |S r0 )�qosZcan_consume_max_estimate�bulk_max_messages)r4 Z max_allowedZmax_if_unlimitedr+ r+ r, r� � s z"Channel._get_max_messages_estimatec sh | j j�|i �}|s$t� �|||�S | �|��| �|�|||�}|rH|S t� d|� | � |||� |gS )Nz3no queues bound to exchange: %s, binding on the fly) �stateZ exchangesrQ rf r� r� �lookupZ get_tablerh r� Z queue_bind)r4 r� r� �defaultZ exchange_info�retrx r+ r, r� � s" ��zChannel._lookup)r� rz c C s� | � |�}|| jvrdS | j| }tj| j| jdtj�� dd�j|j d�}t t ��" t dd� |D ��W d � S 1 sz0 Y d S ) z�Return the number of messages in a queue. This is a *rough* estimation, as Pub/Sub doesn't provide an exact API. r z;pubsub.googleapis.com/subscription/num_undelivered_messagesr# )Zend_time�minutes)rV c s s | ]}|j d jjV qdS )r N)Zpoints�valueZ int64_value)r) �contentr+ r+ r, � <genexpr> s z Channel._size.<locals>.<genexpr>Nr< ) r rb r ZQuery�monitorrn �datetime�nowZselect_resourcesrV r r �sum)r4 r� r� �resultr+ r+ r, �_size� s$ �� �&z Channel._sizec s| |r t d��| j�|�j}|d }|d }|d }t�d||� |d }| �|g|� | j| }|j� |� t � � |� dS )zAcknowledge one message.z multiple acks not implementedr� r� r� z!ack message. queue: %s ack_id: %srW N) r� r� rQ r� rh r� r� rb rY r@ rf � basic_ack) r4 Z delivery_tagZmultipler� Zpubsub_messager� r� rW r� rx r+ r, r� s zChannel.basic_ackz list[str])�ack_idsrW c C s" | j j||d�t| jd�d� d S )N)r� r� r� )r� r� )r� Z acknowledger r� )r4 r� rW r+ r+ r, r� # s �zChannel._do_ack)r� c C sH | � |�}| j�|�}|s dS | �|�}| jj|jtj�� d�d� |S )z'Delete all current messages in a queue.N)r� �timer� ) r rb rQ r� r� �seekrW r� r� )r4 r� r� rN r+ r+ r, �_purge) s ��zChannel._purgec C s� t �� }t�d|� | jd }t|| jd �}| j�|�s�| j � � D ]d}t |j �dkrht� d||j� qDt� d||jt |j �t|j �� | jj|jt|j �| jd�d� qDq.t�d |� d S ) Nz/unacked deadline extension thread: [%s] started� � r z'thread [%s]: no unacked messages for %sz5thread [%s]: extend ack deadline for %s: %d msgs [%s])r� r� r� r� z.unacked deadline extension thread [%s] stopped)rp � get_native_idrh ri �_min_ack_deadline�maxr� rs r rb �valuesrA rY r� rW r9 r� Zmodify_ack_deadline)r4 Z thread_idZmin_deadline_sleepZ sleep_timer� r+ r+ r, rr 9 s@ � ���� �z Channel._extend_unacked_deadlinec C s8 | � |�}| j�| j|�}t�d||� | j�|� d S )Nz0after_reply_message_received: queue: %s, sub: %s)r r� rW rn rh r� rc r� )r4 r� �subr+ r+ r, � after_reply_message_received\ s �z$Channel.after_reply_message_receivedc C s t � S r0 )r r3 r+ r+ r, r� d s zChannel.subscriberc C s t � S r0 )r r3 r+ r+ r, r� h s zChannel.publisherc C s t �� S r0 )r ZMetricServiceClientr3 r+ r+ r, r� l s zChannel.monitorc C s | j jS r0 )� connection�clientr3 r+ r+ r, rj p s zChannel.conninfoc C s | j jjS r0 )r� r� �transport_optionsr3 r+ r+ r, r� t s zChannel.transport_optionsc C s | j �d| j�S )Nr� )r� rQ �default_wait_time_secondsr3 r+ r+ r, r� x s �zChannel.wait_time_secondsc C s | j �d| j�S )Nr� )r� rQ � default_retry_timeout_secondsr3 r+ r+ r, r� ~ s �z Channel.retry_timeout_secondsc C s | j �d| j�S )Nr� )r� rQ � default_ack_deadline_secondsr3 r+ r+ r, r� � s �z Channel.ack_deadline_secondsc C s | j �dd�S )Nr| zkombu-)r� rQ r3 r+ r+ r, r| � s zChannel.queue_name_prefixc C s | j �d| j�S )Nr� )r� rQ �default_expiration_secondsr3 r+ r+ r, r� � s �zChannel.expiration_secondsc C s | j �d| j�S )Nr� )r� rQ �default_bulk_max_messagesr3 r+ r+ r, r� � s �zChannel.bulk_max_messagesc s� t �d� | jrb| j�� }tt��. t �d|� | jjd|id� W d � q 1 sV0 Y q | j� � s�| j � � t j �� t� �� dS )zClose the channel.zclosing channelzdeleting subscription: %sr� r� N)rh r� rc r= r � Exceptionr� r� ro rP rs �setr] ra �joinrf �close)r4 r� rx r+ r, r� � s �& z Channel.closec C s | d � di �� dd�}|S )Nr� r� r� � )rQ )r� r� r+ r+ r, r� � s ���zChannel._get_routing_key)N)NNNNNN)N)N)F)=rE rF rG rH Zsupports_fanoutZ do_restorer� r� r� r� r� r� r� r� ra rZ rp �Eventrs rI ro rb rc r5 �CHARS_REPLACE_TABLEr r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� rr r� r" r� r� r� �propertyrj r� r� r� r� r| r� r� r� � staticmethodr� � __classcell__r+ r+ rx r, r] � s� C � �( +3 # r] c s� e Zd ZdZeZdZdZejj e j f Z ejj e je je je j e jf ejf Z dZdZejjjeddg�d�Z� fd d �Zd d � Zed d d�dd��Z e d d d d�dd��Z d dd�Z dd� Z dd� Z!d d � Z"� Z#S ) rl zGCP Pub/Sub transport.Tg�������?ZgcpubsubZ pubsub_v1r� r� )r� c s( t � j|fi |�� t� | _t� | _d S r0 )rf r5 r �_poolr� �_get_bulk_future_to_queue)r4 r� rw rx r+ r, r5 � s zTransport.__init__c C s t jS r0 )�package_version� __version__r3 r+ r+ r, �driver_version� s zTransport.driver_versionrS )�urirz c C s | � d�d }|�d�S )Nzgcpubsub://projects/r# �/)�split�strip)r r� r+ r+ r, rm � s zTransport.parse_uriF�**c C s |pdS )Nz gcpubsub://r+ )r4 r Zinclude_password�maskr+ r+ r, �as_uri� s zTransport.as_uriNc C sp t � }| j}|r |r ||kr |}z| j|d� W ql tyf |rVt � | |krVt� �|rbt|� Y q 0 qlq d S )Nr )r �polling_interval�_drain_from_active_queuesr �socket_timeoutr )r4 r� r Z time_startr r+ r+ r, � drain_events� s zTransport.drain_eventsc C s� | � � | jdd� t| j|td�\}}dd� |D �}||8 }|D ]}| j�|d � qB|s`t� �t�dt |�� |D ]X}|� � \}}|D ]4}t�d|� || j vr�t� d|� q�| � ||� q�| j�|d � qtd S ) Nr^ r )r Z return_whenc S s h | ]}|� � r|�qS r+ )� exception�r) �fr+ r+ r, � <setcomp> r. z6Transport._drain_from_active_queues.<locals>.<setcomp>zgot %d done get_bulk tasksz consuming message from queue: %sz&Message for queue %s without consumers)�_rm_empty_bulk_requests�_submit_get_bulk_requestsr r� r r= r rh r� rA r� Z _callbacks�warningZ_deliver) r4 r �doner% �emptyr r� Zpayloadsr� r+ r+ r, r � s2 � �z#Transport._drain_from_active_queuesc C s, dd� | j D �}|D ]}| j �|d � qd S )Nc S s h | ]}|� � r|�� r|�qS r+ )r r r r+ r+ r, r s �z4Transport._rm_empty_bulk_requests.<locals>.<setcomp>)r� r= )r4 r r r+ r+ r, r s �z!Transport._rm_empty_bulk_requestsc C sP t | j�� �}| jD ]6}|jD ]*}||v r,q | j�|j||�}|| j|< q qd S r0 )r� r� r� ZchannelsZ_active_queuesr� Zsubmitr� )r4 r Z queues_with_submitted_get_bulkZchannelr� �futurer+ r+ r, r s � z#Transport._submit_get_bulk_requests)Fr )N)$rE rF rG rH r] Z can_parse_urlr r$ rl Zconnection_errors�pubsub_exceptions� TimeoutErrorZchannel_errors�publisher_exceptionsZFlowControlLimitErrorZMessageTooLargeErrorZ PublishErrorZ#PublishToPausedOrderingKeyException�subscriber_exceptionsZAcknowledgeErrorZ driver_typeZ driver_nameZ implementsr; � frozensetr5 r r� rm � classmethodr r r r r r� r+ r+ rx r, rl � sB ����� � # rl )MrH � __future__r r[ r� �stringrp �concurrent.futuresr r r r � contextlibr �osr r� r r r� r r �uuidr r �_socketr r r Zgoogle.api_core.exceptionsr r r Zgoogle.api_core.retryr Z google.cloudr Zgoogle.cloud.monitoring_v3r Zgoogle.cloud.pubsub_v1r r r r Z google.cloud.pubsub_v1.publisherr Z!google.cloud.pubsub_v1.subscriberr Zgoogle.pubsub_v1r r� Z kombu.entityr Z kombu.logr Zkombu.utils.encodingr r Zkombu.utils.jsonr r! Zkombu.utils.objectsr" r� r$ rh r� � punctuationZPUNCTUATIONS_TO_REPLACEr( r� r/ rI � dataclassrR r] rl r+ r+ r+ r, �<module> sZ ' � �
| ver. 1.1 | |
.
| PHP 8.3.30 | Ð“ÐµÐ½ÐµÑ€Ð°Ñ†Ð¸Ñ Ñтраницы: 0 |
proxy
|
phpinfo
|
ÐаÑтройка