)JfKVdZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddlZ ddlmZddlmZddlZ ddlmZddlmZddl mZmZmZmZmZdZejeZ Gdde!Z"Gdde"Z#Gd d e"Z$Gd d e!Z%Gd de j&j'Z(dS)zCore connection objectsN)xrange url_unquotedictkeysdict_itervaluesdict_iteritemszPika Python Client Libraryc.eZdZdZdZdZdZdZej j Z dZ ej eeZdZejZdZdZdZdZd Zd Zd Zd ZdZd ZdZdZdZ dZ!dZ"dZ#e$dZ%e%j&dZ%e$dZ'e'j&dZ'e$dZ(e(j&dZ(e$dZ)e)j&dZ)e$dZ e j&dZ e$dZ*e*j&dZ*e$dZ+e+j&d Z+e$d!Z,e,j&d"Z,e$d#Z-e-j&d$Z-e$d%Z.e.j&d&Z.e$d'Z/e/j&d(Z/e$d)Z0e0j&d*Z0e$d+Z1e1j&d,Z1e$d-Z2e2j&d.Z2e$d/Z3e3j&d0Z3e$d1Z4e4j&d2Z4dS)3 Parametersz1Base connection parameters class definition )_blocked_connection_timeout _channel_max_client_properties_connection_attempts _credentials _frame_max _heartbeat_host_locale_port _retry_delay_socket_timeout_stack_timeout _ssl_options _virtual_host _tcp_optionsguestN localhosten_USi(g@g$@g.@Fi'/cfd|_|j|_d|_|j|_d|_|j|_d|_ |j |_ d|_ |j |_d|_|j|_d|_|j|_d|_|j|_d|_|j|_d|_|j|_d|_|j|_ d|_!|j"|_#d|_$|j%|_&d|_'|j(|_)d|_*|j+|_,d|_-|j.|_/dSN)0r "DEFAULT_BLOCKED_CONNECTION_TIMEOUTblocked_connection_timeoutr DEFAULT_CHANNEL_MAX channel_maxr DEFAULT_CLIENT_PROPERTIESclient_propertiesr DEFAULT_CONNECTION_ATTEMPTSconnection_attemptsrDEFAULT_CREDENTIALS credentialsrDEFAULT_FRAME_MAX frame_maxrDEFAULT_HEARTBEAT_TIMEOUT heartbeatr DEFAULT_HOSThostrDEFAULT_LOCALElocaler DEFAULT_PORTportrDEFAULT_RETRY_DELAY retry_delayrDEFAULT_SOCKET_TIMEOUTsocket_timeoutrDEFAULT_STACK_TIMEOUT stack_timeoutrDEFAULT_SSL_OPTIONS ssl_optionsrDEFAULT_VIRTUAL_HOST virtual_hostrDEFAULT_TCP_OPTIONS tcp_optionsselfs R/home/alex/cs2snipeproduction/venv/lib/python3.11/site-packages/pika/connection.py__init__zParameters.__init__Hs5 ,0(  3 '!3"&!%!?$(!#'#C  3/7 %  )  %  3#"9"!7 3! 5 3c |d|jjd|jd|jd|jdt |jd S)zERepresent the info about the instance. :rtype: str ) __class____name__r0r4r>boolr<rAs rC__repr__zParameters.__repr__~sM((($)))TYYY"""D)9$:$:$:$:< =rEczt|tr |j|jko|j|jkStSr ) isinstancer rrNotImplemented)rBothers rC__eq__zParameters.__eq__s8 eZ ( ( K:,Ju{1J JrEcR||}|tur| StSr )rQrO)rBrPresults rC__ne__zParameters.__ne__s,U##  ' ': rEc|jS)z :returns: blocked connection timeout. Defaults to `DEFAULT_BLOCKED_CONNECTION_TIMEOUT`. :rtype: float|None )r rAs rCr"z%Parameters.blocked_connection_timeouts //rEc|Dt|tjstd||dkrt d|||_dS)a :param value: If not None, blocked_connection_timeout is the timeout, in seconds, for the connection to remain blocked; if the timeout expires, the connection will be torn down, triggering the connection's on_close_callback Nz:blocked_connection_timeout must be a Real number, but got rz1blocked_connection_timeout must be >= 0, but got )rNnumbersReal TypeError ValueErrorr rBvalues rCr"z%Parameters.blocked_connection_timeoutsv  eW\22 Ai8=!@AAAqyy j-2U"5666+0(((rEc|jS)z :returns: max preferred number of channels. Defaults to `DEFAULT_CHANNEL_MAX`. :rtype: int )r rAs rCr$zParameters.channel_max   rEct|tjstd||dks|tjjkr#tdtjj|fz||_dS)z :param int value: max preferred number of channels, between 1 and `channel.MAX_CHANNELS`, inclusive z$channel_max must be an int, but got rz-channel_max must be <= %i and > 0, but got %rN) rNrWIntegralrYpikachannel MAX_CHANNELSrZr r[s rCr$zParameters.channel_maxs%!122 Q)OPP P 199 999L"l7?@AA A!rEc|jS)z :returns: client properties used to override the fields in the default client properties reported to RabbitMQ via `Connection.StartOk` method. Defaults to `DEFAULT_CLIENT_PROPERTIES`. :rtype: dict|None )r rAs rCr&zParameters.client_propertiess &&rEct|ttdfstd|t j||_dS)z :param value: None or dict of client properties used to override the fields in the default client properties reported to RabbitMQ via `Connection.StartOk` method. Nz0client_properties must be dict or None, but got )rNdicttyperYcopydeepcopyr r[s rCr&zParameters.client_propertiessc%T "    5),1E455 5#'-"6"6rEc|jS)z :returns: number of socket connection attempts. Defaults to `DEFAULT_CONNECTION_ATTEMPTS`. See also `retry_delay`. :rtype: int )r rAs rCr(zParameters.connection_attemptss ((rEct|tjstd|dkrt d|||_dS)z| :param int value: number of socket connection attempts of at least 1. See also `retry_delay`. z"connection_attempts must be an intrz)connection_attempts must be > 0, but got N)rNrWr`rYrZr r[s rCr(zParameters.connection_attemptss_%!122 B@AA A 199*AFHJJ J$)!!!rEc|jS)z :rtype: one of the classes from `pika.credentials.VALID_TYPES`. Defaults to `DEFAULT_CREDENTIALS`. )rrAs rCr*zParameters.credentialss   rEct|ttjjs$t dtjjd|t j||_dS)z :param value: authentication credential object of one of the classes from `pika.credentials.VALID_TYPES` z'credentials must be an object of type: z , but got N) rNtuplerar* VALID_TYPESrYrhrirr[s rCr*zParameters.credentialsso%t'7'C!D!DEE N)(,(8(D(D(DeeMNN N!M%00rEc|jS)z :returns: desired maximum AMQP frame size to use. Defaults to `DEFAULT_FRAME_MAX`. :rtype: int )rrAs rCr,zParameters.frame_maxs rEc$t|tjstd||tjkrt dtj|fz|tjkrt dtj|fz||_dS)z :param int value: desired maximum AMQP frame size to use between `spec.FRAME_MIN_SIZE` and `spec.FRAME_MAX_SIZE`, inclusive z"frame_max must be an int, but got z+Min AMQP 0.9.1 Frame Size is %i, but got %rz+Max AMQP 0.9.1 Frame Size is %i, but got %rN) rNrWr`rYspecFRAME_MIN_SIZErZFRAME_MAX_SIZErr[s rCr,zParameters.frame_maxs%!122 O)eeMNN N 4& & &J#N T( ( (J#N  rEc|jS)aI :returns: AMQP connection heartbeat timeout value for negotiation during connection tuning or callable which is invoked during connection tuning. None to accept broker's value. 0 turns heartbeat off. Defaults to `DEFAULT_HEARTBEAT_TIMEOUT`. :rtype: int|callable|None )rrAs rCr.zParameters.heartbeat/s rEc|bt|tjs!t|st d|t|s|dkrt d|||_dS)a< :param int|None|callable value: Controls AMQP heartbeat timeout negotiation during connection tuning. An integer value always overrides the value proposed by broker. Use 0 to deactivate heartbeats and None to always accept the broker's proposal. If a callable is given, it will be called with the connection instance and the heartbeat timeout proposed by broker as its arguments. The callback should return a non-negative integer that will be used to override the broker's proposal. Nz9heartbeat must be an int or a callable function, but got rzheartbeat must >= 0, but got )rNrWr`callablerYrZrr[s rCr.zParameters.heartbeat;s  eW%566 x iu   E?? Ouqyy jee!MNNNrEc|jS)zn :returns: hostname or ip address of broker. Defaults to `DEFAULT_HOST`. :rtype: str )rrAs rCr0zParameters.hostOs zrEc>tj|d||_dS)zE :param str value: hostname or ip address of broker r0N) validatorsrequire_stringrr[s rCr0zParameters.hostXs" !%000 rEc|jS)z :returns: locale value to pass to broker; e.g., 'en_US'. Defaults to `DEFAULT_LOCALE`. :rtype: str )rrAs rCr2zParameters.localeas |rEc>tj|d||_dS)zR :param str value: locale value to pass to broker; e.g., "en_US" r2N)rzr{rr[s rCr2zParameters.localeks" !%222 rEc|jS)z :returns: port number of broker's listening socket. Defaults to `DEFAULT_PORT`. :rtype: int )rrAs rCr4zParameters.portts zrEc~ t||_dS#ttf$rtd|wxYw)zM :param int value: port number of broker's listening socket zport must be an int, but got N)intrrYrZr[s rCr4zParameters.port~sR  JUDJJJ:& J J J)HII I Js$<c|jS)z :returns: interval between socket connection attempts; see also `connection_attempts`. Defaults to `DEFAULT_RETRY_DELAY`. :rtype: float )rrAs rCr6zParameters.retry_delayr^rEclt|tjstd|||_dS)z :param int | float value: interval between socket connection attempts; see also `connection_attempts`. z,retry_delay must be a float or int, but got N)rNrWrXrYrr[s rCr6zParameters.retry_delaysF%.. M)DIEKMM M!rEc|jS)z :returns: socket connect timeout in seconds. Defaults to `DEFAULT_SOCKET_TIMEOUT`. The value None disables this timeout. :rtype: float|None )rrAs rCr8zParameters.socket_timeouts ##rEc|St|tjstd||dkrt d|t |}||_dS)z :param int | float | None value: positive socket connect timeout in seconds. None to disable this timeout. Nz/socket_timeout must be a float or int, but got rz$socket_timeout must be > 0, but got )rNrWrXrYrZfloatrr[s rCr8zParameters.socket_timeouts}  eW\22 9i05!8999zz j@EGIII%LLE$rEc|jS)z :returns: full protocol stack TCP/[SSL]/AMQP bring-up timeout in seconds. Defaults to `DEFAULT_STACK_TIMEOUT`. The value None disables this timeout. :rtype: float )rrAs rCr:zParameters.stack_timeouts ""rEc|St|tjstd||dkrt d|t |}||_dS)z :param int | float | None value: positive full protocol stack TCP/[SSL]/AMQP bring-up timeout in seconds. It's recommended to set this value higher than `socket_timeout`. None to disable this timeout. Nz.stack_timeout must be a float or int, but got rz#stack_timeout must be > 0, but got )rNrWrXrYrZrrr[s rCr:zParameters.stack_timeouts}  eW\22 9i05!8999zz j?DuFHHH%LLE#rEc|jS)z :returns: None for plaintext or `pika.SSLOptions` instance for SSL/TLS. :rtype: `pika.SSLOptions`|None )rrAs rCr<zParameters.ssl_options   rEct|ttdfstd|||_dS)z :param `pika.SSLOptions`|None value: None for plaintext or `pika.SSLOptions` instance for SSL/TLS. Defaults to None. Nz/ssl_options must be None or SSLOptions but got )rN SSLOptionsrgrYrr[s rCr<zParameters.ssl_optionssO%*d4jj!9:: P)GLuNPP P!rEc|jS)z| :returns: rabbitmq virtual host name. Defaults to `DEFAULT_VIRTUAL_HOST`. :rtype: str )rrAs rCr>zParameters.virtual_hosts !!rEc>tj|d||_dS)z? :param str value: rabbitmq virtual host name r>N)rzr{rr[s rCr>zParameters.virtual_hosts% !%888"rEc|jS)zp :returns: None or a dict of options to pass to the underlying socket :rtype: dict|None )rrAs rCr@zParameters.tcp_optionsrrEct|ttdfstd|||_dS)a :param dict|None value: None or a dict of options to pass to the underlying socket. Currently supported are TCP_KEEPIDLE, TCP_KEEPINTVL, TCP_KEEPCNT and TCP_USER_TIMEOUT. Availability of these may depend on your platform. Nz,tcp_options must be a dict or None, but got )rNrfrgrYrr[s rCr@zParameters.tcp_options sO%$T !344 M)DIEKMM M!rE)5rJ __module__ __qualname____doc__ __slots__DEFAULT_USERNAMEDEFAULT_PASSWORDr!rarbrcr#r%r*PlainCredentialsr)r'rrrtr+r-r/r1r3r5r7r9 DEFAULT_SSLr;DEFAULT_SSL_PORTr=r?rDrLrQrTpropertyr"setterr$r&r(r,r.r0r2r4r6r8r:r<r>r@rErCr r $s BI )-&,3 $*;;*,,"#+ $LNL! K444444l===  00X0 &11'&1"!!X! " " "''X' 7 7 7))X) * *  *!!X! 1 1 1X   (  X    &X [[X ]]X [JJ[J!!X! " " "$$X$%%%"##X#$$$&!!X! " " """X"###!!X! " " " " "rEr cdeZdZdZdZGddeZeeeeeeeeeeeeeeeeffd ZxZS)ConnectionParametersz^Connection parameters object that is passed into the connection adapter upon construction.rceZdZdZdS)ConnectionParameters._DEFAULTz0Designates default parameter value; internal useN)rJrrrrrErC_DEFAULTrs>>>>rErc tt|||jur||_||jur||_||jur||_| |jur| |_||jur||_||jur||_ ||jur||_ ||jur||_ | |jur| |_ | |jur| |_ | |jur| |_| |jur| |_||jur||_||jur||_n|jr|jn|j|_||jur||_||jur||_|rt-d|dS)a Create a new ConnectionParameters instance. See `Parameters` for default values. :param str host: Hostname or IP Address to connect to :param int port: TCP port to connect to :param str virtual_host: RabbitMQ virtual host to use :param pika.credentials.Credentials credentials: auth credentials :param int channel_max: Maximum number of channels to allow :param int frame_max: The maximum byte size for an AMQP frame :param int|None|callable heartbeat: Controls AMQP heartbeat timeout negotiation during connection tuning. An integer value always overrides the value proposed by broker. Use 0 to deactivate heartbeats and None to always accept the broker's proposal. If a callable is given, it will be called with the connection instance and the heartbeat timeout proposed by broker as its arguments. The callback should return a non-negative integer that will be used to override the broker's proposal. :param `pika.SSLOptions`|None ssl_options: None for plaintext or `pika.SSLOptions` instance for SSL/TLS. Defaults to None. :param int connection_attempts: Maximum number of retry attempts :param int|float retry_delay: Time to wait in seconds, before the next :param int|float socket_timeout: Positive socket connect timeout in seconds. :param int|float stack_timeout: Positive full protocol stack (TCP/[SSL]/AMQP) bring-up timeout in seconds. It's recommended to set this value higher than `socket_timeout`. :param str locale: Set the locale value :param int|float|None blocked_connection_timeout: If not None, the value is a non-negative timeout, in seconds, for the connection to remain blocked (triggered by Connection.Blocked from broker); if the timeout expires before connection becomes unblocked, the connection will be torn down, triggering the adapter-specific mechanism for informing client app about the closed connection: passing `ConnectionBlockedTimeout` exception to on_close_callback in asynchronous adapters or raising it in `BlockingConnection`. :param client_properties: None or dict of client properties used to override the fields in the default client properties reported to RabbitMQ via `Connection.StartOk` method. :param tcp_options: None or a dict of TCP options to set for socket zunexpected kwargs: N)superrrDrr"r$r&r(r*r,r.r0r2r6r8r:r<r4rr3r>r@rY)rBr0r4r>r*r$r,r.r<r(r6r8r:r2r"r&r@kwargsrIs rCrDzConnectionParameters.__init__ st "D))22444 %T] : :.HD + dm + +*D  DM 1 1%6D " dm 3 3':D $ dm + +*D  DM ) )&DN DM ) )&DN t} $ $DI  & & DK dm + +*D   . ."0D   - -!.D  dm + +*D  t} $ $DII151AX--tGXDI t} , , ,D  dm + +*D   A)vv?@@ @ A ArE) rJrrrrobjectrrD __classcell__rIs@rCrrsI?????6??? !   ( #"'/& #pApApApApApApApApApArErcreZdZdZdZdZfdZdZdZdZ dZ d Z d Z d Z d Zd ZdZdZdZxZS) URLParametersaConnect to RabbitMQ via an AMQP URL in the format:: amqp://username:password@host:port/[?query-string] Ensure that the virtual host is URI encoded when specified. For example if you are using the default "/" virtual host, the value should be `%2f`. See `Parameters` for default values. Valid query string values are: - channel_max: Override the default maximum channel count value - client_properties: dict of client properties used to override the fields in the default client properties reported to RabbitMQ via `Connection.StartOk` method - connection_attempts: Specify how many times pika should try and reconnect before it gives up - frame_max: Override the default maximum frame size for communication - heartbeat: Desired connection heartbeat timeout for negotiation. If not present the broker's value is accepted. 0 turns heartbeat off. - locale: Override the default `en_US` locale value - ssl_options: None for plaintext; for SSL: dict of public ssl context-related arguments that may be passed to :meth:`ssl.SSLSocket` as kwargs, except `sock`, `server_side`,`do_handshake_on_connect`, `family`, `type`, `proto`, `fileno`. - retry_delay: The number of seconds to sleep before attempting to connect on connection failure. - socket_timeout: Socket connect timeout value in seconds (float or int) - stack_timeout: Positive full protocol stack (TCP/[SSL]/AMQP) bring-up timeout in seconds. It's recommended to set this value higher than `socket_timeout`. - blocked_connection_timeout: Set the timeout, in seconds, that the connection may remain blocked (triggered by Connection.Blocked from broker); if the timeout expires before connection becomes unblocked, the connection will be torn down, triggering the connection's on_close_callback - tcp_options: Set the tcp options for the underlying socket. :param str url: The AMQP URL to connect to )_all_url_query_values _set_url_c tt|d|_|dddkr d|ddz}t j|}|jdkr,t j tj |_ n2|jdkrd|_ n|jrtd|jd |j |j|_|j |j|_n|j r|jn|j|_|jIt jt-|jt-|j|_t1|jd kr2t-|jd d |_t j|j|_t=|jD]\}} t?||j |z}n #tB$rtd |wxYw |\}n0#t$r#td |t1||fzwxYw||dS)zUCreate a new URLParameters instance. :param str url: The URL value Nramqphttphttps)contextzUnexpected URL scheme z&; supported scheme values: amqp, amqpsrrzUnknown URL parameter: zFExpected exactly one value for URL parameter %s, but got %i values: %s)"rrrDrlowerracompaturlparseschemersslcreate_default_contextr<rZhostnamer0r4rr3usernamer*rrpasswordlenpathsplitr> url_parse_qsqueryrgetattr_SETTER_PREFIXAttributeError)rBurlpartsnamer\ set_valuerIs rCrDzURLParameters.__init__s mT""++---%)" qs8>>  v % %3qrr7"C $$S)) <7 " " $244 6 6 6D   \V # ##D   \ F*6;lllEFF F > %DI : ! DII!,D..262C I > %#/@@EN++[-H-H J JD  uz??Q   +EJ,<, > >D   s =AAc8tj||_dSr)rrr@r[s rC_set_url_tcp_optionsz"URLParameters._set_url_tcp_optionss+E22rE)rJrrrrrrDrrrrrrrrrrrrrrs@rCrrs22j+I!NBBBBBH E E E ' ' '999 7 7 7 # # # + + + ' ' ' - - - + + +.>.>.>`3333333rErceZdZdZdZddZdS)rzdClass used to provide parameters for optional fine grained control of SSL socket wrapping. rNct|tjs"td|||_||_dS)z :param ssl.SSLContext context: SSLContext instance :param str|None server_hostname: SSLContext.wrap_socket, used to enable SNI z4context must be of ssl.SSLContext type, but got {!r}N)rNrrrYformatrr)rBrrs rCrDzSSLOptions.__init__sW '3>22 FMM  .rEr )rJrrrrrDrrErCrrs9 /I / / / / / /rErcheZdZdZdZdZdZdZdZdZ dZ d Z d Z d Z ed ed e de de de de diZ d`dZdZdZdZdZdZdadZdbdZdcdZddd Zed!Zed"Zed#Zed$Zed%Zed&Zed'Z e!j"d(Z#e!j"d)Z$e!j"d*Z%e!j"d+Z&e!j"d,Z'e!j"d-Z(d.Z)d/Z*d0Z+d1Z,ed2Z-d3Z.d4Z/d5Z0d6Z1d7Z2d8Z3d9Z4d:Z5d;Z6d<Z7d=Z8d>Z9d?Z:d@Z;dAZdDZ?dEZ@dFZAdGZBdHZCdIZDeEdJZFeEdKZGdLZHdMZIdNZJdOZKdPZLdQZMdRZNdSZO dbdTZPdUZQdVZRdWZSdXZTdYZUdcdZZVd[ZWd\ZXd]ZYd^ZZd_Z[dS)e ConnectionzThis is the core class that implements communication with RabbitMQ. This class should not be invoked directly but rather through the use of an adapter such as SelectConnection or BlockingConnection. _on_connection_closed_on_connection_error_on_connection_open_okrrrCLOSEDINITPROTOCOLSTARTTUNEOPENCLOSINGNTc0|j|_d|_d|_d|_d|_|H|j}d|_ tj||_ ||j _||_n#||_wxYwt|_ ||_ tj |_d|_d|_d|_d|_d|_d|_||jd|j|p|jd|r|||r||||j|j r|dSt> ddS)aeConnection initialization expects an object that has implemented the Parameters class and a callback function to notify when we have successfully connected to the AMQP Broker. Available Parameters classes are the ConnectionParameters class and URLParameters class. :param pika.connection.Parameters parameters: Read-only connection parameters. :param callable on_open_callback: Called when the connection is opened: on_open_callback(connection) :param None | method on_open_error_callback: Called if the connection can't be established or connection establishment is interrupted by `Connection.close()`: on_open_error_callback(Connection, exception). :param None | method on_close_callback: Called when a previously fully open connection is closed: `on_close_callback(Connection, exception)`, where `exception` is either an instance of `exceptions.ConnectionClosed` if closed by user or broker or exception of another type that describes the cause of connection failure. :param bool internal_connection_workflow: True for autonomous connection establishment which is default; False for externally-managed connection workflow via the `create_connection()` factory. FNrz#Using external connection workflow.)!CONNECTION_CLOSEDconnection_state_opened_error_blocked_conn_timer_heartbeat_checkerr<rhriparamsr_internal_connection_workflowracallbackCallbackManager callbacksserver_capabilitiesserver_properties_body_max_length known_hosts _frame_buffer _channels_init_connection_stateaddON_CONNECTION_ERROR_default_on_connection_erroradd_on_open_callbackadd_on_close_callback_set_connection_stateCONNECTION_INIT_adapter_connect_streamLOGGERdebug)rB parameterson_open_callbackon_open_error_callbackon_close_callbackinternal_connection_workflowsaved_ssl_optionss rCrDzConnection.__init__s>!% 6   $( "&  !!+ 6 %)J " ;"mJ77 *; '): &&): &::::.00DK-I*6688$( !% $! ##%%%  t')?*.  -u 6 6 6  8  % %&6 7 7 7  :  & &'8 9 9 9 ""4#7888  - @  ( ( * * * * * LL> ? ? ? ? ?s %A'' A0cD||jd|_t|_t |_d|_d|_d|_ d|_ d|_ d|_ | |jdt jj|j|jj^|j!||jd|_||j||jdSdS)zInitialize or reset all of the internal state variables for a given connection. On disconnect or reconnect all of the state needs to be wiped. Nr)r.rr#bytesr&rfr' bytes_sentbytes_received frames_sentframes_receivedrr_add_connection_start_callbackr!r)rrrClose _on_connection_close_from_brokerrr"r_adapter_remove_timeout"add_on_connection_blocked_callback_on_connection_blocked$add_on_connection_unblocked_callback_on_connection_unblockedrAs rCr(z!Connection._init_connection_stateFs2 ""4#9:::"&#WW "&  ++--- 1do3@ B B B ; 1 ='3,,T-EFFF+/(  3 3D4O P P P  5 5- / / / / / > =rEcrtj||jd|j|ddS)aAdd a callback notification when the connection has closed. The callback will be passed the connection and an exception instance. The exception will either be an instance of `exceptions.ConnectionClosed` if a fully-open connection was closed by user or broker or exception of another type that describes the cause of connection closure/failure. :param callable callback: Callback to call on close, having the signature: callback(pika.connection.Connection, exception) rFN)rzrequire_callbackr!r)ON_CONNECTION_CLOSEDrBrs rCr-z Connection.add_on_close_callback}s: #H--- 1d75IIIIIrEctj||jdtjjtj||ddS)aeRabbitMQ AMQP extension - Add a callback to be notified when the connection gets blocked (`Connection.Blocked` received from RabbitMQ) due to the broker running low on resources (memory or disk). In this state RabbitMQ suspends processing incoming data until the connection is unblocked, so it's a good idea for publishers receiving this notification to suspend publishing until the connection becomes unblocked. See also `Connection.add_on_connection_unblocked_callback()` See also `ConnectionParameters.blocked_connection_timeout`. :param callable callback: Callback to call on `Connection.Blocked`, having the signature `callback(connection, pika.frame.Method)`, where the method frame's `method` member is of type `pika.spec.Connection.Blocked` rFone_shotN) rzrHr!r)rrrBlocked functoolspartialrJs rCrCz-Connection.add_on_connection_blocked_callbacks\& #H---  O #  h - -      rEctj||jdtjjtj||ddS)aRabbitMQ AMQP extension - Add a callback to be notified when the connection gets unblocked (`Connection.Unblocked` frame is received from RabbitMQ) letting publishers know it's ok to start publishing again. :param callable callback: Callback to call on `Connection.Unblocked`, having the signature `callback(connection, pika.frame.Method)`, where the method frame's `method` member is of type `pika.spec.Connection.Unblocked` rFrLN) rzrHr!r)rrr UnblockedrOrPrJs rCrEz/Connection.add_on_connection_unblocked_callbacks\ #H---  O %  h - -      rEcrtj||jd|j|ddS)zAdd a callback notification when the connection has opened. The callback will be passed the connection instance as its only arg. :param callable callback: Callback to call when open rFN)rzrHr!r)ON_CONNECTION_OPEN_OKrJs rCr,zConnection.add_on_open_callbacks: #H--- 1d8(EJJJJJrEctj||r&|jd|j|j|jd|j|ddS)aAdd a callback notification when the connection can not be opened. The callback method should accept the connection instance that could not connect, and either a string or an exception as its second arg. :param callable callback: Callback to call when can't connect, having the signature _(Connection, Exception) :param bool remove_default: Remove default exception raising callback rFN)rzrHr!remover*r+r))rBrremove_defaults rCadd_on_open_error_callbackz%Connection.add_on_open_error_callbacksm #H---  E N ! !!T%="&"C E E E 1d6%HHHHHrEcP|jstjd|ztj||s|}||||j|<|||j| |j|S)aUCreate a new channel with the next available channel number or pass in a channel number to use. Must be non-zero if you would like to specify but it is recommended that you let Pika manage the channel numbers. :param int channel_number: The channel number to use, defaults to the next available. :param callable on_open_callback: The callback when the channel is opened. The callback will be invoked with the `Channel` instance as its only argument. :rtype: pika.channel.Channel z2Channel allocation requires an open connection: %s) is_open exceptionsConnectionWrongStateErrorrzrpc_completion_callback_next_channel_number_create_channelr'_add_channel_callbacksopenrBchannel_numberr4s rCrbzConnection.channels| M6DtKMM M *+;<<< 9!6688N)-)=)= ,*.*.~& ##N333 ~&++---~n--rEc|jstjd|ztj||dt j|||t jj gdS)aRabbitMQ AMQP extension - This method updates the secret used to authenticate this connection. It is used when secrets have an expiration date and need to be renewed, like OAuth 2 tokens. Pass a callback to be notified of the response from the server. :param string new_secret: The new secret :param string reason: The reason for the secret update :param callable callback: Callback to call on `Connection.UpdateSecretOk`, having the signature `callback(pika.frame.Method)`, where the method frame's `method` member is of type `pika.spec.Connection.UpdateSecretOk` :raises pika.exceptions.ConnectionWrongStateError: if connection is not open. z-Secret update requires an open connection: %srN) rZr[r\rzr]_rpcrrr UpdateSecretUpdateSecretOk)rB new_secretreasonrs rC update_secretzConnection.update_secrets| H6?$FHH H *8444 !T_11*fEET_;< > > > > >rENormal shutdownc$|js|jrVd||||j|j}t |tj||j r| |||j}| |j t d|||jset dtjd|j|||}||dStj|||_|j s|dSt dt)|j |dS)aDisconnect from RabbitMQ. If there are any open channels, it will attempt to close them prior to fully disconnecting. Channels which have active consumers will attempt to send a Basic.Cancel to RabbitMQ to cleanly stop the delivery of messages prior to closing the channel. :param int reply_code: The code number for the close :param str reply_text: The text reason for the close :raises pika.exceptions.ConnectionWrongStateError: if connection is closed or closing. zVIllegal close({}, {!r}) request on {} because it was called while connection state={}.zClosing connection (%s): %rzhConnection.close() is terminating stream and bypassing graceful AMQP close, since AMQP is still opening.zVConnection.close() called before connection finished opening: prev_state={} ({}): {!r}z8Connection.close is waiting for %d channels to close: %sN) is_closing is_closedr _STATE_NAMESrr1errorr[r\r'_close_channelsr.CONNECTION_CLOSINGinforConnectionOpenAborted_terminate_streamConnectionClosedByClientr_on_close_readyr)rB reply_code reply_textmsg prev_staterqs rCclosezConnection.closes ? ? 9  Z 8 8 8*  ""4#:;;; 1:zJJJ| / KK# $ $ $4==CV%j1:z>K>KLLE  " "5 ) ) ) ) )%=J((DK > /$$&&&&& N''/////rEc"|j|jkSzK Returns a boolean reporting the current connection state. )rrrAs rCrozConnection.is_closedGs $(>>>rEc"|j|jkS)z Returns True if connection is in the process of closing due to client-initiated `close` request, but closing is not yet complete. )rrsrAs rCrnzConnection.is_closingNs $(???rEc"|j|jkSr)rCONNECTION_OPENrAs rCrZzConnection.is_openVs $(<<.1DO 3t~ 3 :z222 3 3rEcztd|tj|||S)a|Create a new channel using the specified channel number and calling back the method specified by on_open_callback :param int channel_number: The channel number to use :param callable on_open_callback: The callback when the channel is opened. The callback will be invoked with the `Channel` instance as its only argument. zCreating channel %s)r1r2rarbChannelrbs rCr_zConnection._create_channel!s4  *N;;;|##D.:JKKKrEc|jj_|jjdkrOtd|jjtj||jjSdS)zCreate a heartbeat checker instance if there is a heartbeat interval set. :rtype: pika.heartbeat.Heartbeat|None NrzCreating a HeartbeatChecker: %r)rr.r1r2raHeartbeatCheckerrAs rC_create_heartbeat_checkerz$Connection._create_heartbeat_checker.sa ; ,1F1J1J LL:. 0 0 0>2249NOO OtrEcX|jr"|jd|_dSdS)z1Stop the heartbeat checker if it exists N)rstoprAs rC_remove_heartbeatzConnection._remove_heartbeat<s<  " +  # ( ( * * *&*D # # # + +rEc|j|jvr)td|j|j|dS|j|j|dS)zDeliver the frame to the channel specified in the frame. :param pika.frame.Method value: The frame to deliver z3Received %s frame for unregistered channel %i on %sN)rcr'r1criticalNAME_handle_content_framer[s rC_deliver_frame_to_channelz$Connection._deliver_frame_to_channelDsh #t~55 OOE E0$ 8 8 8 F u+,BB5IIIIIrEc@|jr|dSdS)z*If the connection is not closed, close it.N)rZr}rAs rC_ensure_closedzConnection._ensure_closedUs' <  JJLLLLL  rEcN|jjtjz tjz S)zaCalculate the maximum amount of bytes that can be in a body frame. :rtype: int )rr,rrFRAME_HEADER_SIZEFRAME_END_SIZErAs rC_get_body_frame_max_lengthz%Connection._get_body_frame_max_lengthZs$  %(>>#$ %rEc|jj|j\}}|s#t j|jjj|jj||fS)zGet credentials for authentication. :param pika.frame.MethodFrame method_frame: The Connection.Start frame :rtype: tuple(str, str) )rr* response_forrr[AuthenticationErrorTYPEerase_credentials)rB method_frame auth_typeresponses rC_get_credentialszConnection._get_credentialscsk[,99,:MNN   O01H1MNN N 11333(""rEcL|j|j|jS)zReturn true if there are any callbacks pending for the specified frame. :param pika.frame.Method value: The frame to check :rtype: bool )r!pendingrcrr[s rC_has_pending_callbacksz!Connection._has_pending_callbacksqs!~%%e&:ELIIIrEc6t|tjS)zReturns true if the frame is a method frame. :param pika.frame.Frame value: The frame to evaluate :rtype: bool )rNrMethodr[s rC_is_method_framezConnection._is_method_frame{s%...rEc6t|tjS)zMReturns True if it's a protocol header frame. :rtype: bool )rNrrr[s rC_is_protocol_header_framez$Connection._is_protocol_header_frames %!5666rEc*|jjptjj}t |j|krtjtdt |jdzD]}||jvr|cSt |jdzS)z^Return the next available channel number or raise an exception. :rtype: int r) rr$rarbrcrr'r[NoFreeChannelsr)rBlimitnums rCr^zConnection._next_channel_numbers  'D4<+D t~  % ' '+-- -!S001455  C$.(( )4>""Q&&rEc |j|j=td|jn0#t$r#td|jYnwxYw|jrZ|js|dSdt|jD}|rt d|dSdSdS)zRemove the channel from the dict of channels when Channel.CloseOk is sent. If connection is closing and no more channels remain, proceed to `_on_close_ready`. :param pika.channel.Channel channel: channel instance zRemoved channel %szChannel %r not in channelsc g|] }|j | Sr)rn).0rs rC z2Connection._on_channel_cleanup..s0111!?1111rEz8Connection in CLOSING state has non-CLOSING channels: %rN) r'rcr1r2KeyErrorrqrnrxrr)rBrbchannels_not_in_closing_states rCrzConnection._on_channel_cleanups Ow56 LL-w/E F F F F O O O LL5w7M N N N N N O ? G> G$$&&&&& 11%4T^%D%D111-1GOO'(EGGGGG G GGGs-0*AAc|jrtddS||jj|jjdS)zCalled when the Connection is in a state that it can close after a close has been requested by client. This happens after all of the channels are closed that were open when the close request was made. z+_on_close_ready invoked when already closedN)ror1warning_send_connection_closerryrzrAs rCrxzConnection._on_close_readysY >  NNH I I I F ##DK$:$(K$: < < < <  > > > >  ( ()A B B B'+D $ $ $rEctd||tj|jj|jjdS)zCalled when the connection is closed remotely via Connection.Close frame from broker. :param pika.frame.Method method_frame: The Connection.Close frame z*_on_connection_close_from_broker: frame=%sN)r1r2rvr[ConnectionClosedByBrokerrryrzrBrs rCrAz+Connection._on_connection_close_from_brokersc  A<PPP   / 0C0N0<0C0N P P Q Q Q Q QrEcftd||ddS)zCalled when Connection.CloseOk is received from remote. :param pika.frame.Method method_frame: The Connection.CloseOk frame z!_on_connection_close_ok: frame=%sN)r1r2rvrs rC_on_connection_close_okz"Connection._on_connection_close_ok s3  8,GGG t$$$$$rEc|)zDefault behavior when the connecting connection cannot connect and user didn't supply own `on_connection_error` callback. :raises: the given error r)rB_connection_unusedrqs rCr+z'Connection._default_on_connection_errors  rEcd|_|jj|_||j|jd|j||dS)z This is called once we have tuned the connection with the server and called the Connection.Open on the server and it has replied with Connection.Ok. TrN)rrr%r.rr!processrTrs rCr z!Connection._on_connection_open_oksY  '.: ""4#7888 q$" Both peers negotiate the limits to the lowest agreed value as follows: > - The server MUST tell the client what limits it proposes. > - The client responds and **MAY reduce those limits** for its connection If the client specifies a value, it always takes precedence. :param client_value: None to accept server_value; otherwise, an integral number in seconds; 0 (zero) to disable heartbeat. :param server_value: integral value of the heartbeat timeout proposed by broker; 0 (zero) to disable heartbeat. :returns: the value of the heartbeat timeout to use and return to broker :rtype: int r)rrtimeouts rC_tune_heartbeat_timeoutz"Connection._tune_heartbeat_timeout]s*  "GG"GrEc||jt|jj|jj|j_t|jj|jj|j_t|jj rT|j ||jj }|t|rtd|||j_ | |jj |jj |j_ | |_ ||_||dS)aFOnce the Broker sends back a Connection.Tune, we will set our tuning variables that have been returned to us and kick off the Heartbeat monitor if required, send our TuneOk and then the Connection. Open rpc call on channel 0. :param pika.frame.Method method_frame: The frame received Nz=heartbeat callback must not return None or callable, but got )rr)r.CONNECTION_TUNErrrr$rr,rwr.rYrrr$rr_send_connection_tune_ok_send_connection_open)rBr ret_heartbeats rCrzConnection._on_connection_tunezs} ""4#7888#-"E"E K #\%8%D#F#F  * C C K !<#6#@!B!B  DK) * * 2 K11$2>2E2OQQM$(?(?$i=J]!MNNN%2DK !!% < <.%,6!=!8!8  !% ? ? A A#'"@"@"B"B %%''' ""$$$$$rEc|xj|z c_|jrN|\}}|sdS|||||jLdSdS)zThis is called by our Adapter, passing in the data from the socket. As long as we have buffer try and map out frame data. :param str data_in: The data that is available to read N)r& _read_frame_trim_frame_buffer_process_frame)rBdata_inconsumed_count frame_values rC_on_data_availablezConnection._on_data_availables g%  -*.*:*:*<*< 'NK   # #N 3 3 3    , , ,   - - - - -rEct|tdtfsJd||||_n|j Jd||dS)aDeactivate heartbeat instance if activated already, and initiate termination of the stream (TCP) connection asynchronously. When connection terminates, the appropriate user callback will be invoked with the given error: "on open error" or "on connection closed". :param Exception | None error: exception instance describing the reason for termination; None for normal closing, such as upon receipt of Connection.CloseOk. Nz:error arg is neither None nor instance of Exception: {!r}.zN_terminate_stream() expected self._error to be set when passed None error arg.)rNrgr rrrr)rBrqs rCrvzConnection._terminate_streams%$t**i!899   H O O       DKK;**)+**     '')))))rEctd|j||j|0|j!td|j|||_n|j Jd||dtjj tjj g|j j 6|dtjj tjjg|jsit|jt jt jfr=|j|jkrFtdt!jt/|j|_n|j|jkrFtdt!jt/|j|_n|j|jkrFtdt!jt/|j|_n;|j|j|j|jfvr td |j| |jtC|j"D]1}||j"vr |j"|#|j2|jsItd |j|j$%d|j&|||jnHtd |j|j$%d|j'|||j|(dS) aHandle termination of stack (including TCP layer) or failure to establish the stack. Notify registered ON_CONNECTION_ERROR or ON_CONNECTION_CLOSED callbacks, depending on whether the connection was opening or open. :param Exception | None error: None means that the transport was aborted internally and exception in `self._error` represents the cause. Otherwise it's an exception object that describes the unexpected loss of connection. z_AMQP stack terminated, failed to connect, or aborted: opened=%r, error-arg=%r; pending-error=%rNz<_on_stream_terminated(): overriding pending-error=%r with %rz__on_stream_terminated() expected self._error to be populated with reason for terminating stack.rz'Probably incompatible Protocol VersionszQConnection closed while authenticating indicating a probable authentication errorzrConnection closed while tuning the connection indicating a probable permission error when accessing a virtual hostz-Unexpected connection state on disconnect: %iz%Connection setup terminated due to %rzStack terminated due to %r))r1rtrrr2r_remove_callbacksrrrr@rrr"rNrRrNr[StreamLostErrorrrrrqIncompatibleProtocolErrorreprrProbableAuthenticationErrorrProbableAccessDeniedErrorrrrsrr.rr'_on_close_metar!rr*rIr()rBrqrbs rC_on_stream_terminatedz Connection._on_stream_terminateds~   8 L% . . .  {& /04 UDDD DKK;**5+**     q $ 5t7LM O O O ; 1 =  " "1_,do.GH J J J| 6 4;+Z-PQ!S!S 6$(@@@ FGGG(B%%'' &$*??? 4555)D%%'' &$*>>> 8999)B%%'' &($*@+/N#4666 ""4#9::: // @ @Gdn,, N7 # 2 24; ? ? ? ?| 0 KK? M M M N " "1d&>d#'; 0 0 0 0 KK4dk B B B N " "1d&?t#'; 0 0 0 ##%%%%%rEc||r>||r)|j|j|j||dSdS)zProcess the callbacks for the frame if the frame is a method frame and if it has any callbacks pending. :param pika.frame.Method frame_value: The frame to process :rtype: bool TF)rrr!rrcrrBr!s rC_process_callbackszConnection._process_callbacks-sh  ! !+ . . ++K88  N " "*"     4urEcb|jdkrdS|xjdz c_||rdSt|tjr>|jr|jdSt ddS|j dkr| |dSdS)zProcess an inbound frame from the socket. :param pika.frame.Frame|pika.frame.Method frame_value: The frame to process rNrz4Received heartbeat frame without a heartbeat checker) frame_typer>r/rNr Heartbeatrreceivedr1rrcrr.s rCrzConnection._process_frame?s  !A % % F !  " "; / /  F k5? 3 3 8& *'0022222 )***** '! + +  * *; 7 7 7 7 7, +rEc4tj|jS)zoTry and read from the frame buffer and decode a frame. :rtype tuple: (int, pika.frame.Frame) )r decode_framer&rAs rCrzConnection._read_frame]s !$"4555rEc`|D]*}|jt||+dS)a5Remove the callbacks for the specified channel number and list of method frames. :param int channel_number: The channel number to remove the callback on :param sequence method_classes: The method classes (derived from `pika.amqp_object.Method`) for the callbacks N)r!rVr)rBrcmethod_classes method_clss rCr%zConnection._remove_callbacksesD) C CJ N ! !#n"5"5z B B B B C CrEc|r$t|tstd|5tj||D]}|j||||||dS)aMake an RPC call for the given callback, channel number and method. acceptable_replies lists out what responses we'll process from the server with the specified callback. :param int channel_number: The channel number for the RPC call :param pika.amqp_object.Method method: The method frame to call :param callable callback: The callback for the RPC response :param list acceptable_replies: The replies this RPC call expects z)acceptable_replies should be list or NoneN)rNlistrYrzrHr!r) _send_method)rBrcrracceptable_repliesreplys rCrezConnection._rpcqs  Ij1CT&J&J IGHH H    ' 1 1 1+ D D"">5(CCCC .&11111rEc |dtj||dd|jtjjgdS)zSend a Connection.Close method frame. :param int reply_code: The reason for the close :param str reply_text: The text reason for the close rN)rerrrr@rCloseOk)rBryrzs rCrz!Connection._send_connection_closesS !T_**:z1aHH.1H0I K K K K KrEc|dtj|jjd|jtjjgdS)zSend a Connection.Open framerT)insistN)rerrrOpenrr>r OpenOkrAs rCrz Connection._send_connection_opensX !T_)) K $T*33484O?)* , , , , ,rEc |dtj|j|||jjdS)zSend a Connection.StartOk frame :param str authentication_type: The auth type value :param str response: The encoded value to send rN)r;rrrStartOkr rr2)rBauthentication_typers rCr z$Connection._send_connection_start_oksP  O # #D$;$7$(K$6 8 8 9 9 9 9 9rEc|dtj|jj|jj|jjdS)zSend a Connection.TuneOk framerN)r;rrrTuneOkrr$r,r.rAs rCrz#Connection._send_connection_tune_oksS  O " "4;#:#';#8#';#8 : : ; ; ; ; ;rEc|jr.tdtjd|}||gdS)a+This appends the fully generated frame to send to the broker to the output buffer which will be then sent via the connection adapter. :param pika.frame.Frame|pika.frame.ProtocolHeader frame_value: The frame to write :raises: exceptions.ConnectionClosed z#Attempted to send frame when closedz/Attempted to send a frame on closed connection.N)ror1rqr[r\marshal_output_marshaled_frames)rBr!marshaled_frames rCrzConnection._send_framesm > C LL> ? ? ?6ACC C&--// %%&788888rEc|r||||dS|tj||dS)aHConstructs a RPC method frame and then sends it to the broker. :param int channel_number: The channel number for the frame :param pika.amqp_object.Method method: The method to send :param tuple content: If set, is a content frame, is tuple of properties and body. N) _send_messagerrr)rBrcrcontents rCr;zConnection._send_methodsQ  C   ~vw ? ? ? ? ?   U\.&AA B B B B BrEct|d}g}tj||}tj|||d}|||||drt tjt||j z }td|D]h} | |j z} | |j z} | |kr|} tj ||d| | } || i| |dS)aPublish a message. :param int channel_number: The channel number for the frame :param pika.object.Method method_frame: The method frame to send :param tuple content: A content frame, which is tuple of properties and body. rrN)rrrHeaderappendrJrmathceilrr$rBodyrK) rBrcrrOlengthmarshaled_body_frames frame_method frame_headerchunkschunkstartend frame_bodys rCrNzConnection._send_messagesTWQZ "|NLAA |NFGAJGG $$\%9%9%;%;<<<$$\%9%9%;%;<<< 1: C5==43H#HIIJJF6** C C 55d33<< C"Z 598MNN %,,Z-?-?-A-ABBBB %%&;<<<< : : 7&~666 7 7rEcN|j|d|_|xj|z c_dS)zTrim the leading N bytes off the frame buffer and increment the counter that keeps track of how many bytes have been read/used from the socket. :param int byte_count: The number of bytes consumed N)r&r<)rB byte_counts rCrzConnection._trim_frame_buffer s2"/ < z)rEc|D]D}|xjt|z c_|xjdz c_||EdS)zOutput list of marshaled frames to buffer and update stats :param list marshaled_frames: A list of frames marshaled to bytes rN)r;rr=r)rBmarshaled_framesrLs rCrKz#Connection._output_marshaled_frames s` 0 5 5O OOs?33 3OO    !    # #O 4 4 4 4 5 5rE)NNNNT)T)NNr )rkrl)\rJrrrrIr*rTrr/rrrrrsrprDr(r-rCrEr,rXrbrjr}rrornrZrrrrabcabstractmethodrrBrr0rrr`r?rrr rrr_rrrrrrrrrr^rrxrrrDrFrArr+r r staticmethodrrrr"rvr,r/rrr%rerrr rrr;rNr.rrrKrrErCrrs+304OOO 8Z'IL!"&(,#'.2 d@d@d@d@L5/5/5/n J J J4$KKKIIII"....:>>>>.;/;/;/;/B??X? @@X@==X=AAXAMMXMQQXQIIXI  " " " """  " " "   " " " """  " " " & & &PPPNNN < < <X2 3 3 3 L L L   +++JJJ" %%% # # #JJJ///777 ' ' 'GGG< < < <111888555&,,," Q Q Q%%%JJJ ****\0\8+%+%+%Z--- *** 1 1 1 7 7 7 * * * 5 5 5 5 5rEr))rrfrrhrOloggingrSrWrr pika.callbackra pika.channel pika.compatpika.credentialspika.exceptionsr[ pika.framerpika.heartbeat pika.specrrpika.validatorsrzrrrrrr getLoggerrJr1rr rrrr AbstractBaserrrErCrusR   $$$$$$$$$$$$ '  8 $ $o"o"o"o"o"o"o"o"dzAzAzAzAzA:zAzAzAzS3S3S3S3S3JS3S3S3l////////0[5[5[5[5[5)[5[5[5[5[5rE