>i5dZddlZddlZddlZddlZddlZddlZddlmZeje Z Gdde Z Gdde ZGdd e ZGd d e ZGd d e ZGddeZGddeZGddeZGdde ZGdde ZGdde ZGddeZGddej4j6ZGddeZy) 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__ceZdZdZy)AMQPConnectorExceptionzBase exception for this moduleN__name__ __module__ __qualname____doc__a/var/www/html/aiagents/fe/lib/python3.12/site-packages/pika/adapters/utils/connection_workflow.pyrrs(r rceZdZdZy)AMQPConnectorStackTimeoutz:Overall TCP/[SSL]/AMQP stack connection attempt timed out.Nrr r r rrsDr rceZdZdZy)AMQPConnectorAbortedz Asynchronous request was abortedNrr r r rrs*r rceZdZdZy)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. c4tt| |||_y)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 )494@"r cbdj|jj|jS)Nz{}: {!r})formatrrrrs r __repr__z$AMQPConnectorPhaseErrorBase.__repr__7s#  !8!8$..IIr rrr r rr! __classcell__rs@r rr(s#Jr rceZdZdZy)AMQPConnectorSocketConnectErrorz*Error connecting TCP socket to remote peerNrr r r r&r&;s4r r&ceZdZdZy) AMQPConnectorTransportSetupErrorzOError setting up transport after TCP connected but before AMQP handshake. Nrr r r r(r(?rr r(ceZdZdZy)AMQPConnectorAMQPHandshakeErrorzError during AMQP handshakeNrr r r r*r*Es%r r*ceZdZdZy)AMQPConnectionWorkflowAbortedz%AMQP Connection workflow was aborted.Nrr r r r,r,Is/r r,ceZdZdZy) 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. cFtt| |t||_y)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__Xs! *D:DA +r cdj|jjt|j|jdt|jdkDr|jdSdS)NzG{}: {} exceptions in all; last exception - {!r}; first exception - {!r}r)rrrlenr3r s r r!z%AMQPConnectionWorkflowFailed.__repr__bsk#$*FNN++S-AOOB'*-doo*>*BDOOA&%N NIM%N Nr r"r$s@r r0r0Ss,Nr r0cpeZdZdZdZdZdZdZdZdZ dZ d Z d Z d Z d Zd ZdZdZdZdZdZddZy) AMQPConnectorz;Performs a single TCP/[SSL]/AMQP connection workflow. rr6c||_||_d|_d|_d|_d|_d|_d|_d|_d|_ |j|_ y)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|jk7r$tdj|j||_||_||_|j|_tj|jdd|_ |jjtjjtjdtjj!|j j"|j|jj%d|jd}t&j)dt*||j,j/|j||j0|_d|_|j j6?|j,j9|j j6|j:|_d|_|j j>@|j,j9|j j>|j@|_yy) 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;;OQ Q(' oo ]]D$5$5bq$9:  dkk1163E3EqI **4+<+<+H+H+/:: 7 u%  # 4k4H22 JJd&B&B3D!%    + + 7$(JJ$9$9!!00//%1D ! #'    * * 6&*jj&;&;!!//1I1I'KD # 7r c2|j|jk(r td|j|jk(r td|j|_|j t jd|jj|j|jWt jd|jjtj |j"t%y|jj&s2t jd|jj)ddyt jd |j|j*k(s Jd j-|jy) 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.abortsI ;;$** *)*IJ J ;;$** *)*VW W**     --22D4E4E G ?? " JJD E JJ . .!!$"E"E"6"8: ;??-- GH%%NP  01{{d&9&99+PfT[[)+9r c|j|j!|jjd|_d|_d|_d|_d|_|j|_y)zqCancel asynchronous tasks and clean up to assist garbage collection. Transition to STATE_DONE. N) rorHrwr@rArBrDrmrKr s r _closezAMQPConnector._closes\  :: ! JJ   DJ!   && r cl|j Jdj|j|j!|jj d|_|j !|j j d|_|j "|j j d|_yy)$Cancel asynchronous tasks. Nz:_deactivate called with self._amqp_conn not None; state={})rIrrKrEcancelrFrGr s r rozAMQPConnector._deactivates&  H O O   &  ,  ! ! ( ( *$(D !  " " .  # # * * ,&*D # >> % NN ! ! #!DN &r ct|trtjd|ntj d||j }|j ||y)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_cleanupsE fm , JJ> G II=v F-- r cd|_ttjdj |j j |j}|j|y)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,sW !%/ NNFMM!!&&(9(9; <= ++E2r c d|_|j}|j|_||jk(rdj |j j |jt|j j}tj||jjr Jdj |j|jjs|jjd|y||j k(rCt#t%dj |j j |j}nq||j&k(sJt)t%dj |j j |jt|j j}|j+|y)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!!&&(9(9T&&2235C JJsO.. H/66tG H.??--%%c3/   (3)@GG))..0A0ACDEE !6!66 664)KF4,,1143D3D 1 1 = =>@ABE ++E2r cd|_|j!|jjd|_| K++ (,, o    ( ( 4++77??K"//;;KKO&"&"3"3"8"8??&..t/A/A/3/@/@B#+99 @; r c  d|_t|trotj d||j j |jt|j j|jt|ytjd||\}|_ |j|_|jj!|j"d|jj%|j"y)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(9(E(E#F H  / /08 :  6?&,# DO&&  22  ( ( 3 ? ,,T-I-IJr Nc ftjd|j||jj|j d|_|j|jk(r t}n3|j|jk(rattdj|jj|j t|jj}n|j|jk(r}|9tjd|jj|j ||}netjd|jj|j |t|}n#tjd|j||y|j!|y)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_donesg  #{{E43D3D3I3I      ;;$.. .)+F [[D// /4)BII))..0A0AT..::;=>?F [[D,, ,} N%%**D,=,=zK$  !%!2!2!7!79J9J9? JJH UJ 0  ++F3r )N)rrr r rJrQrrrxrnrmrrhryr{rorurbrdr_rrr r r r9r9jskKJKNOK'8/Kb,+\'&". 3,3\+Z KD34r r9ceZdZdZdZdZy)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 rceZdZdZej ZejZdZ dZ dZ dZ ddZ dZdZd Zd Zd Zd Zd ZdZdZdZdZy)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;cd|_d|_||_d|_d|_d|_d|_d|_d|_d|_ d|_ g|_ |j|_ y)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__3ss $(  )B& &*"#' "& "#%&& r c||_y)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 cj|j|jk7r$tdj|j t ||stdj|||_||_ ||_ |dj|_ |dj|_|j|_t j#d|j$j'dt)j*|j,d |_y#t $r}t dj|d}~wwxYw) 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;;OQ Q  # $ "4;;>((  FG.. y  ! >**   ( ) ?? " JJD E JJ . .!!$"E"E"?"AC D JJ, - OO ! ! #r c|jd|_d|_d|_d|_d|_d|_d|_|j|_ y)zrCancel asynchronous tasks and clean up to assist garbage collection. Transition to _STATE_DONE. N) rorrArrDrrrrmrKr s r r{zAMQPConnectionWorkflow._closesR #'  "& ""&&& r c`|j"|jjd|_yy)r}N)rGr~r s r roz"AMQPConnectionWorkflow._deactivates* >> % NN ! ! #!DN &r ct|trtjd|ntj d||j }|j ||y)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_cleanupsE fm , JJG P IIF O-- r cd|_|jdk\sJ|j|jdkr=t|j}tj d||j |y|xjdzc_tjd|jd|_|jj|rdn |j|j|_y)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,Fd.F.FF,  # #q (01H1HIE JJ=u E  / / 6    A%   '(,(@(@ B&*"..AD--t/J/JLr cJd|_|jd|_n|xjdz c_|jt|jk\r(tj d|j dy|j|j}tj d|j|j|jJ|jj|j|j|j|j|j|_y) 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-E)F F JJF G  ' 'e ' 4 ))$*D*DE 2FKKM~~%%%//__--33 05r c8d|_t|trDtj d||j j ||jdytjdt|t||_ |jy)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 JJ02B C  # # * *+; <  ' 'e ' 4  4c:J6KL"#34 '')r cl t|j}tj d||j |_|jj||j|j|jy#t$r(tj d|j YywxYw)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  t223K BKP113 #001K1KL++  -  JJJ L  ' ' )   sB.B32B3cd|_tjd|t|tr|j j |t|trS|j|jk(s Jdj|j|jty|jrt|trqtjdt|jt j"j$rt}nt'|j }|j|y|j)y|j|y)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;G k= 1  # # * *; 7+';<{{d&:&::Q<CCDKKPQ:3313500[*IJ NOk33"ooCCE9E8//1E33E://1  / / rs w")Y)E 6E+1+4 J"8J&5&A5'B &&A&0$:0'= N#9N.4F4D *"T[[%=%=*"ZI=;I=r