File: //usr/local/lib/python3.9/site-packages/kombu/transport/__pycache__/base.cpython-39.pyc
a
X>h � @ s d Z ddlmZ ddlZddlZddlmZ ddlmZ ddl m
Z
m Z ddl m
Z
ddlmZ dd lmZ dd
lmZ er�dd lmZ d Zd
efdefdefdefdefd�Zdd� Zdd� Zdd� ZG dd� d�Z G dd � d �Z G d d � d e �Z e d e g d ��d d!�Z!G d"d#� d#�Z"dS )$zBase transport interface.� )� annotationsN)�
TYPE_CHECKING)�RecoverableConnectionError)� ChannelError�ConnectionError)�Message)�
dictfilter)�cached_property)�
maybe_s_to_ms)�
TracebackType)r �
StdChannel�
Management� Transportz x-expiresz
x-message-ttlz x-max-lengthzx-max-length-byteszx-max-priority)�expiresZ message_ttl�
max_lengthZmax_length_bytesZ max_priorityc K s2 t tdd� |�� D ���}|r.t| fi |��S | S )a! Convert queue arguments to RabbitMQ queue arguments.
This is the implementation for Channel.prepare_queue_arguments
for AMQP-based transports. It's used by both the pyamqp and librabbitmq
transports.
Arguments:
arguments (Mapping):
User-supplied arguments (``Queue.queue_arguments``).
Keyword Arguments:
expires (float): Queue expiry time in seconds.
This will be converted to ``x-expires`` in int milliseconds.
message_ttl (float): Message TTL in seconds.
This will be converted to ``x-message-ttl`` in int milliseconds.
max_length (int): Max queue length (in number of messages).
This will be converted to ``x-max-length`` int.
max_length_bytes (int): Max queue size in bytes.
This will be converted to ``x-max-length-bytes`` int.
max_priority (int): Max priority steps for queue.
This will be converted to ``x-max-priority`` int.
Returns
-------
Dict: RabbitMQ compatible queue arguments.
c s s | ]\}}t ||�V qd S �N)�_to_rabbitmq_queue_argument)�.0�key�value� r �>/usr/local/lib/python3.9/site-packages/kombu/transport/base.py� <genexpr>= s �z.to_rabbitmq_queue_arguments.<locals>.<genexpr>)r �dict�items)� arguments�options�preparedr r r �to_rabbitmq_queue_arguments! s
�
r c C s$ t | \}}||d ur ||�n|fS r )�RABBITMQ_QUEUE_ARGUMENTS)r r �opt�typr r r r D s r c C s t d�| j|��S )Nz<Transport {0.__module__}.{0.__name__} does not implement {1})�NotImplementedError�format� __class__)�obj�methodr r r �
_LeftBlankJ s
��r' c @ sX e Zd ZdZdZdd� Zdd� Zdd� Zd d
� Zd d � Z d
d� Z
ddddd�dd�Z dS )r z Standard channel base class.Nc O s" ddl m} || g|�R i |��S )Nr )�Consumer)�kombu.messagingr( )�self�args�kwargsr( r r r r( U s zStdChannel.Consumerc O s" ddl m} || g|�R i |��S )Nr )�Producer)r) r- )r* r+ r, r- r r r r- Y s zStdChannel.Producerc C s t | d��d S �N� get_bindings�r' �r* r r r r/ ] s zStdChannel.get_bindingsc C s dS )z�Callback called after RPC reply received.
Notes
-----
Reply queue semantics: can be used to delete the queue
after transient reply message received.
Nr )r* �queuer r r � after_reply_message_received` s z'StdChannel.after_reply_message_receivedc K s |S r r )r* r r, r r r �prepare_queue_argumentsi s z"StdChannel.prepare_queue_argumentsc C s | S r r r1 r r r � __enter__l s zStdChannel.__enter__ztype[BaseException] | NonezBaseException | NonezTracebackType | None�None)�exc_type�exc_val�exc_tb�returnc C s | � � d S r )�close)r* r7 r8 r9 r r r �__exit__o s zStdChannel.__exit__) �__name__�
__module__� __qualname__�__doc__Zno_ack_consumersr( r- r/ r3 r4 r5 r< r r r r r P s r c @ s e Zd ZdZdd� Zdd� ZdS )r
z!AMQP Management API (incomplete).c C s
|| _ d S r )� transport)r* rA r r r �__init__{ s zManagement.__init__c C s t | d��d S r. r0 r1 r r r r/ ~ s zManagement.get_bindingsN)r= r> r? r@ rB r/ r r r r r
x s r
c @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) �
Implementsz/Helper class used to define transport features.c C s* z
| | W S t y$ t|��Y n0 d S r )�KeyError�AttributeError)r* r r r r � __getattr__� s
zImplements.__getattr__c C s || |< d S r r )r* r r r r r � __setattr__� s zImplements.__setattr__c K s | j | fi |��S r )r$ )r* r, r r r �extend� s zImplements.extendN)r= r> r? r@ rF rG rH r r r r rC � s rC F)�directZtopicZfanout�headers)� asynchronousZ
exchange_type�
heartbeatsc @ s e Zd ZdZeZdZdZdZefZ e
fZ dZ dZ
dZe�� Zdd� Zdd� Zd d
� Zd d � Zd
d� Zdd� Zd3dd�Zdd� Zdd� Zdd� Zdd� Z d d � Z e j e j e!j"e!j#ffd d �Z$d d!� Z%d"d#� Z&d4d%d%d&�d'd(�Z'e(d)d*� �Z)d+d,� Z*e+d-d.� �Z,e(d/d0� �Z-e(d1d2� �Z.dS )5r zBase class for transports.NF�N/Ac K s
|| _ d S r )�client)r* rN r, r r r rB � s zTransport.__init__c C s t | d��d S )N�establish_connectionr0 r1 r r r rO � s z Transport.establish_connectionc C s t | d��d S )N�close_connectionr0 �r* �
connectionr r r rP � s zTransport.close_connectionc C s t | d��d S )N�create_channelr0 rQ r r r rS � s zTransport.create_channelc C s t | d��d S )N�
close_channelr0 rQ r r r rT � s zTransport.close_channelc K s t | d��d S )N� drain_eventsr0 )r* rR r, r r r rU � s zTransport.drain_events� c C s d S r r )r* rR Zrater r r �heartbeat_check� s zTransport.heartbeat_checkc C s dS )NrM r r1 r r r �driver_version� s zTransport.driver_versionc C s dS )Nr r rQ r r r �get_heartbeat_interval� s z Transport.get_heartbeat_intervalc C s d S r r �r* rR �loopr r r �register_with_event_loop� s z"Transport.register_with_event_loopc C s d S r r rZ r r r �unregister_from_event_loop� s z$Transport.unregister_from_event_loopc C s dS �NTr rQ r r r �verify_connection� s zTransport.verify_connectionc s �j �� �����fdd�� � S )Nc
s~ �j std��z�dd� W nP �y0 Y d S �yl } z&|j�v rVW Y d }~d S � W Y d }~n
d }~0 0 | �� | � d S )NzSocket was disconnectedr )�timeout)� connectedr �errno� call_soon)r[ �exc��_read�_unavailrR rU �errorr` r r rf � s
z%Transport._make_reader.<locals>._read)rU )r* rR r` rh rg r re r � _make_reader� s
zTransport._make_readerc C s dS r^ r rQ r r r �qos_semantics_matches_spec� s z$Transport.qos_semantics_matches_specc C s* | j }|d u r | �|� }| _ ||� d S r )�_Transport__readerri )r* rR r[ �readerr r r � on_readable� s zTransport.on_readable�**�str)�urir: c C s
t � �dS )z(Customise the display format of the URI.N)r" )r* rp Zinclude_password�maskr r r �as_uri� s zTransport.as_uric C s i S r r r1 r r r �default_connection_params� s z#Transport.default_connection_paramsc O s
| � | �S r )r
)r* r+ r, r r r � get_manager s zTransport.get_managerc C s | � � S r )rt r1 r r r �manager s zTransport.managerc C s | j jS r )�
implementsrL r1 r r r �supports_heartbeats s z Transport.supports_heartbeatsc C s | j jS r )rv rK r1 r r r � supports_ev
s zTransport.supports_ev)rV )Frn )/r= r> r? r@ r
rN Z
can_parse_url� default_portr Zconnection_errorsr Zchannel_errorsZ driver_typeZ driver_namerk � default_transport_capabilitiesrH rv rB rO rP rS rT rU rW rX rY r\ r] r_ �socketr` rh rb �EAGAINZEINTRri rj rm rr �propertyrs rt r ru rw rx r r r r r � sL
�
r )#r@ �
__future__r rb r{ �typingr Zamqp.exceptionsr Zkombu.exceptionsr r Z
kombu.messager Zkombu.utils.functionalr Zkombu.utils.objectsr Zkombu.utils.timer
�typesr �__all__�intr r r r' r r
r rC � frozensetrz r r r r r �<module> s> � #(
�