File: //usr/local/lib/python3.9/site-packages/kombu/transport/__pycache__/gcpubsub.cpython-39.pyc
a
X>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+ �B/usr/local/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: %sZttlr� )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# )�end_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_tag�multipler� 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 � 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 � 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� �
_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� �submitr� )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 Z dataclassrR r] rl r+ r+ r+ r, �<module> sZ ' � �