)Jf5dZddlZddlZddlZddlZddlZddlZddlmZej e Z Gdde Z Gdde ZGdd e ZGd d e ZGd d e ZGddeZGddeZGddeZGdde ZGdde ZGdde ZGddeZGddejjZGddeZdS) a,Implements `AMQPConnectionWorkflow` - the default workflow of performing multiple TCP/[SSL]/AMQP connection attempts with timeouts and retries until one succeeds or all attempts fail. Defines the interface `AbstractAMQPConnectionWorkflow` that facilitates implementing custom connection workflows. N) __version__ceZdZdZdS)AMQPConnectorExceptionzBase exception for this moduleN__name__ __module__ __qualname____doc__j/home/alex/cs2snipeproduction/venv/lib/python3.11/site-packages/pika/adapters/utils/connection_workflow.pyrrs((((r rceZdZdZdS)AMQPConnectorStackTimeoutz:Overall TCP/[SSL]/AMQP stack connection attempt timed out.Nrr r r rrsDDDDr rceZdZdZdS)AMQPConnectorAbortedz Asynchronous request was abortedNrr r r rrs****r rceZdZdZdS)AMQPConnectorWrongStatezjAMQPConnector operation requested in wrong state, such as aborting after completion was reported. Nrr r r rr"r rc(eZdZdZfdZdZxZS)AMQPConnectorPhaseErrorBasezMWrapper for exception that occurred during a particular bring-up phase. cPtt|j|||_dS)z :param BaseException exception: error that occurred while waiting for a subclass-specific protocol bring-up phase to complete. :param args: args for parent class N)superr__init__ exception)selfrargs __class__s r rz$AMQPConnectorPhaseErrorBase.__init__-s+ :)40094@@"r cLd|jj|jS)Nz{}: {!r})formatrrrrs r __repr__z$AMQPConnectorPhaseErrorBase.__repr__7s  !8$.IIIr rrr r rr! __classcell__rs@r rr(sX#####JJJJJJJr rceZdZdZdS)AMQPConnectorSocketConnectErrorz*Error connecting TCP socket to remote peerNrr r r r&r&;s4444r r&ceZdZdZdS) AMQPConnectorTransportSetupErrorzOError setting up transport after TCP connected but before AMQP handshake. Nrr r r r(r(?rr r(ceZdZdZdS)AMQPConnectorAMQPHandshakeErrorzError during AMQP handshakeNrr r r r*r*Es%%%%r r*ceZdZdZdS)AMQPConnectionWorkflowAbortedz%AMQP Connection workflow was aborted.Nrr r r r,r,Is////r r,ceZdZdZdS) AMQPConnectionWorkflowWrongStatezuAMQP Connection Workflow operation requested in wrong state, such as aborting after completion was reported. Nrr r r r.r.Mrr r.c(eZdZdZfdZdZxZS)AMQPConnectionWorkflowFailedz5Indicates that AMQP connection workflow failed. cjtt|j|t||_dS)z :param sequence exceptions: Exceptions that occurred during the workflow. :param args: args to pass to base class N)rr0rtuple exceptions)rr3rrs r rz%AMQPConnectionWorkflowFailed.__init__Xs3 ;*D11:DAA ++r cd|jjt|j|jdt|jdkr |jdndS)NzG{}: {} exceptions in all; last exception - {!r}; first exception - {!r}r)rrrlenr3r s r r!z%AMQPConnectionWorkflowFailed.__repr__bs_#$*FN+S-A-AOB'*-do*>*>*B*BDOA&&%N%N Nr r"r$s@r r0r0SsX,,,,,NNNNNNNr r0creZdZdZdZdZdZdZdZdZ dZ d Z d Z d Z d Zd ZdZdZdZdZdZddZdS) AMQPConnectorz;Performs a single TCP/[SSL]/AMQP connection workflow. rr6c||_||_d|_d|_d|_d|_d|_d|_d|_d|_ |j |_ dS)a :param callable conn_factory: A function that takes `pika.connection.Parameters` as its only arg and returns a brand new `pika.connection.Connection`-based adapter instance each time it is called. The factory must instantiate the connection with `internal_connection_workflow=False`. :param pika.adapters.utils.nbio_interface.AbstractIOServices nbio: N) _conn_factory_nbio _addr_record _conn_params_on_done_tcp_timeout_ref_stack_timeout_ref _task_ref_sock _amqp_conn _STATE_INIT_state)r conn_factorynbios r rzAMQPConnector.__init__ws_*    !%"& & r c|j|jkr'td|j||_||_||_|j|_tj|jdd|_ |j tj j tjdtj|jj|j |j d|jd}t&dt*||j|j ||j|_d|_|jj/|j|jj|j|_d|_|jj1|j|jj|j |_dSdS) aAsynchronously perform a single TCP/[SSL]/AMQP connection attempt. :param tuple addr_record: a single resolved address record compatible with `socket.getaddrinfo()` format. :param pika.connection.Parameters conn_params: :param callable on_done: Function to call upon completion of the workflow: `on_done(pika.connection.Connection | BaseException)`. If exception, it's going to be one of the following: `AMQPConnectorSocketConnectError` `AMQPConnectorTransportSetupError` `AMQPConnectorAMQPHandshakeError` `AMQPConnectorAborted` )Already in progress or finished; state={}Nr;r6Fr<z Pika version %s connecting to %r)on_done)!rKrJrrrBrCrD _STATE_TCPsocketrH setsockoptpikacompatSOL_TCP TCP_NODELAYtcp_socket_opts set_sock_opts tcp_options setblocking_LOGinforrAconnect_socket_on_tcp_connection_donerGrEsocket_timeout call_later_on_tcp_connection_timeoutrF stack_timeout_on_overall_timeout)r addr_record conn_paramsrPaddrs r startzAMQPConnector.starts ;$* * *);BB4;OOQQ Q(' o ]D$5bqb$9:  dk163EqIII **4+<+H+/: 7 7 7 u%%% # 4k4HHH22 Jd&B3DD!%   + 7$(J$9$9!0/%1%1D ! #'   * 6&*j&;&;!/1I'K'KD # # # 7 6r c|j|jkrtd|j|jkrtd|j|_|t d|jj |j |j Zt d|j tj|jt%dS|j js7t d|j dddSt d |j|jks"Jd |jdS) aAbort the workflow asynchronously. The completion callback will be called with an instance of AMQPConnectorAborted. NOTE: we can't cancel/close synchronously because aborting pika Connection and its transport requires an asynchronous operation. :raises AMQPConnectorWrongState: If called after completion has been reported or the workflow not started yet. Cannot abort before starting.*Cannot abort after completion was reportedzCAMQPConnector: beginning client-initiated asynchronous abort; %r/%sNzXAMQPConnector.abort(): no connection, so just scheduling completion report via I/O loop.z*AMQPConnector.abort(): closing Connection.@z3Client-initiated abort of AMQP Connection Workflow.zCAMQPConnector.abort(): closing of Connection was already initiated.z9Connection is closing, but not in TIMEOUT state; state={})rKrJr _STATE_DONE_STATE_ABORTING _deactivater\r]rChostrBrIdebugrAadd_callback_threadsafe functoolspartial_report_completion_and_cleanupr is_closingclose_STATE_TIMEOUTrr s r abortzAMQPConnector.aborts ;$* * *)*IJJ J ;$* * *)*VWW W*     -2D4E G G G ? " JJD E E E J . .!$"E"6"8"8:: ; ; ; ; ;?- + GHHH%%NPPPPP  0111{d&9999PfT[)):9999r c||j |jd|_d|_d|_d|_d|_|j|_dS)zqCancel asynchronous tasks and clean up to assist garbage collection. Transition to STATE_DONE. N) rorHrwr@rArBrDrmrKr s r _closezAMQPConnector._closese  : ! J     DJ!   & r cF|j"Jd|j|j |jd|_|j |jd|_|j"|jd|_dSdS)$Cancel asynchronous tasks. Nz:_deactivate called with self._amqp_conn not None; state={})rIrrKrEcancelrFrGr s r rozAMQPConnector._deactivates&& H O O   '&&  ,  ! ( ( * * *$(D !  " .  # * * , , ,&*D # > % N ! ! # # #!DNNN & %r ct|trtd|ntd||j}|||dS)zClean up and invoke client's `on_done` callback. :param pika.connection.Connection | BaseException result: value to pass to user's `on_done` callback. z%AMQPConnector - reporting failure: %rz%AMQPConnector - reporting success: %rN isinstance BaseExceptionr\errorr]rDr{rresultrPs r ruz,AMQPConnector._report_completion_and_cleanupsi fm , , G JJ> G G G G II=v F F F- r cd|_ttjd|jj|j}||dS)ztHandle TCP connection timeout Reports AMQPConnectorSocketConnectError with socket.timeout inside. Nz)TCP connection attempt timed out: {!r}/{}) rEr&rRtimeoutrrCrprBru)rrs r rbz(AMQPConnector._on_tcp_connection_timeout,sg !%/ NFMM!&(9;; < <== ++E22222r c 4d|_|j}|j|_||jkrd|jj|jt|jj }t ||j j r"Jd|j |j js|j d|dS||jkr@t#t%d|jj|j}nd||jksJt)t%d|jj|jt|jj }||dS)aHandle overall TCP/[SSL]/AMQP connection attempt timeout by reporting `Timeout` error to the client. Reports AMQPConnectorSocketConnectError if timeout occurred during socket TCP connection attempt. Reports AMQPConnectorTransportSetupError if timeout occurred during tramsport [SSL] setup attempt. Reports AMQPConnectorAMQPHandshakeError if timeout occurred during AMQP handshake. Nz0Timeout while setting up AMQP to {!r}/{}; ssl={}zUnexpected open state of {!r}rlz*Timeout while connecting socket to {!r}/{}z5Timeout while setting up transport to {!r}/{}; ssl={})rFrKrx _STATE_AMQPrrCrprBbool ssl_optionsr\rrIis_openrvrwrQr&r_STATE_TRANSPORTr(ru)r prev_statemsgrs r rdz!AMQPConnector._on_overall_timeout9s#'[ ) ) ) )ELL!&(9T&23355C JJsOOO. H H/66tGG H H H?- 0%%c3/// F  ( (3)@GG).0ACCDDEEEE !666664)KF4,143D 1 =>>@@AABBE ++E22222r cld|_|j |jd|_|Etd||j|t|dStd|j |j |_ dx}}|j j 0|j j j}|j j j}| |j j}|jt'j|j|j |j |||j|_d|_ dS)aHandle completion of asynchronous socket connection attempt. Reports AMQPConnectorSocketConnectError if TCP socket connection failed. :param None|BaseException exc: None on success; exception object on failure Nz*TCP Connection attempt failed: %r; dest=%rz)TCP connection to broker established: %r.)protocol_factorysock ssl_contextserver_hostnamerP)rGrEr~r\rrBrur&rqrHrrKrCrcontextrrprAcreate_streaming_connectionrsrtr@ _on_transport_establishment_done)rexcrrs r r_z%AMQPConnector._on_tcp_connection_donegsA  ,  ! ( ( * * *$(D ! ? JJCS( * * *  / //44 6 6 6 F > KKK+ (,, o   ( 4+7?K"/;KO&"&"3"8??&.t/A/3/@BB#+9 @;; r c d|_t|trhtd||jj|jt|jj | t|dSt d||\}|_ |j|_|j |jd|j |jdS)aQHandle asynchronous completion of `AbstractIOServices.create_streaming_connection()` Reports AMQPConnectorTransportSetupError if transport ([SSL]) setup failed. :param sequence|BaseException result: On success, a two-tuple (transport, protocol); on failure, exception instance. NzCAttempt to create the streaming transport failed: %r; %r/%s; ssl=%sz"Streaming transport linked up: %r.T)remove_default)rGrrr\rrCrprBrrrur(r]rIrrKadd_on_open_error_callback_on_amqp_handshake_doneadd_on_open_callback)rr _transports r rz.AMQPConnector._on_transport_establishment_dones fm , ,  JJ !'):)?!4(9(E#F#F H H H  / /088 : : : F 6???&,# DO&  22  ( 3 ? ? ? ,,T-IJJJJJr Nc td|j||jj|jd|_|j|jkrt}n |j|j krXttd |jj|jt|jj}n|j|jkrm|/td|jj|j||}n`td|jj|j|t|}n$td|j||dS||dS)aHandle completion of AMQP connection handshake attempt. NOTE: we handle two types of callbacks - success with just connection arg as well as the open-error callback with connection and error Reports AMQPConnectorAMQPHandshakeError if AMQP handshake failed. :param pika.connection.Connection connection: :param BaseException | None error: None on success, otherwise failure zJAMQPConnector: AMQP handshake attempt completed; state=%s; error=%r; %r/%sNz,Timeout during AMQP handshake{!r}/{}; ssl={}z8AMQPConnector: AMQP connection established for %r/%s: %rz=AMQPConnector: AMQP connection handshake failed for %r/%s: %rzgAMQPConnector: Ignoring AMQP handshake completion notification due to wrong state=%s; error=%r; conn=%r)r\rqrKrCrprBrIrnrrxr*rrrrrru)r connectionrrs r rz%AMQPConnector._on_amqp_handshake_dones  #{E43D3I       ;$. . .)++FF [D/ / /4)BII).0AT.:;;==>>??FF [D, , ,} N%*D,=zKKK$  !%!2!79J9?? JJH UJ 0 0 0 F ++F33333r )N)rrr r rJrQrrrxrnrmrrhryr{rorurbrdr_rrr r r r9r9jsKJKNOK'''8/K/K/Kb,+,+,+\'''&""". 3 3 3,3,3,3\+++Z K K KD343434343434r r9ceZdZdZdZdZdS)AbstractAMQPConnectionWorkflowzMInterface for implementing a custom TCP/[SSL]/AMQP connection workflow. ct)aAsynchronously perform the workflow until success or all retries are exhausted. Called by the adapter. :param sequence connection_configs: A sequence of one or more `pika.connection.Parameters`-based objects. Will attempt to connect using each config in the given order. :param callable connector_factory: call it without args to obtain a new instance of `AMQPConnector` for each connection attempt. See `AMQPConnector` for details. :param native_loop: Native I/O loop passed by app to the adapter or obtained by the adapter by default. :param callable on_done: Function to call upon completion of the workflow: `on_done(pika.connection.Connection | AMQPConnectionWorkflowFailed | AMQPConnectionWorkflowAborted)`. `Connection`-based adapter on success, `AMQPConnectionWorkflowFailed` on failure, `AMQPConnectionWorkflowAborted` if workflow was aborted. :raises AMQPConnectionWorkflowWrongState: If called in wrong state, such as after starting the workflow. NotImplementedError)rconnection_configsconnector_factory native_looprPs r rhz$AbstractAMQPConnectionWorkflow.starts 2"!r ct)aAbort the workflow asynchronously. The completion callback will be called with an instance of AMQPConnectionWorkflowAborted. NOTE: we can't cancel/close synchronously because aborting pika Connection and its transport requires an asynchronous operation. :raises AMQPConnectionWorkflowWrongState: If called in wrong state, such as before starting or after completion has been reported. rr s r ryz$AbstractAMQPConnectionWorkflow.abort s "!r N)rrr r rhryr r r rrs<"""6 " " " " "r rceZdZdZejZejZdZ dZ dZ dZ ddZ dZd Zd Zd Zd Zd ZdZdZdZdZdZdS)AMQPConnectionWorkflowaImplements Pika's default workflow for performing multiple TCP/[SSL]/AMQP connection attempts with timeouts and retries until one succeeds or all attempts fail. The workflow: while not success and retries remain: 1. For each given config (pika.connection.Parameters object): A. Perform DNS resolution of the config's host. B. Attempt to establish TCP/[SSL]/AMQP for each resolved address until one succeeds, in which case we're done. 2. If all configs failed but retries remain, resume from beginning after the given retry pause. NOTE: failure of DNS resolution is equivalent to one cycle and will be retried after the pause if retries remain. rr6r:r;Fcd|_d|_||_d|_d|_d|_d|_d|_d|_d|_ d|_ g|_ |j |_ dS)a :param int | float retry_pause: Non-negative number of seconds to wait before retrying the config sequence. Meaningful only if retries is greater than 0. Defaults to 2 seconds. :param bool _until_first_amqp_attempt: INTERNAL USE ONLY; ends workflow after first AMQP handshake attempt, regardless of outcome (success or failure). The automatic connection logic in `pika.connection.Connection` enables this because it's not designed/tested to reset all state properly to handle more than one AMQP handshake attempt. TODO: Do we need getaddrinfo timeout? TODO: Would it be useful to implement exponential back-off? N)_attempts_remaining _retry_pause_until_first_amqp_attemptrA_current_config_index_connection_configs_connector_factoryrD _connectorrG_addrinfo_iter_connection_errorsrJrK)rrs r rzAMQPConnectionWorkflow.__init__3st $(  )B& &*"#' "& "#%& r c||_dS)aCalled by the conneciton adapter only on pika's `AMQPConnectionWorkflow` instance to provide it the adapter-specific `AbstractIOServices` object before calling the `start()` method. NOTE: Custom workflow implementations should use the native I/O loop directly because `AbstractIOServices` is private to Pika implementation and its interface may change without notice. :param pika.adapters.utils.nbio_interface.AbstractIOServices nbio: N)rA)rrMs r set_io_servicesz&AMQPConnectionWorkflow.set_io_services]s r cv|j|jkr'td|j t |n4#t $r'}t d|d}~wwxYw|s"td|||_||_ ||_ |dj |_ |dj |_|j|_t d|jdt)j|jd |_dS) afOverride `AbstractAMQPConnectionWorkflow.start()`. NOTE: This implementation uses `connection_attempts` and `retry_delay` values from the last element of the given `connection_configs` sequence as the overall number of connection attempts of the entire `connection_configs` sequence and pause between each sequence. rOz3connection_configs does not support iteration: {!r}Nz"connection_configs is empty: {!r}.r5z1Starting AMQP Connection workflow asynchronously.rTfirst)rKrJrriter Exception TypeError ValueErrorrrrDconnection_attemptsr retry_delayr _STATE_ACTIVEr\rqrArarsrt_start_new_cycle_asyncrG)rrrrrPrs r rhzAMQPConnectionWorkflow.startks[ ;$* * *);BB4;OOQQ Q  # $ $ $ $   ELL  " Q4;;(  FGGG.. y !> >*   ( ) ) ) ? " JJD E E E J . .!$"E"?"A"ACC D D D D D JJ, - - - O ! ! # # # # #r c|d|_d|_d|_d|_d|_d|_d|_|j|_ dS)zrCancel asynchronous tasks and clean up to assist garbage collection. Transition to _STATE_DONE. N) rorrArrDrrrrmrKr s r r{zAMQPConnectionWorkflow._closesY #'  "& ""&& r cX|j"|jd|_dSdS)r}N)rGr~r s r roz"AMQPConnectionWorkflow._deactivates3 > % N ! ! # # #!DNNN & %r ct|trtd|ntd||j}|||dS)zClean up and invoke client's `on_done` callback. :param pika.connection.Connection | AMQPConnectionWorkflowFailed result: value to pass to user's `on_done` callback. z.AMQPConnectionWorkflow - reporting failure: %rz.AMQPConnectionWorkflow - reporting success: %rNrrs r ruz5AMQPConnectionWorkflow._report_completion_and_cleanupsi fm , , P JJG P P P P IIF O O O- r cd|_|jdksJ|j|jdkrFt|j}td|||dS|xjdzc_td|jd|_|j |rdn|j |j |_dS)aQStart a new workflow cycle (if any more attempts are left) beginning with the first Parameters object in self._connection_configs. If out of attempts, report `AMQPConnectionWorkflowFailed`. :param bool first: if True, don't delay; otherwise delay next attempt by `self._retry_pause` seconds. Nrz$AMQP connection workflow failed: %r.r6zQBeginning a new AMQP connection workflow cycle; attempts remaining after this: %s) rGrr0rr\rrurqrrArar_try_next_config_async)rrrs r rz-AMQPConnectionWorkflow._start_new_cycle_asyncs'1,,,d.F,,,  #q ( (01HIIE JJ=u E E E  / / 6 6 6 F   A%    '(,(@ B B B&*".. -AAD-t/JLLr cd|_|jd|_n|xjdz c_|jt|jkr2td|ddS|j|j}td|j|j|jJ|j |j|j|j |j |j |_dS) zwAttempt to connect using the next Parameters config. If there are no more configs, start a new cycle. Nrr6z-_try_next_config_async: starting a new cycle.Frz_try_next_config_async: %r:%s)rpportsocktypeprotorP)rGrr7rr\rqrrprrA getaddrinfo _SOCK_TYPE_IPPROTO_on_getaddrinfo_async_done)rparamss r rz-AMQPConnectionWorkflow._try_next_config_asyncs   % -)*D & &  & &! + & &  %T-E)F)F F F JJF G G G  ' 'e ' 4 4 4 F)$*DE 2FKMMM~%%%//_-3 055r cxd|_t|trMtd||j||ddStdt|t||_ | dS)zHandles completion callback from asynchronous `getaddrinfo()`. :param list | BaseException addrinfos_or_exc: resolved address records returned by `getaddrinfo()` or an exception object from failure. Nzgetaddrinfo failed: %r.Frzgetaddrinfo returned %s records) rGrrr\rrappendrrqr7rr_try_next_resolved_address)raddrinfos_or_excs r rz1AMQPConnectionWorkflow._on_getaddrinfo_async_dones  & 6 6  JJ02B C C C  # * *+; < < <  ' 'e ' 4 4 4 F 4c:J6K6KLLL"#344 '')))))r c| t|j}n?#t$r2td|YdSwxYwtd|||_|j||j |j |j dS)z}Try connecting using next resolved address. If there aren't any left, continue with next Parameters config. z8_try_next_resolved_address: continuing with next config.Nz-Attempting to connect using address record %r)rerfrP) nextr StopIterationr\rqrrrrhrr_on_connector_done)rres r rz1AMQPConnectionWorkflow._try_next_resolved_address)s  t233KK    JJJ L L L  ' ' ) ) ) FF   BKPPP1133 #01KL+  - - - - -s8AAcd|_td|t|tr'|j|t|trU|j|j ks"Jd |j| tdS|j rt|trqtdt|jt jjrt}nt'|j}| |dS|dS| |dS)zHandle completion of connection attempt by `AMQPConnector`. :param pika.connection.Connection | BaseException conn_or_exc: See `AMQPConnector.start()` for exception details. Nz$Connection attempt completed with %rz&Expected _STATE_ABORTING, but got {!r}zcEnding AMQP connection workflow after first failed AMQP handshake due to _until_first_amqp_attempt.)rr\rqrrrrrrKrnrrur,rr*rrTr3ConnectionOpenAbortedr0r)r conn_or_excrs r rz)AMQPConnectionWorkflow._on_connector_done?s 9;GGG k= 1 1 =  # * *; 7 7 7+';<< 2{d&::::<CCDKPP;::33133555550 2[*IJJ 2 NOOOk3"oCEE19EE8/11E33E::::://11111  / / < < < < rs  w"")))))Y)))EEEEE 6EEE+++++1+++4 JJJJJ"8JJJ&55555&A555'B &&&&&&A&&&00000$:000'= NNNNN#9NNN.44444F444D *"*"*"*"*"T[%=*"*"*"ZI=I=I=I=I=;I=I=I=I=I=r