File: //usr/local/lib/python3.9/site-packages/kombu/transport/__pycache__/pyamqp.cpython-39.pyc
a
X>hG � @ s� d Z ddlmZ ddlZddlmZ ddlmZ ddlm Z ddl m
Z
d Z d
Z G d d � d e j
�Z
G d
d� deje j�ZG dd� dej�ZG dd� de j�ZG dd� de�ZdS )a� pyamqp transport module for Kombu.
Pure-Python amqp transport using py-amqp library.
Features
========
* Type: Native
* Supports Direct: Yes
* Supports Topic: Yes
* Supports Fanout: Yes
* Supports Priority: Yes
* Supports TTL: Yes
Connection String
=================
Connection string can have the following formats:
.. code-block::
amqp://[USER:PASSWORD@]BROKER_ADDRESS[:PORT][/VIRTUALHOST]
[USER:PASSWORD@]BROKER_ADDRESS[:PORT][/VIRTUALHOST]
amqp://
For TLS encryption use:
.. code-block::
amqps://[USER:PASSWORD@]BROKER_ADDRESS[:PORT][/VIRTUALHOST]
Transport Options
=================
Transport Options are passed to constructor of underlying py-amqp
:class:`~kombu.connection.Connection` class.
Using TLS
=========
Transport over TLS can be enabled by ``ssl`` parameter of
:class:`~kombu.Connection` class. By setting ``ssl=True``, TLS transport is
used::
conn = Connect('amqp://', ssl=True)
This is equivalent to ``amqps://`` transport URI::
conn = Connect('amqps://')
For adding additional parameters to underlying TLS, ``ssl`` parameter should
be set with dict instead of True::
conn = Connect('amqp://broker.example.com', ssl={
'keyfile': '/path/to/keyfile'
'certfile': '/path/to/certfile',
'ca_certs': '/path/to/ca_certfile'
}
)
All parameters are passed to ``ssl`` parameter of
:class:`amqp.connection.Connection` class.
SSL option ``server_hostname`` can be set to ``None`` which is causing using
hostname from broker URL. This is useful when failover is used to fill
``server_hostname`` with currently used broker::
conn = Connect('amqp://broker1.example.com;broker2.example.com', ssl={
'server_hostname': None
}
)
� )� annotationsN)� get_manager)�version_string_as_tuple� )�base��to_rabbitmq_queue_argumentsi( i' c s" e Zd ZdZd� fdd� Z� ZS )�Messagez
AMQP Message.Nc sL |j }t� jf |j||j|�d�|�d�|j|j |�d�p<i d�|�� d S )N� content_type�content_encoding�application_headers)�body�channel� delivery_tagr
r �
delivery_info�
properties�headers)r �super�__init__r
r �getr )�self�msgr �kwargs�props�� __class__� �@/usr/local/lib/python3.9/site-packages/kombu/transport/pyamqp.pyr X s � �zMessage.__init__)N��__name__�
__module__� __qualname__�__doc__r �
__classcell__r r r r r U s r c @ s< e Zd ZdZeZdddddejfdd�Zdd� Zdd� ZdS ) �Channelz
AMQP Channel.Nc C s ||f||||d�|pi ��S )z<Prepare message so that it can be sent using this transport.)�priorityr
r r r )r r
r% r
r r r Z_Messager r r �prepare_messagek s ���zChannel.prepare_messagec K s t |fi |��S �Nr )r � argumentsr r r r �prepare_queue_argumentsx s z Channel.prepare_queue_argumentsc C s | j || d�S )z4Convert encoded message body back to a Python value.�r )r )r Z raw_messager r r �message_to_python{ s zChannel.message_to_python) r r r! r" r �amqpr& r) r+ r r r r r$ f s �
r$ c @ s e Zd ZdZeZdS )�
ConnectionzAMQP Connection.N)r r r! r" r$ r r r r r- � s r- c @ s� e Zd ZdZeZeZeZe jj
Z
e jj Z e jj Z e jj
Z
dZdZejjjddd�Zd$dd�Zd d
� Zd d � Zd
d� Zdd� Zdd� Zdd� Zdd� Zdd� Z dd� Z d%d d �Z d d � Z e d d!� �Z!d"d#� Z"dS )&� TransportzAMQP Transport.zpy-amqpr, T)Z asynchronousZ
heartbeatsNc K s" || _ |p| j| _|p| j| _d S r' )�client� default_port�default_ssl_port)r r/ r0 r1 r r r r r � s zTransport.__init__c C s t jS r' )r, � __version__�r r r r �driver_version� s zTransport.driver_versionc C s |� � S r' r* �r �
connectionr r r �create_channel� s zTransport.create_channelc K s |j f i |��S r' )� drain_events)r r6 r r r r r8 � s zTransport.drain_eventsc C s |d ur|� � d S r' )Zcollectr5 r r r �_collect� s zTransport._collectc C s� | j }| j�� D ] \}}t||d�st|||� q|jdkrBd|_t|jt�rrd|jv rr|jd du rr|j|jd< t|j |j
|j |j |j
|j|j|j|jd� fi |jp�i ��}| jf i |��}| j |_ |�� |S )z(Establish connection to the AMQP broker.N� localhostz 127.0.0.1�server_hostname) �host�userid�password� login_method� virtual_host�insist�ssl�connect_timeout� heartbeat)r/ �default_connection_params�items�getattr�setattr�hostname�
isinstancerB �dictr< r= r> r? r@ rA rC rD Ztransport_optionsr- �connect)r Zconninfo�name�
default_value�opts�connr r r �establish_connection� s:
� � �
� z Transport.establish_connectionc C s |j S r' )� connectedr5 r r r �verify_connection� s zTransport.verify_connectionc C s d|_ |�� dS )z!Close the AMQP broker connection.N)r/ �closer5 r r r �close_connection� s zTransport.close_connectionc C s |j S r' )rD r5 r r r �get_heartbeat_interval� s z Transport.get_heartbeat_intervalc C s d|j _|�|j| j||� d S �NT)� transportZraise_on_initial_eintr�
add_reader�sockZ on_readable)r r6 �loopr r r �register_with_event_loop� s z"Transport.register_with_event_loop� c C s |j |d�S )N)�rate)Zheartbeat_tick)r r6 r^ r r r �heartbeat_check� s zTransport.heartbeat_checkc C s( |j }|�d�dkr$t|d �dk S dS )N�productZRabbitMQ�version)� rb T)Zserver_propertiesr r )r r6 r r r r �qos_semantics_matches_spec� s z$Transport.qos_semantics_matches_specc C s dd| j jr| jn| jddd�S )NZguestr: ZPLAIN)r= r> �portrI r? )r/ rB r1 r0 r3 r r r rE � s �z#Transport.default_connection_paramsc O s t | jg|�R i |��S r' )r r/ �r �argsr r r r r � s zTransport.get_manager)NN)r] )#r r r! r" r- � DEFAULT_PORTr0 �DEFAULT_SSL_PORTr1 r, Zconnection_errorsZchannel_errorsZ recoverable_connection_errorsZrecoverable_channel_errorsZ driver_nameZ driver_typer r. Z
implements�extendr r4 r7 r8 r9 rQ rS rU rV r\ r_ rc �propertyrE r r r r r r. � s> �� �
r. c s e Zd ZdZ� fdd�Z� ZS )� SSLTransportzAMQP SSL Transport.c s& t � j|i |�� | jjs"d| j_d S rW )r r r/ rB re r r r r � s zSSLTransport.__init__r r r r r rk � s rk )r" �
__future__r r, Zkombu.utils.amq_managerr Zkombu.utils.textr � r r rg rh r r$ Z
StdChannelr- r. rk r r r r �<module> s F o