File: //home/arjun/projects/env/lib/python3.10/site-packages/amqp/__pycache__/connection.cpython-310.pyc
o
%we�k � @ s> d Z ddlZddlZddlZddlZddlmZ ddlmZ ddlm Z ddl
m Z m Z m
Z
ddlmZ dd lmZ dd
lmZmZmZmZmZmZmZmZmZ dd l m Z m Z dd l m Z zdd
l!m"Z" W n e#yw G dd� de$�Z"Y nw dZ%d�&� Z'dZ(e�)d�Z*e�)d�Z+de d�Z,dddd�Z-G dd� de�Z.dS )zAMQP Connections.� N)�array)� monotonic)�ensure_promise� )� __version__�sasl�spec)�AbstractChannel)�Channel) �AMQPDeprecationWarning� ChannelError�ConnectionError�ConnectionForced�
MessageNacked�RecoverableChannelError�RecoverableConnectionError�
ResourceError�error_for_code)�
frame_handler� frame_writer�� Transport)�SSLErrorc @ s e Zd ZdS )r N)�__name__�
__module__� __qualname__� r r �H/home/arjun/projects/env/lib/python3.10/site-packages/amqp/connection.pyr s r z�The .{attr} attribute on the connection was accessed before
the connection was established. This is supported for now, but will
be deprecated in amqp 2.2.0.
Since amqp 2.0 you have to explicitly call Connection.connect()
before using the connection.
zP
Start from server, version: %d.%d, properties: %s, mechanisms: %s, locales: %s
)�
Connection�amqpz)amqp.connection.Connection.heartbeat_tickzpy-amqp)�product�product_versionT)�consumer_cancel_notifyzconnection.blocked� authentication_failure_closec s� e Zd ZdZeZeZeZdZ dZ
dZ dZ dZ
dZdZdZdZe�ejjd�e�ejj�e�ejjd�e�ejjd�e�ejjd�e�ejj�e�ejj�e�ejj �hZ dd � e D �Z ejjejj fZ e e j!e"e#fZ$e%fZ&e'e(e j!e"e#fZ)e*fZ+d
d d ddd d
ddddddddddddddde,e-f� fdd� Z.dd� Z/dd� Z0dd� Z1d`dd�Z2dd� Z3d`d d �Z4d d � Z5e6d d!� �Z7e7j8d"d!� �Z7e6d#d$� �Z9e9j8d%d$� �Z9e6d&d'� �Z-e-j8d(d'� �Z- )dad*d+�Z:d,d-� Z;dbd.d/�Z<dcd1d2�Z=d3d4� Z> ddd5d6�Z?e6d7d8� �Z@d9d:� ZAd;d<� ZBd=d>� ZCded?d@�ZDdAdB� ZEd`dCdD�ZFd`dEdF�ZGdGdH� ZH J dfdKdL�ZIdMdN� ZJdOdP� ZKdQdR� ZLdSdT� ZMdUdV� ZNdWdX� ZOdgdZd[�ZPe6d\d]� �ZQe6d^d_� �ZR� ZSS )hr a� AMQP Connection.
The connection class provides methods for a client to establish a
network connection to a server, and for both peers to operate the
connection thereafter.
GRAMMAR::
connection = open-connection *use-connection close-connection
open-connection = C:protocol-header
S:START C:START-OK
*challenge
S:TUNE C:TUNE-OK
C:OPEN S:OPEN-OK
challenge = S:SECURE C:SECURE-OK
use-connection = *channel
close-connection = C:CLOSE S:CLOSE-OK
/ S:CLOSE C:CLOSE-OK
Create a connection to the specified host, which should be
a 'host[:port]', such as 'localhost', or '1.2.3.4:5672'
(defaults to 'localhost', if a port is not specified then
5672 is used)
Authentication can be controlled by passing one or more
`amqp.sasl.SASL` instances as the `authentication` parameter, or
setting the `login_method` string to one of the supported methods:
'GSSAPI', 'EXTERNAL', 'AMQPLAIN', or 'PLAIN'.
Otherwise authentication will be performed using any supported method
preferred by the server. Userid and passwords apply to AMQPLAIN and
PLAIN authentication, whereas on GSSAPI only userid will be used as the
client name. For EXTERNAL authentication both userid and password are
ignored.
The 'ssl' parameter may be simply True/False, or
a dictionary of options to pass to :class:`ssl.SSLContext` such as
requiring certain certificates. For details, refer ``ssl`` parameter of
:class:`~amqp.transport.SSLTransport`.
The "socket_settings" parameter is a dictionary defining tcp
settings which will be applied as socket options.
When "confirm_publish" is set to True, the channel is put to
confirm mode. In this mode, each published message is
confirmed using Publisher confirms RabbitMQ extension.
Nr �ooFSS�s�BlB�BsBBc C s i | ]}|j |�qS r )�
method_sig��.0�mr r r �
<dictcomp>� s zConnection.<dictcomp>zlocalhost:5672�guestr �/�en_USFc s� t �� j| _| p d} |
p
d}
|r t|tj�r|f}|| _np|d urw|dkr+t�|�}nG|dkr4t� � }n>|dkrK|d u s@|d u rDt
d��t� ||�}n'|dkrb|d u sW|d u r[t
d��t� ||�}n|d urmt�
||�}nt
d|��|f| _ntj|d d
�t� � t� ||�t� ||�f| _t| jfi | p�i ��| _|| _|| _|| _t|�| _|| _|| _d | _i | _t� �| d � d | _ d | _ d | _ | | _ |
| _ || _!|| _"|
| _#|| _$|| _%|| _&|| _'|| _(t|�| _)t*d
�| _+d | _,d | _-i | _.g | _/g | _0| | _1d S )Ni�� i �GSSAPI�EXTERNAL�AMQPLAINz-Must supply authentication or userid/password�PLAINzInvalid login methodT)� fail_softFr �H)2�uuid�uuid4�hex�_connection_id�
isinstancer �SASL�authenticationr0 r1 �
ValueErrorr2 r3 �RAW�dict�library_properties�client_properties�locale�host� virtual_hostr �
on_tune_ok�frame_handler_cls�frame_writer_cls�_handshake_complete�channels�super�__init__�
_frame_writer�_on_inbound_frame�
_transport� channel_max� frame_max�client_heartbeat�confirm_publish�ssl� read_timeout�
write_timeout�socket_settings�
on_blocked� on_unblocked�on_openr �_used_channel_ids�
version_major�
version_minor�server_properties�
mechanisms�locales�connect_timeout)�selfrC �userid�password� login_method�login_responser< rD rB rA rS r` rO rP � heartbeatrY rW rX rR rE rT rU rV r r �kwargs�auth�� __class__r r rK � s�
��
���
zConnection.__init__c C sR | j rd| j� d| j� d| j � dt| �d�d� S d| j� d| j� dt| �d�d�S )Nz<AMQP Connection: r. z using z at z#x�>z (disconnected) at )rN rC rD �id�ra r r r �__repr__ s ���zConnection.__repr__c C s | � � | S �N)�connectrm r r r � __enter__" s zConnection.__enter__c G s | � � d S ro )�close)ra �eargsr r r �__exit__& s zConnection.__exit__c C s | j �||�S ro )rY �then)ra �
on_success�on_errorr r r ru ) s zConnection.thenc C s` | j �tjj| jtjj| jtjj| j tjj
| j tjj | j
tjj| jtjj| jtjj| ji� d S ro )�
_callbacks�updater r �Start� _on_start�OpenOk� _on_open_ok�Secure�
_on_secure�Tune�_on_tune�Close� _on_close�Blocked� _on_blocked� Unblocked�
_on_unblocked�CloseOk� _on_close_okrm r r r �_setup_listeners, s
�zConnection._setup_listenersc C s� | j r
|r|� S d S z:| j| j| j| j| j| j| jd�| _| j� � | �
| | j �| _ | �
| | j�| _| jsB| j| jd� | jr5W d S W d S ttfyR | �� � w )N)rV )�timeout)� connectedr rC r` rS rT rU rV � transportrp rF �on_inbound_method�on_inbound_framerG r rH � drain_events�OSErrorr �collect)ra �callbackr r r rp 8 s( �
���zConnection.connectc C s t �ttj|d��� d S )N)�attr)�warnings�warnr �W_FORCE_CONNECT�format)ra r� r r r �_warn_force_connectP s
�z Connection._warn_force_connectc C �" | j d u r| �d� | �� | j S )Nr� )rN r� rp rm r r r r� T �
zConnection.transportc C �
|| _ d S ro )rN )ra r� r r r r� [ �
c C r� )Nr� )rM r� rp rm r r r r� _ r� zConnection.on_inbound_framec C r� ro )rM )ra r� r r r r� f r� c C r� )Nr )rL r� rp rm r r r r j r� zConnection.frame_writerc C r� ro )rL )ra r r r r r q r� �FsSsc s, | j }|| _|| _|| _t|t�r|�d�}|�d�| _|�d�| _ t
� t | j| j| j| j| j � |�
d�p7i � |�di �}|�� fdd�| j�� D �� |sU|�dd � | jD ]} | j| jv rk| �| �}
|
turk nqXtd�d�d d
� | jD ���� d�| j��� ���| �t j j ||| j|
| j f� d S ) Nzutf-8� � � capabilitiesc s i | ] \}}� � |�r||�qS r )�get)r* �
wanted_cap�
enable_cap��scapr r r, � s ��z(Connection._on_start.<locals>.<dictcomp>zGCouldn't find appropriate auth mechanism (can offer: {}; available: {})s , c s s � | ] }|j r|j V qd S ro )� mechanismr) r r r � <genexpr>� s
� �z'Connection._on_start.<locals>.<genexpr>) rA r[ r\ r] r: �str�encode�splitr^ r_ � AMQP_LOGGER�debug�START_DEBUG_FMTr� �
setdefaultry �negotiate_capabilities�items�popr< r� �start�NotImplementedr
r� �join�decode� send_methodr r �StartOkrB ) ra r[ r\ r] r^ r_ �argsigrA �capr< re r r� r r{ u sT
� �
�
�����zConnection._on_startc C s d S ro r )ra � challenger r r r � s zConnection._on_securec C s� | j pd}|p | j| _|p| j| _|pd| _| jdks |dkr't| j|�| _nt| j|�| _| j s4d| _| jtj j
|| j| j| jf| j d� d S )Nr �r� ) rQ rO rP �server_heartbeat�maxrf �minr� r r �TuneOk�
_on_tune_sent)ra rO rP r� r� rQ r r r r� � s
�zConnection._on_tune�ssbc C s | � tjj|| jddf� d S )N� F)r� r r �OpenrD )ra r� r r r r� � s �zConnection._on_tune_sentc C s d| _ | �| � d S )NT)rH rY rm r r r r} � s zConnection._on_open_okc K s t |f|||||d�|��S )N)r` rS rT rU rV r )ra rC r` rS rT rU rV rg r r r r � s ���zConnection.Transportc C s | j o| j jS ro )rN r� rm r r r r� � s zConnection.connectedc sV � j r� j �� � jr � fdd�� j�� D �}|D ]}|�� qd � _ � _� _d S )Nc s g | ]}|� ur|�qS r r )r* �chrm r r �
<listcomp>� s
�z&Connection.collect.<locals>.<listcomp>)rN rr rI �valuesr� �
connection)ra rI r� r rm r r� � s
�
zConnection.collectc C sZ t | j�}td| jd �D ]}||vr | j�|� | S q
td�t| j�| j�t j
j ��)Nr z/No free channel ids, current={}, channel_max={}) �setrZ �rangerO �appendr r� �lenrI r r
r� )ra �used_channel_ids�
channel_idr r r �_get_free_channel_id� s
� ��z Connection._get_free_channel_idc C s* || j v r
td|�d���| j �|� |S )NzChannel z
already open)rZ r
r� )ra r� r r r �_claim_channel_id� s
z Connection._claim_channel_idc C sN | j du r td��z| j | W S ty& | j| ||d�}|�� | Y S w )z�Create new channel.
Fetch a Channel object identified by the numeric channel_id, or
create that object if it doesn't already exist.
NzConnection already closed.)rY )rI r �KeyErrorr
�open)ra r� r� �channelr r r r� � s
�zConnection.channelc C s t d��)NzUse AMQP heartbeats)�NotImplementedErrorrm r r r �is_alive s zConnection.is_alivec C s | � |�s
| � |�rd S d S ro )�
blocking_read)ra r� r r r r� s
�zConnection.drain_eventsc C s@ | j �|��
| j �� }W d � n1 sw Y | �|�S ro )r� �having_timeout�
read_framer� )ra r� �framer r r r� s �
zConnection.blocking_readc C s& | j d u r td��| j | �|||�S )NzConnection already closed)rI r �dispatch_method)ra r� r( �payload�contentr r r r� s
�z Connection.on_inbound_methodr� �r r c C sp | j du rdS z,z d| _| jtjj||||d |d ftjjd�W W d| _S ttfy3 | � � � w d| _w )� Request a connection close.
This method indicates that the sender wants to close the
connection. This may be due to internal conditions (e.g. a
forced shut-down) or due to an error handling a specific
method, i.e. an exception. When a close is due to an
exception, the sender provides the class and method id of the
method which caused the exception.
RULE:
After sending this method any received method except the
Close-OK method MUST be discarded.
RULE:
The peer sending this method MAY use a counter or timeout
to detect failure of the other peer to respond correctly
with the Close-OK method.
RULE:
When a server receives the Close method from a client it
MUST delete all server-side resources associated with the
client's context. A client CANNOT reconnect to a context
after sending or receiving a Close method.
PARAMETERS:
reply_code: short
The reply code. The AMQ reply codes are defined in AMQ
RFC 011.
reply_text: shortstr
The localised reply text. This text can be logged as an
aid to resolving issues.
class_id: short
failing method class
When the close is provoked by a method exception, this
is the class of the method.
method_id: short
failing method ID
When the close is provoked by a method exception, this
is the ID of the method.
NTr r )�waitF)
rN �
is_closingr� r r r� r� r� r r� )ra �
reply_code�
reply_textr( r� r r r rr s
6�
��zConnection.closec C s | � � t||||ft��)r� )� _x_close_okr r
)ra r� r� �class_id� method_idr r r r� f s 5�zConnection._on_closec C s | j tjj| jd� dS )ah Confirm a connection close.
This method confirms a Connection.Close method and tells the
recipient that it is safe to release resources for the
connection and close the socket.
RULE:
A peer that detects a socket closure without having
received a Close-Ok handshake method SHOULD log the error.
r� N)r� r r r� r� rm r r r r� � s zConnection._x_close_okc C s | � � dS )ai Confirm a connection close.
This method confirms a Connection.Close method and tells the
recipient that it is safe to release resources for the
connection and close the socket.
RULE:
A peer that detects a socket closure without having
received a Close-Ok handshake method SHOULD log the error.
N)r� rm r r r r� � s zConnection._on_close_okc C s d}| j r
| � |�S dS )zlCallback called when connection blocked.
Notes:
This is an RabbitMQ Extension.
z#connection blocked, see broker logsN)rW )ra �reasonr r r r� � s
�zConnection._on_blockedc C s | j r| � � S d S ro )rX rm r r r r� � s �zConnection._on_unblockedc C s | � ddd d d � d S )N� r )r rm r r r �send_heartbeat� s zConnection.send_heartbeat� c C s
t �d| j� | js dS |dkrd}| j}| j}| jdu s"| j|kr&t� | _| j du s0| j |kr4t� | _
t� }t �d| j| j |||| j| j| j� ||| _| _ || j| j| krit �d| j� | � � t� | _d| j }| j
| }|t� k }| j
r�|r�t d��dS dS )a� Send heartbeat packets if necessary.
Raises:
~amqp.exceptions.ConnectionForvced: if none have been
received recently.
Note:
This should be called frequently, on the order of
once per second.
Keyword Arguments:
rate (int): Number of heartbeat frames to send during the heartbeat
timeout
z"heartbeat_tick : for connection %sNr r� z�heartbeat_tick : Prev sent/recv: %s/%s, now - %s/%s, monotonic - %s, last_heartbeat_sent - %s, heartbeat int. - %s for connection %sz3heartbeat_tick: sending heartbeat for connection %szToo many heartbeats missed)
�AMQP_HEARTBEAT_LOGGERr� r9 rf �
bytes_sent�
bytes_recv� prev_sentr �last_heartbeat_sent� prev_recv�last_heartbeat_receivedr� r )ra �rate�sent_now�recv_now�now�two_heartbeats�two_heartbeats_interval�heartbeats_missedr r r �heartbeat_tick� sH �� �
�zConnection.heartbeat_tickc C s | j jS ro )r� �sockrm r r r r�
s zConnection.sockc C s | j �d�pi S )Nr� )r] r� rm r r r �server_capabilities s z Connection.server_capabilitiesro )r� )r& )r� )FNNN)NN)r r� r� r' )r� )Tr r r �__doc__r
�NEGOTIATE_CAPABILITIESr� �LIBRARY_PROPERTIESr@ rf rQ r� r� r� r� r� r� r� r �methodr rz r| r~ r� r� r� r� r� �_METHODS� _ALLOWED_METHODS_WHEN_CLOSINGr
�socket�error�IOErrorr� �connection_errorsr �channel_errorsr r � recoverable_connection_errorsr �recoverable_channel_errorsr r rK rn rq rt ru r� rp r� �propertyr� �setterr� r{ r r� r� r} r r� r� r� r� r� r� r� r� r� rr r� r� r� r� r� r� r� r� r� �
__classcell__r r ri r r ? s� . �
�����[
�1
�
�H9
?
r )/r� �loggingr r6 r� r �timer �viner r� r r r �abstract_channelr r� r
�
exceptionsr r r
r r r r r r �method_framingr r r� r rS r � ImportError� Exceptionr� �stripr� �__all__� getLoggerr� r� r� r� r r r r r �<module> sH , � �
���