o ei5@s2dZddlZddlZddlZddlZddlZddlZddlmZe e Z Gddde Z Gddde ZGdd d e ZGd d d e ZGd d d e ZGdddeZGdddeZGdddeZGddde ZGddde ZGddde ZGdddeZGdddejjZGdddeZdS) 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__c@eZdZdZdS)AMQPConnectorExceptionzBase exception for this moduleN__name__ __module__ __qualname____doc__r r f/var/www/html/pca-backend/venv/lib/python3.10/site-packages/pika/adapters/utils/connection_workflow.pyrrc@r)AMQPConnectorStackTimeoutz:Overall TCP/[SSL]/AMQP stack connection attempt timed out.Nrr r r r r r r c@r)AMQPConnectorAbortedz Asynchronous request was abortedNrr r r r rr rc@r)AMQPConnectorWrongStatezjAMQPConnector operation requested in wrong state, such as aborting after completion was reported. Nrr r r r r"r rc(eZdZdZfddZddZZS)AMQPConnectorPhaseErrorBasezMWrapper for exception that occurred during a particular bring-up phase. cstt|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__r r r-s z$AMQPConnectorPhaseErrorBase.__init__cCsd|jj|jS)Nz{}: {!r})formatrrrrr r r __repr__7sz$AMQPConnectorPhaseErrorBase.__repr__rrrr rr __classcell__r r rr r(  rc@r)AMQPConnectorSocketConnectErrorz*Error connecting TCP socket to remote peerNrr r r r r;r rc@r) AMQPConnectorTransportSetupErrorzOError setting up transport after TCP connected but before AMQP handshake. Nrr r r r r ?r r c@r)AMQPConnectorAMQPHandshakeErrorzError during AMQP handshakeNrr r r r r!Er r!c@r)AMQPConnectionWorkflowAbortedz%AMQP Connection workflow was aborted.Nrr r r r r"Ir r"c@r) AMQPConnectionWorkflowWrongStatezuAMQP Connection Workflow operation requested in wrong state, such as aborting after completion was reported. Nrr r r r r#Mr r#cr)AMQPConnectionWorkflowFailedz5Indicates that AMQP connection workflow failed. cstt|j|t||_dS)z :param sequence exceptions: Exceptions that occurred during the workflow. :param args: args to pass to base class N)rr$rtuple exceptions)rr&rrr r rXsz%AMQPConnectionWorkflowFailed.__init__cCs:d|jjt|j|jdt|jdkr|jdSdS)NzG{}: {} exceptions in all; last exception - {!r}; first exception - {!r}r)rrrlenr&rr r r rbsz%AMQPConnectionWorkflowFailed.__repr__rr r rr r$Srr$c@seZdZdZdZdZdZdZdZdZ dZ d d Z d d Z d dZ ddZddZddZddZddZddZddZd ddZdS)! AMQPConnectorz;Performs a single TCP/[SSL]/AMQP connection workflow. rr(cCsH||_||_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_factorynbior r r rws  zAMQPConnector.__init__cCs |j|jkrtd|j||_||_||_|j|_tj|jdd|_ |j t j j tjdt j|jj|j |j d|jd}tdt||jj|j ||jd|_d|_|jjdurm|j|jj|j|_d|_|jjdur|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,r(Fr-z Pika version %s connecting to %r)on_done)!r;r:rrr2r3r4 _STATE_TCPsocketr8 setsockoptpikacompatSOL_TCP TCP_NODELAYtcp_socket_opts set_sock_opts tcp_options setblocking_LOGinforr1connect_socket_on_tcp_connection_doner7r5socket_timeout call_later_on_tcp_connection_timeoutr6 stack_timeout_on_overall_timeout)r addr_record conn_paramsr?addrr r r starts>          zAMQPConnector.startcCs|j|jkr td|j|jkrtd|j|_|td|jj |j |j dur>t d|j t|jtdS|j jsPt d|j dddSt d |j|jkscJd |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={})r;r:r _STATE_DONE_STATE_ABORTING _deactivaterKrLr3hostr2r9debugr1add_callback_threadsafe functoolspartial_report_completion_and_cleanupr is_closingclose_STATE_TIMEOUTrrr r r aborts6       zAMQPConnector.abortcCsF||jdur|jd|_d|_d|_d|_d|_|j|_dS)zqCancel asynchronous tasks and clean up to assist garbage collection. Transition to STATE_DONE. N) r]r8rer0r1r2r4r[r;rr r r _closes   zAMQPConnector._closecCsp|jdus Jd|j|jdur|jd|_|jdur'|jd|_|jdur6|jd|_dSdS)$Cancel asynchronous tasks. Nz:_deactivate called with self._amqp_conn not None; state={})r9rr;r5cancelr6r7rr r r r]s        zAMQPConnector._deactivatecC>t|tr td|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 BaseExceptionrKerrorrLr4rhrresultr?r r r rc   z,AMQPConnector._report_completion_and_cleanupcCs0d|_ttd|jj|j}||dS)ztHandle TCP connection timeout Reports AMQPConnectorSocketConnectError with socket.timeout inside. Nz)TCP connection attempt timed out: {!r}/{}) r5rrAtimeoutrr3r^r2rc)rror r r rQ,s z(AMQPConnector._on_tcp_connection_timeoutcCsd|_|j}|j|_||jkr;d|jj|jt|jj }t ||j j r.Jd|j |j js9|j d|dS||jkrNttd|jj|j}n||jksUJttd|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}rZz*Timeout while connecting socket to {!r}/{}z5Timeout while setting up transport to {!r}/{}; ssl={})r6r;rf _STATE_AMQPrr3r^r2bool ssl_optionsrKror9is_openrdrer@rr _STATE_TRANSPORTr rc)r prev_statemsgror r r rS9sB           z!AMQPConnector._on_overall_timeoutcCsd|_|jdur|jd|_|dur%td||j|t|dStd|j |j |_ d}}|j j durL|j j j}|j j j}|durL|j j}|jjt|j|j |j |||jd|_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_hostnamer?)r7r5rjrKror2rcrr_r8rxr;r3rvcontextr~r^r1create_streaming_connectionrarbr0 _on_transport_establishment_done)rexcr}r~r r r rNgs<      z%AMQPConnector._on_tcp_connection_donecCsd|_t|tr!td||jj|jt|jj | t |dSt d||\}|_ |j|_|j j|jdd|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)r7rmrnrKror3r^r2rurvrcr rLr9rtr;add_on_open_error_callback_on_amqp_handshake_doneadd_on_open_callback)rrq _transportr r r rs$   z.AMQPConnector._on_transport_establishment_doneNcCstd|j||jj|jd|_|j|jkrt}nL|j|j kr3t t d |jj|jt |jj}n3|j|jkr[|durKtd|jj|j||}ntd|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)rKr_r;r3r^r2r9r\rrfr!r rrurvrtrc)r connectionrorqr r r rsH          z%AMQPConnector._on_amqp_handshake_done)N)rrrr r:r@rxrtrfr\r[rrWrgrhr]rcrQrSrNrrr r r r r*js(1. .-"r*c@s eZdZdZddZddZdS)AbstractAMQPConnectionWorkflowzMInterface for implementing a custom TCP/[SSL]/AMQP connection workflow. cCt)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_loopr?r r r rWsz$AbstractAMQPConnectionWorkflow.startcCr)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. rrr r r rg s z$AbstractAMQPConnectionWorkflow.abortN)rrrr rWrgr r r r rs rc@seZdZdZejZejZdZ dZ dZ dZ d ddZ d d Zd d Zd dZddZddZddZddZddZddZddZddZdS)!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. rr(r+r,FcCsTd|_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_attemptr1_current_config_index_connection_configs_connector_factoryr4 _connectorr7_addrinfo_iter_connection_errorsr:r;)rrr r r r3s zAMQPConnectionWorkflow.__init__cCs ||_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)r1)rr=r r r set_io_services]s z&AMQPConnectionWorkflow.set_io_servicesc Cs|j|jkrtd|jzt|Wnty'}ztd|d}~ww|s1td|||_||_ ||_ |dj |_ |dj |_|j|_td|jdtj|jdd |_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. r>z3connection_configs does not support iteration: {!r}Nz"connection_configs is empty: {!r}.r'z1Starting AMQP Connection workflow asynchronously.rTfirst)r;r:rriter Exception TypeError ValueErrorrrr4connection_attemptsr retry_delayr _STATE_ACTIVErKr_r1rPrarb_start_new_cycle_asyncr7)rrrrr?ror r r rWks8       zAMQPConnectionWorkflow.startcCs|j|jkr td|j|jkrtd|j|_|td|jdur9t d|j t |jtdSt d|jdS)z|d|_d|_d|_d|_d|_d|_d|_|j|_ dS)zrCancel asynchronous tasks and clean up to assist garbage collection. Transition to _STATE_DONE. N) r]rr1rr4rrrr[r;rr r r rhs zAMQPConnectionWorkflow._closecCs"|jdur|jd|_dSdS)riN)r7rjrr r r r]s   z"AMQPConnectionWorkflow._deactivatecCrk)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: %rNrlrpr r r rcrrz5AMQPConnectionWorkflow._report_completion_and_cleanupcCsd|_|jdks J|j|jdkr$t|j}td|||dS|jd8_td|jd|_|j |rJ|j j |j|j|j |j |j d|_dS) zwAttempt to connect using the next Parameters config. If there are no more configs, start a new cycle. Nrr(z-_try_next_config_async: starting a new cycle.Frz_try_next_config_async: %r:%s)r^portsocktypeprotor?)r7rr)rrKr_rr^rr1 getaddrinfo _SOCK_TYPE_IPPROTO_on_getaddrinfo_async_done)rparamsr r r rs$     z-AMQPConnectionWorkflow._try_next_config_asynccCs^d|_t|trtd||j||jdddStdt |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) r7rmrnrKrorappendrr_r)rr_try_next_resolved_address)raddrinfos_or_excr r r rs      z1AMQPConnectionWorkflow._on_getaddrinfo_async_donecCslzt|j}Wntytd|YdSwtd|||_|jj||j |j |j ddS)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)rTrUr?) nextr StopIterationrKr_rrrrWrr_on_connector_done)rrTr r r r)s      z1AMQPConnectionWorkflow._try_next_resolved_addresscCsd|_td|t|trY|j|t|tr/|j|j ks'Jd |j| t dS|j rSt|trStdt|jtjjrGt }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.)rrKr_rmrnrrrr;r\rrcr"rr!rrCr&ConnectionOpenAbortedr$r)r conn_or_excror r r r?s2       z)AMQPConnectionWorkflow._on_connector_doneN)F)rrrr rA SOCK_STREAMr IPPROTO_TCPrr:rr\r[rrrWrgrhr]rcrrrrrr r r r rs( *, r)r raloggingrA pika.compatrCpika.exceptionspika.tcp_socket_optsr getLoggerrrKrrr rrrrr r!r"r#r$objectr*rD AbstractBaserrr r r r s4   -