Файловый менеджер - Редактировать - /opt/imh-python/lib/python3.9/site-packages/kombu/__pycache__/connection.cpython-39.pyc
Ðазад
a o�h � � @ s� d Z ddlmZ ddlZddlZddlZddlmZ ddlm Z m Z ddl m Z ddl mZmZ zddlmZ d ZW n ey� dZd ZY n0 dd lmZ d d lmZ d dlmZ d dlmZm Z d dl m Z d dl m Z m!Z!m"Z"m#Z# d dl$m%Z% d dl&m'Z'm(Z(m)Z)m*Z*m+Z+ e�rTddl,m-Z- ej.dk �r<ddl/m0Z0 n ddl m0Z0 ddl1m2Z2 dZ3ee4�Z5e Z6ddd�Z7e6e#d�Z8ej9�:d d �Z;ej9�:d d �Z<G d d � d �Z=e=Z>G d d!� d!e=�Z?G d"d#� d#e�Z@G d$d%� d%e�ZAd&d'd(�d)d*�ZBd+d,d-�d.d/�ZCdS )0zClient (Connection).� )� annotationsN)�contextmanager)�count�cycle)� itemgetter)� TYPE_CHECKING�Any)� CERT_NONETF)� exceptions� )� get_logger)�Resource)�get_transport_cls�supports_librabbitmq)� HashedSeq)� dictfilter�lazy�retry_over_time� shufflecycle)�cached_property)�as_url�maybe_sanitize_url� parse_url�quote�urlparse)�Channel)� � )� TypeGuard)� TracebackType)� Connection�ConnectionPool� ChannelPool�amqp)Zpyamqp� librabbitmq)� round-robin�shuffleZKOMBU_LOG_CONNECTIONZKOMBU_LOG_CHANNELc @ s� e Zd ZdZdZdZdZdZdZdZ dZ dZ dZ dZ dZdZdZdZeZeZd Z Z Z ZZd�d d �Zd d � Zd d� Zdd� Z dd� Z dd� Z dd� Z dd� Z d�dd�Z!d d � Z"d d � Z#d d!� Z$d"d#� Z%d$d%� Z&d�d&d'�Z'd(d)� Z(e(Z)d*d+� Z*d�d.d/�Z+e,e-j.e-j.fd0d1��Z/e,d2d3� �Z0d4d5� Z1d6d7� Z2d�d9d:�Z3d�d;d<�Z4d=d>� Z5d?d@� Z6dAdB� Z7dCdD� Z8d�dEdF�Z9dGdH� Z:dIdJ� Z;ddKe<dLdMdNdOdP�fdQdR�dSdT�Z=d�dUdV�Z>d�dWdX�Z?d�dYdZ�Z@d�d[d\�ZAd�d]d^�ZBd�d_d`�ZCdadb� ZDdcdd� ZEdedf� ZFdgdh� ZGdidj� ZHdkdl� ZIdmdndodpdq�drds�ZJeKdtdu� �ZLdvdw� ZMeKdxdy� �ZNeKdzd{� �ZOd|d}� ZPeKd~dR�dd���ZQeKd�d�� �ZReKd�d�� �ZSeTd�d�� �ZUd�d�� ZVeTd�d�� �ZWeTd�d�� �ZXeTd�d�� �ZYeTd�d�� �ZZeKd�d�� �Z[eKd�d�� �Z\dS )�r af A connection to the broker. Example: ------- >>> Connection('amqp://guest:guest@localhost:5672//') >>> Connection('amqp://foo;amqp://bar', ... failover_strategy='round-robin') >>> Connection('redis://', transport_options={ ... 'visibility_timeout': 3000, ... }) >>> import ssl >>> Connection('amqp://', login_method='EXTERNAL', ssl={ ... 'ca_certs': '/etc/pki/tls/certs/something.crt', ... 'keyfile': '/etc/something/system.key', ... 'certfile': '/etc/something/system.cert', ... 'cert_reqs': ssl.CERT_REQUIRED, ... }) Note: ---- SSL currently only works with the py-amqp, qpid and redis transports. For other transports you can use stunnel. Arguments: --------- URL (str, Sequence): Broker URL, or a list of URLs. Keyword Arguments: ----------------- ssl (bool/dict): Use SSL to connect to the server. Default is ``False``. May not be supported by the specified transport. transport (Transport): Default transport if not specified in the URL. connect_timeout (float): Timeout in seconds for connecting to the server. May not be supported by the specified transport. transport_options (Dict): A dict of additional connection arguments to pass to alternate kombu channel implementations. Consult the transport documentation for available options. heartbeat (float): Heartbeat interval in int/float seconds. Note that if heartbeats are enabled then the :meth:`heartbeat_check` method must be called regularly, around once per second. Note: ---- The connection is established lazily when needed. If you need the connection to be established, then force it by calling :meth:`connect`:: >>> conn = Connection('amqp://') >>> conn.connect() and always remember to close the connection:: >>> conn.release() These options have been replaced by the URL argument, but are still supported for backwards compatibility: :keyword hostname: Host name/address. NOTE: You cannot specify both the URL argument and use the hostname keyword argument at the same time. :keyword userid: Default user name if not provided in the URL. :keyword password: Default password if not provided in the URL. :keyword virtual_host: Default virtual host if not provided in the URL. :keyword port: Default port if not provided in the URL. N�/� Fr% � localhostr c K s� |d u r g n|}||||||||| | | d� }| _ |r^t|t�s^|�|� |d }|j|d� |�rd|v r�|�d�| }|d }|j|d� d|v r�d|d |�d�� v r�|�dd�\|d< |d < |d | _nJd|v �r|p�t|�j }t |�j �st |�}|jt |�|d d� ||d< | jf i |�� || _|�p8d | _| j�| j��pP| j| _| j�rt| �| j�| _t| j� | d u �r�i } | | _t�r�d | _| �r�| | _t� | _d S ) N) �hostname�userid�password� virtual_host�port�insist�ssl� transport�connect_timeout� login_method� heartbeatr )r* �;�://�+r r1 r* r% T)�_initial_params� isinstance�str�extend�update�split�index� uri_prefixr �schemer � can_parse_urlr r � _init_params�alt�_failover_strategy�failover_strategies�get�failover_strategyr �next�transport_options�_log_connection�_logger�set�declared_entities)�selfr* r+ r, r- r. r/ r0 r1 r2 rI r3 r? r4 rG � alternates�kwargsrC �paramsZ url_params� rR ��/root/rpmbuild/BUILDROOT/imh-python39-modules-3.9.7-92.el8.x86_64/opt/imh-python/lib/python3.9/site-packages/kombu/connection.py�__init__� s` � � � �� zConnection.__init__c C sR | � � | j�� d| _d|v r(t|�nd|i}| jf i t| jfi |���� dS )z�Switch connection parameters to use a new URL or hostname. Note: ---- Does not reconnect! Arguments: --------- conn_str (str): either a hostname or URL. Fr6 r* N)�closerM �clear�_closedr rB �dictr8 )rN Zconn_strZ conn_paramsrR rR rS �switch� s �zConnection.switchc C s | j r| �t| j �� dS )z:Switch to next URL given by the current failover strategy.N)r rY rH �rN rR rR rS �maybe_switch_next s z Connection.maybe_switch_nextc C s� |pd}|dkrt � rd}|dkr<tr<|s<t�d� dti}|| _|| _|| _| | _|p\| j | _ |ph| j | _ || _ | | _ || _ || _| o�t| �| _d S )Nr# r$ ZredisszaSecure redis scheme specified (rediss) with no ssl options, defaulting to insecure SSL behaviour.Z ssl_cert_reqs)r � ssl_available�logger�warningr r* r+ r, r3 r- r. r/ r2 r0 � transport_cls�floatr4 ) rN r* r+ r, r- r. r/ r0 r1 r2 r3 r4 rR rR rS rB s& � zConnection._init_paramsc C s | j �| j|� d S �N)r1 �register_with_event_loop� connection)rN �looprR rR rS rb s z#Connection.register_with_event_loopc O s8 | j r4d}tj|jt| �t|�d�g|�R i |�� d S )Nz [Kombu connection:{id:#x}] {msg})�id�msg)rK r] �debug�formatre r: )rN rf �argsrP �fmtrR rR rS �_debug s ��zConnection._debugc C s | j ddd�S )z+Establish connection to server immediately.r F�� max_retries�reraise_as_library_errors��_ensure_connectionrZ rR rR rS �connect& s �zConnection.connectc C s8 | � d� | j�| j�}tr4ddlm} ||dd�S |S )z Create and return a new channel.zcreate channelr )� Logwrappedz kombu.channelz [Kombu channel:{0.channel_id}] )rk r1 Zcreate_channelrc � _log_channelZ utils.debugrr )rN Zchanrr rR rR rS �channel, s �zConnection.channel� c C s | j j| j|d�S )a� Check heartbeats. Allow the transport to perform any periodic tasks required to make heartbeats work. This should be called approximately every second. If the current transport does not support heartbeats then this is a noop operation. Arguments: --------- rate (int): Rate is how often the tick is called compared to the actual heartbeat value. E.g. if the heartbeat is set to 3 seconds, and the tick is called every 3 / 2 seconds, then the rate is 2. This value is currently unused by any transports. )�rate)r1 �heartbeat_checkrc )rN rv rR rR rS rw 6 s zConnection.heartbeat_checkc K s | j j| jfi |��S )z�Wait for a single event from the server. Arguments: --------- timeout (float): Timeout in seconds before we give up. Raises ------ socket.timeout: if the timeout is exceeded. )r1 � drain_eventsrc �rN rP rR rR rS rx J s zConnection.drain_eventsc C s, z |� � W n | j| j y& Y n0 dS )z>Close given channel, but ignore connection and channel errors.N)rU �connection_errors�channel_errors�rN rt rR rR rS �maybe_close_channelW s z Connection.maybe_close_channelc C s^ | j �� | jr | �| j� | jrZz| j�| j� W n | jtt j f yR Y n0 d | _d S ra ) rM rV �_default_channelr} � _connectionr1 Zclose_connectionrz �AttributeError�socket�errorrZ rR rR rS �_do_close_self^ s zConnection._do_close_selfc C s$ | � � | �� | �d� d| _dS )z;Really close connection, even if part of a connection pool.�closedTN)r� �_do_close_transportrk rW rZ rR rR rS �_closej s zConnection._closec C s | j rd | j _d | _ d S ra )� _transport�clientrZ rR rR rS r� q s z Connection._do_close_transportc C s� z | j j}W nb tyn t�� }t�|� z0z | �� W n tjyN Y n0 W t�|� n t�|� 0 Y n 0 || j� | � � | j � � d | _d S ra ) r� Z_collectr� r� �getdefaulttimeout�setdefaulttimeoutr� �timeoutr r� rM rV )rN �socket_timeoutZ gc_transportZ_timeorR rR rS �collectv s zConnection.collectc C s | � � dS )z Close the connection (if open).N�r� rZ rR rR rS �release� s zConnection.releasec O s | j |i |�� | S )ztPublic interface of _ensure_connection for retro-compatibility. Returns kombu.Connection instance. ro �rN ri rP rR rR rS �ensure_connection� s z Connection.ensure_connection� Tc sx �j r �jS d� �fdd� } �j} |s,�j} | � �2 t�j�jdi | ||||||d� W d � S 1 sj0 Y dS )a� Ensure we have a connection to the server. If not retry establishing the connection with the settings specified. Arguments: --------- errback (Callable): Optional callback called each time the connection can't be established. Arguments provided are the exception raised and the interval that will be slept ``(exc, interval)``. max_retries (int): Maximum number of times to retry. If this limit is exceeded the connection error will be re-raised. interval_start (float): The number of seconds we start sleeping for. interval_step (float): How many seconds added to the interval for each retry. interval_max (float): Maximum number of seconds to sleep between each retry. callback (Callable): Optional callback that is called for every internal iteration (1 s). timeout (int): Maximum amount of time in seconds to spend attempting to connect, total over all retries. r c s8 �� |�}|rt|�}� r$� | |� ��� |r4|S dS �Nr )�completes_cyclerH r[ )�excZ intervals�retries�interval�round��errbackrN rR rS �on_error� s z/Connection._ensure_connection.<locals>.on_errorrR )r� N)r )� connectedr �_reraise_as_library_errors�_dummy_contextr �_connection_factory� recoverable_connection_errors) rN r� rm �interval_start� interval_step� interval_max�callbackrn r� r� �ctxrR r� rS rp � s ! �z Connection._ensure_connectionc c s� z d V W n| ||fy" � Y nf | j yT } z|t|��|�W Y d }~n<d }~0 | jy� } z|t|��|�W Y d }~n d }~0 0 d S ra )r� r: �recoverable_channel_errors)rN �ConnectionErrorZ ChannelErrorr� rR rR rS r� � s "z%Connection._reraise_as_library_errorsc c s d V d S ra rR rZ rR rR rS r� � s zConnection._dummy_contextc C s | j r|d t| j � S dS )z?Return true if the cycle is complete after number of `retries`.r T)rC �len)rN r� rR rR rS r� � s zConnection.completes_cyclec C s& | j r"|| j ur"| �| j � d| _ dS )z2Revive connection after connection re-established.N)r~ r} )rN Z new_channelrR rR rS �revive� s zConnection.reviver c sN �du rt � �� ��������� f dd�} �j� d�| _�j| _�j| _| S )a. Ensure operation completes. Regardless of any channel/connection errors occurring. Retries by establishing the connection, and reapplying the function. Arguments: --------- obj: The object to ensure an action on. fun (Callable): Method to apply. errback (Callable): Optional callback called each time the connection can't be established. Arguments provided are the exception raised and the interval that will be slept ``(exc, interval)``. max_retries (int): Maximum number of times to retry. If this limit is exceeded the connection error will be re-raised. interval_start (float): The number of seconds we start sleeping for. interval_step (float): How many seconds added to the interval for each retry. interval_max (float): Maximum number of seconds to sleep between each retry. on_revive (Callable): Optional callback called whenever revival completes successfully retry_errors (tuple): Optional list of errors to retry on regardless of the connection state. Examples -------- >>> from kombu import Connection, Producer >>> conn = Connection('amqp://') >>> producer = Producer(conn) >>> def errback(exc, interval): ... logger.error('Error: %r', exc, exc_info=1) ... logger.info('Retry in %s seconds.', interval) >>> publish = conn.ensure(producer, producer.publish, ... errback=errback, max_retries=3) >>> publish({'hello': 'world'}, routing_key='dest') Nc s� d}� j }� j}t� jd�}� �� ��� td�D �]�}z"�| i |��W W d � S �y� } z.�d urz|�krz� � jd|dd� W Y d }~q0d }~0 |�y\ } z�|r�|s�� �d ur�|�krʂ � jd|dd� � �� � o� |d� d }�d u�r t�| d�}� j � |���dd� � j } �� | � ��r@�| � |d7 }W Y d }~q0d }~0 |�y� } zB�d u�r�|�k�r�� � jd |dd� � �o�� |d� W Y d }~q0d }~0 0 q0W d � n1 �s�0 Y d S ) Nr r� z ensure retry policy error: %rr )�exc_infozensure connection error: %rF)rn zensure channel error: %r) r� r� �hasattrr1 r� r rk r� �maxrp �default_channelr� ) ri rP Zgot_connectionZ conn_errorsZ chan_errorsZhas_modern_errorsr� r� Zremaining_retriesrt � r� �funr� r� r� rm �obj� on_revive� retry_errorsrN rR rS �_ensured" s^ � "�� � �z#Connection.ensure.<locals>._ensuredz (ensured))�tuple�__name__�__doc__� __module__) rN r� r� r� rm r� r� r� r� r� r� rR r� rS �ensure� s 1 1zConnection.ensurec s6 |g� G � �fdd�d�}|| �}| j ||fi |��S )a Decorator for functions supporting a ``channel`` keyword argument. The resulting callable will retry calling the function if it raises connection or channel related errors. The return value will be a tuple of ``(retval, last_created_channel)``. If a ``channel`` is not provided, then one will be automatically acquired (remember to close it afterwards). See Also -------- :meth:`ensure` for the full list of supported keyword arguments. Example: ------- >>> channel = connection.channel() >>> try: ... ret, channel = connection.autoretry( ... publish_messages, channel) ... finally: ... channel.close() c sR e Zd Ze�dd�Z e�dd�Ze�dd�Zdd� Z� fdd�Z� �fd d �ZdS ) z%Connection.autoretry.<locals>.Revivalr� Nr� r� c S s || _ d S ra )rc )rN rc rR rR rS rT v s z.Connection.autoretry.<locals>.Revival.__init__c s |� d< d S r� rR r| )�channelsrR rS r� y s z,Connection.autoretry.<locals>.Revival.revivec s8 � d d u r| � | jj� �|d� d i|��� d fS )Nr rt )r� rc r� r� �r� r� rR rS �__call__| s z.Connection.autoretry.<locals>.Revival.__call__)r� r� � __qualname__�getattrr� rT r� r� rR r� rR rS �Revivalq s r� )r� )rN r� rt Zensure_optionsr� r� rR r� rS � autoretryX s zConnection.autoretryc C s | � � | d�S )N)r� )r rZ rR rR rS �create_transport� s zConnection.create_transportc C s | j }|rt|t�r t|�}|S )z'Get the currently used transport class.)r_ r9 r: r )rN r_ rR rR rS r � s z Connection.get_transport_clsc K s$ | j f i t| jdd�fi |����S )z3Create a copy of the connection with same settings.F��resolve)� __class__rX �_infory rR rR rS �clone� s zConnection.clonec C s | j �| j�S ra )r1 �get_heartbeat_intervalrc rZ rR rR rS r� � s z!Connection.get_heartbeat_intervalc C s | j }|r| j�||�}| jj}| jsT|�d�rTt�d|�d�� d�� |�d�}n| j}| jrp| j� d|� �}d|fd| j p�|�d�fd| j p�|�d�fd| j p�|�d�fd| j p�|�d�fd | j fd | jfd |fd | jfd | jfd| jp�|�d�fd| jfd| jfd| jfd| jff}|S )Nr* z0No hostname was supplied. Reverting to default '�'r7 r+ r, r- r. r/ r0 r1 r2 rI r3 r? r4 rG rO )r_ �resolve_aliasesrF r1 Zdefault_connection_paramsr* r] r^ r? r+ r, r- r. r/ r0 r2 rI r3 r4 rD rC )rN r� r_ �Dr* �inforR rR rS r� � sB ��� �zConnection._infoc C s t | �� �S )zGet connection info.)rX r� rZ rR rR rS r� � s zConnection.infoc C s& t | j| j| j| j| j| jt| j��S ra ) r r_ r* r+ r, r- r. �reprrI rZ rR rR rS � __eqhash__� s �zConnection.__eqhash__z**r. r+ r, r- r1 r: )�returnc C s� | j pd}| jjrd| j }z| j�|||�W S ty< Y n0 | jrT| j� d|� �}|s`t|�}|S | jr�| j� d|� �}|s�t|�}|S | �� }||�\}}} } } t| |||| t | �| |d�S )z*Convert connection parameters to URL form.r) r7 )�sanitize�mask) r* r1 rA �as_uri�NotImplementedErrorr? r r� r r ) rN Zinclude_passwordr� Z getfieldsr* Zconnection_as_uri�fieldsr. r+ r, Zvhostr1 rR rR rS r� � s2 � �zConnection.as_uric K s t | |fi |��S )a/ Pool of connections. See Also -------- :class:`ConnectionPool`. Arguments: --------- limit (int): Maximum number of active connections. Default is no limit. Example: ------- >>> connection = Connection('amqp://') >>> pool = connection.Pool(2) >>> c1 = pool.acquire() >>> c2 = pool.acquire() >>> c3 = pool.acquire() Traceback (most recent call last): File "<stdin>", line 1, in <module> File "kombu/connection.py", line 354, in acquire raise ConnectionLimitExceeded(self.limit) kombu.exceptions.ConnectionLimitExceeded: 2 >>> c1.release() >>> c3 = pool.acquire() )r! �rN �limitrP rR rR rS �Pool� s zConnection.Poolc K s t | |fi |��S )a' Pool of channels. See Also -------- :class:`ChannelPool`. Arguments: --------- limit (int): Maximum number of active channels. Default is no limit. Example: ------- >>> connection = Connection('amqp://') >>> pool = connection.ChannelPool(2) >>> c1 = pool.acquire() >>> c2 = pool.acquire() >>> c3 = pool.acquire() Traceback (most recent call last): File "<stdin>", line 1, in <module> File "kombu/connection.py", line 354, in acquire raise ChannelLimitExceeded(self.limit) kombu.connection.ChannelLimitExceeded: 2 >>> c1.release() >>> c3 = pool.acquire() )r" r� rR rR rS r" � s zConnection.ChannelPoolc O s& ddl m} ||p| g|�R i |��S )z,Create new :class:`kombu.Producer` instance.r )�Producer)� messagingr� )rN rt ri rP r� rR rR rS r� s zConnection.Producerc O s( ddl m} ||p| |g|�R i |��S )z,Create new :class:`kombu.Consumer` instance.r )�Consumer)r� r� )rN Zqueuesrt ri rP r� rR rR rS r� ! s zConnection.Consumerc K s* ddl m} ||p| |||||fi |��S )a� Simple persistent queue API. Create new :class:`~kombu.simple.SimpleQueue`, using a channel from this connection. If ``name`` is a string, a queue and exchange will be automatically created using that name as the name of the queue and exchange, also it will be used as the default routing key. Arguments: --------- name (str, kombu.Queue): Name of the queue/or a queue. no_ack (bool): Disable acknowledgments. Default is false. queue_opts (Dict): Additional keyword arguments passed to the constructor of the automatically created :class:`~kombu.Queue`. queue_args (Dict): Additional keyword arguments passed to the constructor of the automatically created :class:`~kombu.Queue` for setting implementation extensions (e.g., in RabbitMQ). exchange_opts (Dict): Additional keyword arguments passed to the constructor of the automatically created :class:`~kombu.Exchange`. channel (ChannelT): Custom channel to use. If not specified the connection default channel is used. r )� SimpleQueue)�simpler� ) rN �name�no_ack� queue_opts� queue_args� exchange_optsrt rP r� rR rR rS r� & s ��zConnection.SimpleQueuec K s* ddl m} ||p| |||||fi |��S )a� Simple ephemeral queue API. Create new :class:`~kombu.simple.SimpleQueue` using a channel from this connection. See Also -------- Same as :meth:`SimpleQueue`, but configured with buffering semantics. The resulting queue and exchange will not be durable, also auto delete is enabled. Messages will be transient (not persistent), and acknowledgments are disabled (``no_ack``). r )� SimpleBuffer)r� r� ) rN r� r� r� r� r� rt rP r� rR rR rS r� F s ��zConnection.SimpleBufferc C s$ | � d� | j�� }| � d| � |S )Nzestablishing connection...zconnection established: %r)rk r1 Zestablish_connection)rN �connrR rR rS �_establish_connectionZ s z Connection._establish_connectionc C s || j jjv S ra )r1 � implements� exchange_type)rN r� rR rR rS �supports_exchange_type` s z!Connection.supports_exchange_typec C s d| � � � dt| �d�d�S )Nz <Connection: z at z#x�>)r� re rZ rR rR rS �__repr__c s zConnection.__repr__c C s | � � S ra )r� rZ rR rR rS �__copy__f s zConnection.__copy__c C s | j t| �� �� �d fS ra )r� r� r� �valuesrZ rR rR rS � __reduce__i s zConnection.__reduce__c C s | S ra rR rZ rR rR rS � __enter__l s zConnection.__enter__ztype[BaseException] | NonezBaseException | NonezTracebackType | None�None)�exc_type�exc_val�exc_tbr� c C s | � � d S ra )r� �rN r� r� r� rR rR rS �__exit__o s zConnection.__exit__c C s | j �| j�S ra )r1 �qos_semantics_matches_specrc rZ rR rR rS r� w s z%Connection.qos_semantics_matches_specc C s� d| j i}| j}|r�d|v r(|d |d< d|v r<|d |d< d|v rP|d |d< d|v rd|d |d< d|v rx|d |d< d|v r�|d |d< d|v r�|d |d< |S ) Nr� rm r� r� r� Zconnect_retries_timeoutr� r� )r2 rI )rN � conn_optsZtransport_optsrR rR rS �_extract_failover_opts{ s& � z!Connection._extract_failover_optsc C s | j o | jduo | j�| j�S )z3Return true if the connection has been established.N)rW r r1 Zverify_connectionrZ rR rR rS r� � s � �zConnection.connectedc C s$ | j s | js| jddd�S | jS dS )z�The underlying connection object. Warning: ------- This instance is transport specific, so do not depend on the interface of this object. r Frl N)rW r� rp r rZ rR rR rS rc � s �zConnection.connectionc C s& | j �� d | _| �� | _d| _| jS )NF)rM rV r~ r� r rW rZ rR rR rS r� � s z Connection._connection_factoryr c C s2 | � � }| jf i |�� | jdu r,| �� | _| jS )aw Default channel. Created upon access and closed when the connection is closed. Note: ---- Can be used for automatic channel handling when you only need one channel, and also it is the channel implicitly used if a connection is passed instead of a channel, to functions that require a channel. N)r� rp r~ rt )rN r� rR rR rS r� � s zConnection.default_channelc C s d� | jt| j�g�S )z5The host as a host name/port pair separated by colon.�:)�joinr* r: r. rZ rR rR rS �host� s zConnection.hostc C s | j d u r| �� | _ | j S ra )r� r� rZ rR rR rS r1 � s zConnection.transportc C s | j jS )z�AMQP Management API. Experimental manager that can be used to manage/monitor the broker instance. Not available for all transports. )r1 �managerrZ rR rR rS r� � s zConnection.managerc O s | j j|i |��S ra )r1 � get_managerr� rR rR rS r� � s zConnection.get_managerc C s0 z | � � jW S ty* | j| j Y S 0 dS )z�Recoverable connection errors. List of connection related exceptions that can be recovered from, but where the connection must be closed and re-established first. N)r r� r� rz r{ rZ rR rR rS r� � s z(Connection.recoverable_connection_errorsc C s& z | � � jW S ty Y dS 0 dS )z�Recoverable channel errors. List of channel related exceptions that can be automatically recovered from without re-establishing the connection. rR N)r r� r� rZ rR rR rS r� � s z%Connection.recoverable_channel_errorsc C s | � � jS )z8List of exceptions that may be raised by the connection.)r rz rZ rR rR rS rz � s z Connection.connection_errorsc C s | � � jS )z5List of exceptions that may be raised by the channel.)r r{ rZ rR rR rS r{ � s zConnection.channel_errorsc C s | j jjS ra )r1 r� Z heartbeatsrZ rR rR rS �supports_heartbeats s z Connection.supports_heartbeatsc C s | j jjS ra )r1 r� Z asynchronousrZ rR rR rS � is_evented s zConnection.is_evented)r) NNNNFFNr( NNNr r% N)ru )N)NNru ru r� NTN)NNr r r NN)N)T)N)N)N)NN)NNNNN)NNNNN)]r� r� r� r� r. r- r2 rW r r~ r� rK r? rM r rI rG r4 r� rE r* r+ r, r0 r3 rT rY r[ rB rb rk rq rt rw rx r} r� r� r� r� r� rU r� rp r r ZOperationalErrorr� r� r� r� r� r� r� r r� r� r� r� r� r r� r� r" r� r� r� r� r� r� r� r� r� r� r� �propertyr� r� r� rc r� r� r� r1 r r� r� r� r� rz r{ r� r� rR rR rR rS r ? s� E � D � 9� � j , %�� � � r c s4 e Zd ZdZ� fdd�Zdd� Z� fdd�Z� ZS )�PooledConnectionz�Wraps :class:`kombu.Connection`. This wrapper modifies :meth:`kombu.Connection.__exit__` to close the connection in case any exception occurred while the context was active. c s || _ t� jf i |�� d S ra )�_pool�superrT )rN �poolrP �r� rR rS rT s zPooledConnection.__init__c C s | S ra rR rZ rR rR rS r� s zPooledConnection.__enter__c s, |d ur | j jr | j �| � t� �|||�S ra )r� r� �replacer� r� r� r rR rS r� s zPooledConnection.__exit__)r� r� r� r� rT r� r� � __classcell__rR rR r rS r� s r� c sl e Zd ZdZejZdZd� fdd� Zdd� Z dd � Z d d � Z dd d�Z e ddd��Zdd� Zdd� Z� ZS )r! zPool of connections.TNc s || _ t� j|d� d S �N)r� �rc r� rT �rN rc r� rP r rR rS rT ( s zConnectionPool.__init__c C s t | fi t| jjdd����S )NFr� )r� rX rc r� rZ rR rR rS �new, s zConnectionPool.newc C s& z|� d� W n ty Y n0 d S )NZreleased)rk r� �rN �resourcerR rR rS �release_resource/ s z ConnectionPool.release_resourcec C s |� � d S ra r� r rR rR rS �close_resource5 s z ConnectionPool.close_resource皙�����?c C s t |t�s|�|�S d S ra )r9 r r� )rN r r� rR rR rS �collect_resource8 s z ConnectionPool.collect_resourceFc c s< | j |d�� }||jfV W d � n1 s.0 Y d S )N)�block)�acquirer� )rN r rc rR rR rS �acquire_channel< s z ConnectionPool.acquire_channelc C s> | j r:| jj}t|�| j t| j� k r:| j�t| j�� qd S ra )r� � _resource�queuer� �_dirty� put_nowaitr r )rN �qrR rR rS �setupA s zConnectionPool.setupc C s t |�r|� }|�d� |S )NZacquired)�callablerk r rR rR rS �prepareH s zConnectionPool.prepare)N)r )F)r� r� r� r� r ZConnectionLimitExceeded� LimitExceededZclose_after_forkrT r r r r r r r r r rR rR r rS r! " s r! c s@ e Zd ZdZejZd � fdd� Zdd� Zdd� Z d d � Z � Z S ) r" zPool of channels.Nc s || _ t� j|d� d S r r r r rR rS rT T s zChannelPool.__init__c C s t | jj�S ra )r rc rt rZ rR rR rS r X s zChannelPool.newc C sD | � � }| jr@| jj}t|�| jt| j� k r@| j�t|�� qd S ra )r r� r r r� r r r )rN rt r rR rR rS r [ s zChannelPool.setupc C s t |�r|� }|S ra )r r| rR rR rS r c s zChannelPool.prepare)N) r� r� r� r� r ZChannelLimitExceededr rT r r r r rR rR r rS r" O s r" zChannel | Connectionr )rt r� c C s t | �r| jS | S )z�Get channel from object. Return the default channel if argument is a connection instance, otherwise just return the channel given. )� is_connectionr� )rt rR rR rS � maybe_channeli s r r zTypeGuard[Connection])r� r� c C s t | t�S ra )r9 r )r� rR rR rS r t s r )Dr� � __future__r �osr� �sys� contextlibr � itertoolsr r �operatorr �typingr r r0 r r\ � ImportErrorZkombur �logr r r r1 r r Zutils.collectionsr Zutils.functionalr r r r Z utils.objectsr Z utils.urlr r r r r Zkombu.transport.virtualr � version_info�typing_extensionsr �typesr �__all__r� r] Zroundrobin_failoverr� rE �environrF rJ rs r ZBrokerConnectionr� r! r" r r rR rR rR rS �<module> sj �� S-
| ver. 1.1 | |
.
| PHP 8.3.30 | Ð“ÐµÐ½ÐµÑ€Ð°Ñ†Ð¸Ñ Ñтраницы: 0 |
proxy
|
phpinfo
|
ÐаÑтройка