l Ì g]c@sdZddlZddlZddlZddlZddlZddlZddlmZej e ƒZ Gd„de ƒZ Gd„de ƒZGd„d e ƒZGd „d e ƒZGd „d e ƒZGd„deƒZGd„deƒZGd„deƒZGd„de ƒZGd„de ƒZGd„de ƒZGd„deƒZGd„dejjƒZGd„deƒZdS( u,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. iN(u __version__cBs|EeZdZdS(uBase exception for this moduleN(u__name__u __module__u__doc__(u __locals__((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyuAMQPConnectorExceptions uAMQPConnectorExceptioncBs|EeZdZdS(u:Overall TCP/[SSL]/AMQP stack connection attempt timed out.N(u__name__u __module__u__doc__(u __locals__((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyuAMQPConnectorStackTimeouts uAMQPConnectorStackTimeoutcBs|EeZdZdS(u Asynchronous request was abortedN(u__name__u __module__u__doc__(u __locals__((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyuAMQPConnectorAborteds uAMQPConnectorAbortedcBs|EeZdZdS(ujAMQPConnector operation requested in wrong state, such as aborting after completion was reported. N(u__name__u __module__u__doc__(u __locals__((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyuAMQPConnectorWrongState"s uAMQPConnectorWrongStatecs,|EeZdZ‡fd†Zd„Z‡S(uMWrapper for exception that occurred during a particular bring-up phase. cs#tt|ƒj|Œ||_dS(uÈ :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(usuperuAMQPConnectorPhaseErrorBaseu__init__u exception(uselfu exceptionuargs(u __class__(uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu__init__-scCsdj|jj|jƒS(Nu{}: {!r}(uformatu __class__u__name__u exception(uself((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu__repr__7s(u__name__u __module__u__doc__u__init__u__repr__(u __locals__((u __class__uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyuAMQPConnectorPhaseErrorBase(s  uAMQPConnectorPhaseErrorBasecBs|EeZdZdS(u*Error connecting TCP socket to remote peerN(u__name__u __module__u__doc__(u __locals__((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyuAMQPConnectorSocketConnectError;s uAMQPConnectorSocketConnectErrorcBs|EeZdZdS(uOError setting up transport after TCP connected but before AMQP handshake. N(u__name__u __module__u__doc__(u __locals__((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu AMQPConnectorTransportSetupError?s u AMQPConnectorTransportSetupErrorcBs|EeZdZdS(uError during AMQP handshakeN(u__name__u __module__u__doc__(u __locals__((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyuAMQPConnectorAMQPHandshakeErrorEs uAMQPConnectorAMQPHandshakeErrorcBs|EeZdZdS(u%AMQP Connection workflow was aborted.N(u__name__u __module__u__doc__(u __locals__((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyuAMQPConnectionWorkflowAbortedIs uAMQPConnectionWorkflowAbortedcBs|EeZdZdS(uuAMQP Connection Workflow operation requested in wrong state, such as aborting after completion was reported. N(u__name__u __module__u__doc__(u __locals__((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu AMQPConnectionWorkflowWrongStateMs u AMQPConnectionWorkflowWrongStatecs,|EeZdZ‡fd†Zd„Z‡S(u5Indicates that AMQP connection workflow failed. cs)tt|ƒj|Œt|ƒ|_dS(u˜ :param sequence exceptions: Exceptions that occurred during the workflow. :param args: args to pass to base class N(usuperuAMQPConnectionWorkflowFailedu__init__utupleu exceptions(uselfu exceptionsuargs(u __class__(uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu__init__XscCsNdj|jjt|jƒ|jdt|jƒdkrG|jdndƒS(NuG{}: {} exceptions in all; last exception - {!r}; first exception - {!r}iiiÿÿÿÿ(uformatu __class__u__name__ulenu exceptionsuNone(uself((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu__repr__bs (u__name__u __module__u__doc__u__init__u__repr__(u __locals__((u __class__uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyuAMQPConnectionWorkflowFailedSs  uAMQPConnectionWorkflowFailedcBs¤|EeZdZdZdZdZdZdZdZdZ d„Z d „Z d „Z d „Z d „Zd „Zd„Zd„Zd„Zd„Zdd„ZdS(u;Performs a single TCP/[SSL]/AMQP connection workflow. iiiiiiicCsj||_||_d|_d|_d|_d|_d|_d|_d|_ d|_ |j |_ dS(uª :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( u _conn_factoryu_nbiouNoneu _addr_recordu _conn_paramsu_on_doneu_tcp_timeout_refu_stack_timeout_refu _task_refu_socku _amqp_connu _STATE_INITu_state(uselfu conn_factoryunbio((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu__init__ws          cCs„|j|jkr-tdj|jƒƒ‚n||_||_||_|j|_tj|jdd…Œ|_ |j j t j j tjdƒt jj|jj|j ƒ|j jdƒ|jd}tjdt|ƒ|jj|j |d|jƒ|_d|_|jjdk rA|jj|jj|jƒ|_nd|_ |jj!dk r€|jj|jj!|j"ƒ|_ ndS( u—Asynchronously 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` u)Already in progress or finished; state={}Niiiu Pika version %s connecting to %ruon_doneF(#u_stateu _STATE_INITuAMQPConnectorWrongStateuformatu _addr_recordu _conn_paramsu_on_doneu _STATE_TCPusocketu_socku setsockoptupikaucompatuSOL_TCPu TCP_NODELAYutcp_socket_optsu set_sock_optsu tcp_optionsu setblockinguFalseu_LOGuinfou __version__u_nbiouconnect_socketu_on_tcp_connection_doneu _task_refuNoneu_tcp_timeout_refusocket_timeoutu call_lateru_on_tcp_connection_timeoutu_stack_timeout_refu stack_timeoutu_on_overall_timeout(uselfu addr_recordu conn_paramsuon_doneuaddr((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyustart“s2            cCs|j|jkr!tdƒ‚n|j|jkrBtdƒ‚n|j|_|jƒtjd|jj |j ƒ|j d krµtj dƒ|jjtj|jtƒƒƒnf|j jsätj dƒ|j jddƒn7tj dƒ|j|jkstd j|jƒƒ‚d S( ušAbort 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. uCannot abort before starting.u*Cannot abort after completion was reporteduCAMQPConnector: beginning client-initiated asynchronous abort; %r/%suXAMQPConnector.abort(): no connection, so just scheduling completion report via I/O loop.u*AMQPConnector.abort(): closing Connection.i@u3Client-initiated abort of AMQP Connection Workflow.uCAMQPConnector.abort(): closing of Connection was already initiated.u9Connection is closing, but not in TIMEOUT state; state={}N(u_stateu _STATE_INITuAMQPConnectorWrongStateu _STATE_DONEu_STATE_ABORTINGu _deactivateu_LOGuinfou _conn_paramsuhostu _addr_recordu _amqp_connuNoneudebugu_nbiouadd_callback_threadsafeu functoolsupartialu_report_completion_and_cleanupuAMQPConnectorAbortedu is_closingucloseu_STATE_TIMEOUTuAssertionErroruformat(uself((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyuabortÄs,           cCsf|jƒ|jdk r2|jjƒd|_nd|_d|_d|_d|_|j|_ dS(uqCancel asynchronous tasks and clean up to assist garbage collection. Transition to STATE_DONE. N( u _deactivateu_sockuNoneucloseu _conn_factoryu_nbiou _addr_recordu_on_doneu _STATE_DONEu_state(uself((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu_closeòs       cCs£|jdks'tdj|jƒƒ‚|jdk rO|jjƒd|_n|jdk rw|jjƒd|_n|jdk rŸ|jjƒd|_ndS(u$Cancel asynchronous tasks. u:_deactivate called with self._amqp_conn not None; state={}N( u _amqp_connuNoneuAssertionErroruformatu_stateu_tcp_timeout_refucancelu_stack_timeout_refu _task_ref(uself((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu _deactivates     cCsSt|tƒr"tjd|ƒntjd|ƒ|j}|jƒ||ƒdS(u´Clean up and invoke client's `on_done` callback. :param pika.connection.Connection | BaseException result: value to pass to user's `on_done` callback. u%AMQPConnector - reporting failure: %ru%AMQPConnector - reporting success: %rN(u isinstanceu BaseExceptionu_LOGuerroruinfou_on_doneu_close(uselfuresultuon_done((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu_report_completion_and_cleanups   cCsDd|_ttjdj|jj|jƒƒƒ}|j |ƒdS(utHandle TCP connection timeout Reports AMQPConnectorSocketConnectError with socket.timeout inside. u)TCP connection attempt timed out: {!r}/{}N( uNoneu_tcp_timeout_refuAMQPConnectorSocketConnectErrorusocketutimeoutuformatu _conn_paramsuhostu _addr_recordu_report_completion_and_cleanup(uselfuerror((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu_on_tcp_connection_timeout,s   cCsDd|_|j}|j|_||jkr¯dj|jj|jt |jj ƒƒ}t j |ƒ|j j s‰tdj|j ƒƒ‚|j js«|j jd|ƒndS||jkrèttdj|jj|jƒƒƒ}nK||jksýt‚ttdj|jj|jt |jj ƒƒƒƒ}|j|ƒdS(uËHandle 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. u0Timeout while setting up AMQP to {!r}/{}; ssl={}uUnexpected open state of {!r}i@Nu*Timeout while connecting socket to {!r}/{}u5Timeout while setting up transport to {!r}/{}; ssl={}(uNoneu_stack_timeout_refu_stateu_STATE_TIMEOUTu _STATE_AMQPuformatu _conn_paramsuhostu _addr_recorduboolu ssl_optionsu_LOGuerroru _amqp_connuis_openuAssertionErroru is_closingucloseu _STATE_TCPuAMQPConnectorSocketConnectErroruAMQPConnectorStackTimeoutu_STATE_TRANSPORTu AMQPConnectorTransportSetupErroru_report_completion_and_cleanup(uselfu prev_stateumsguerror((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu_on_overall_timeout9s2     c Cs6d|_|jdk r1|jjƒd|_n|dk rjtjd||jƒ|jt|ƒƒdStj d|j ƒ|j |_ d}}|j jdk rá|j jj}|j jj}|dkrá|j j}qán|jjdtj|j|j ƒd|j d|d|d|jƒ|_d|_ dS( uHandle completion of asynchronous socket connection attempt. Reports AMQPConnectorSocketConnectError if TCP socket connection failed. :param None|BaseException exc: None on success; exception object on failure u*TCP Connection attempt failed: %r; dest=%rNu)TCP connection to broker established: %r.uprotocol_factoryusocku ssl_contextuserver_hostnameuon_done(uNoneu _task_refu_tcp_timeout_refucancelu_LOGuerroru _addr_recordu_report_completion_and_cleanupuAMQPConnectorSocketConnectErrorudebugu_socku_STATE_TRANSPORTu_stateu _conn_paramsu ssl_optionsucontextuserver_hostnameuhostu_nbioucreate_streaming_connectionu functoolsupartialu _conn_factoryu _on_transport_establishment_done(uselfuexcu ssl_contextuserver_hostname((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu_on_tcp_connection_donegs4              cCs¸d|_t|tƒr]tjd||jj|jt |jj ƒƒ|j t |ƒƒdStj d|ƒ|\}|_|j|_|jj|jddƒ|jj|jƒdS(uQHandle 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. uCAttempt to create the streaming transport failed: %r; %r/%s; ssl=%sNu"Streaming transport linked up: %r.uremove_defaultT(uNoneu _task_refu isinstanceu BaseExceptionu_LOGuerroru _conn_paramsuhostu _addr_recorduboolu ssl_optionsu_report_completion_and_cleanupu AMQPConnectorTransportSetupErroruinfou _amqp_connu _STATE_AMQPu_stateuadd_on_open_error_callbacku_on_amqp_handshake_doneuTrueuadd_on_open_callback(uselfuresultu _transport((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu _on_transport_establishment_done”s     cCs9tjd|j||jj|jƒd|_|j|jkrLt ƒ}nÜ|j|j kr—t t dj |jj|jt|jjƒƒƒƒ}n‘|j|jkr |dkrÝtjd|jj|j|ƒ|}q(tjd|jj|j|ƒt |ƒ}ntjd|j||ƒdS|j|ƒdS(uµHandle 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 uJAMQPConnector: AMQP handshake attempt completed; state=%s; error=%r; %r/%su,Timeout during AMQP handshake{!r}/{}; ssl={}u8AMQPConnector: AMQP connection established for %r/%s: %ru=AMQPConnector: AMQP connection handshake failed for %r/%s: %rugAMQPConnector: Ignoring AMQP handshake completion notification due to wrong state=%s; error=%r; conn=%rN(u_LOGudebugu_stateu _conn_paramsuhostu _addr_recorduNoneu _amqp_connu_STATE_ABORTINGuAMQPConnectorAbortedu_STATE_TIMEOUTuAMQPConnectorAMQPHandshakeErroruAMQPConnectorStackTimeoutuformatuboolu ssl_optionsu _STATE_AMQPu_report_completion_and_cleanup(uselfu connectionuerroruresult((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu_on_amqp_handshake_done¶s:      N(u__name__u __module__u__doc__u _STATE_INITu _STATE_TCPu_STATE_TRANSPORTu _STATE_AMQPu_STATE_TIMEOUTu_STATE_ABORTINGu _STATE_DONEu__init__ustartuabortu_closeu _deactivateu_report_completion_and_cleanupu_on_tcp_connection_timeoutu_on_overall_timeoutu_on_tcp_connection_doneu _on_transport_establishment_doneuNoneu_on_amqp_handshake_done(u __locals__((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu AMQPConnectorjs&   1 .    . - "u AMQPConnectorcBs&|EeZdZd„Zd„ZdS(uMInterface for implementing a custom TCP/[SSL]/AMQP connection workflow. cCs t‚dS(u¦Asynchronously 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. N(uNotImplementedError(uselfuconnection_configsuconnector_factoryu native_loopuon_done((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyustartñscCs t‚dS(u·Abort 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. N(uNotImplementedError(uself((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyuabort s N(u__name__u __module__u__doc__ustartuabort(u __locals__((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyuAbstractAMQPConnectionWorkflowìs  uAbstractAMQPConnectionWorkflowcBs­|EeZdZejZejZdZdZ dZ dZ dd„Z d„Zd„Zd„Zd „Zd „Zd „Zd „Zd „Zd„Zd„Zd„ZdS(uImplements 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. iiiicCs|d|_d|_||_d|_d|_d|_d|_d|_d|_ d|_ d|_ g|_ |j |_dS(uã :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(uNoneu_attempts_remainingu _retry_pauseu_until_first_amqp_attemptu_nbiou_current_config_indexu_connection_configsu_connector_factoryu_on_doneu _connectoru _task_refu_addrinfo_iteru_connection_errorsu _STATE_INITu_state(uselfu_until_first_amqp_attempt((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu__init__3s            cCs ||_dS(uîCalled 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(u_nbio(uselfunbio((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyuset_io_services]s cCs|j|jkr-tdj|jƒƒ‚nyt|ƒWn:tk rw}ztdj|ƒƒ‚WYdd}~XnX|s–tdj|ƒƒ‚n||_||_ ||_ |d j |_ |d j |_|j|_tjdƒ|jjdtj|jdd ƒƒ|_dS( ufOverride `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. u)Already in progress or finished; state={}u3connection_configs does not support iteration: {!r}Nu"connection_configs is empty: {!r}.iu1Starting AMQP Connection workflow asynchronously.iufirstiÿÿÿÿiÿÿÿÿT(u_stateu _STATE_INITuAMQPConnectorWrongStateuformatuiteru Exceptionu TypeErroru ValueErroru_connection_configsu_connector_factoryu_on_doneuconnection_attemptsu_attempts_remainingu retry_delayu _retry_pauseu _STATE_ACTIVEu_LOGudebugu_nbiou call_lateru functoolsupartialu_start_new_cycle_asyncuTrueu _task_ref(uselfuconnection_configsuconnector_factoryu native_loopuon_doneuerror((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyustartks*      cCsÄ|j|jkr!tdƒ‚n!|j|jkrBtdƒ‚n|j|_|jƒtjdƒ|jdkr¦tj dƒ|j j t j|jtƒƒƒntj dƒ|jjƒdS(u<Override `AbstractAMQPConnectionWorkflow.abort()`. uCannot abort before starting.u*Cannot abort after completion was reporteduFAMQPConnectionWorkflow: beginning client-initiated asynchronous abort.u`AMQPConnectionWorkflow.abort(): no connector, so just scheduling completion report via I/O loop.u=AMQPConnectionWorkflow.abort(): requesting connector.abort().N(u_stateu _STATE_INITuAMQPConnectorWrongStateu _STATE_DONEu_STATE_ABORTINGu _deactivateu_LOGuinfou _connectoruNoneudebugu_nbiouadd_callback_threadsafeu functoolsupartialu_report_completion_and_cleanupuAMQPConnectionWorkflowAborteduabort(uself((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyuabort—s        cCsY|jƒd|_d|_d|_d|_d|_d|_d|_|j |_ dS(urCancel asynchronous tasks and clean up to assist garbage collection. Transition to _STATE_DONE. N( u _deactivateuNoneu_connection_configsu_nbiou_connector_factoryu_on_doneu _connectoru_addrinfo_iteru_connection_errorsu _STATE_DONEu_state(uself((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu_close²s        cCs,|jdk r(|jjƒd|_ndS(u$Cancel asynchronous tasks. N(u _task_refuNoneucancel(uself((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu _deactivateÄs cCsSt|tƒr"tjd|ƒntjd|ƒ|j}|jƒ||ƒdS(uÃClean up and invoke client's `on_done` callback. :param pika.connection.Connection | AMQPConnectionWorkflowFailed result: value to pass to user's `on_done` callback. u.AMQPConnectionWorkflow - reporting failure: %ru.AMQPConnectionWorkflow - reporting success: %rN(u isinstanceu BaseExceptionu_LOGuerroruinfou_on_doneu_close(uselfuresultuon_done((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu_report_completion_and_cleanupÌs   cCs¿d|_|jdks't|jƒ‚|jdkrft|jƒ}tjd|ƒ|j|ƒdS|jd8_tj d|jƒd|_ |j j |r¦dn|j |jƒ|_dS(uQStart 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. iu$AMQP connection workflow failed: %r.NiuQBeginning a new AMQP connection workflow cycle; attempts remaining after this: %s(uNoneu _task_refu_attempts_remaininguAssertionErroruAMQPConnectionWorkflowFailedu_connection_errorsu_LOGuerroru_report_completion_and_cleanupudebugu_current_config_indexu_nbiou call_lateru _retry_pauseu_try_next_config_async(uselfufirstuerror((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu_start_new_cycle_asyncÜs     c Csíd|_|jdkr$d|_n|jd7_|jt|jƒkrltjdƒ|jdd ƒdS|j|j}tjd|j |j ƒ|jdksªt ‚|j j d|j d|j d |jd |jd |jƒ|_dS( uwAttempt to connect using the next Parameters config. If there are no more configs, start a new cycle. iiu-_try_next_config_async: starting a new cycle.ufirstNu_try_next_config_async: %r:%suhostuportusocktypeuprotouon_doneF(uNoneu _task_refu_current_config_indexulenu_connection_configsu_LOGudebugu_start_new_cycle_asyncuFalseuhostuportuAssertionErroru_nbiou getaddrinfou _SOCK_TYPEu_IPPROTOu_on_getaddrinfo_async_done(uselfuparams((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu_try_next_config_asyncøs"        cCsd|_t|tƒrLtjd|ƒ|jj|ƒ|jddƒdStj dt |ƒƒt |ƒ|_ |jƒdS(uäHandles completion callback from asynchronous `getaddrinfo()`. :param list | BaseException addrinfos_or_exc: resolved address records returned by `getaddrinfo()` or an exception object from failure. ugetaddrinfo failed: %r.ufirstNugetaddrinfo returned %s recordsF(uNoneu _task_refu isinstanceu BaseExceptionu_LOGuerroru_connection_errorsuappendu_start_new_cycle_asyncuFalseudebugulenuiteru_addrinfo_iteru_try_next_resolved_address(uselfuaddrinfos_or_exc((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu_on_getaddrinfo_async_dones cCs’yt|jƒ}Wn-tk rBtjdƒ|jƒdSYnXtjd|ƒ|jƒ|_|jjd|d|j |j d|j ƒdS(u}Try connecting using next resolved address. If there aren't any left, continue with next Parameters config. u8_try_next_resolved_address: continuing with next config.Nu-Attempting to connect using address record %ru addr_recordu conn_paramsuon_done( unextu_addrinfo_iteru StopIterationu_LOGudebugu_try_next_config_asyncu_connector_factoryu _connectorustartu_connection_configsu_current_config_indexu_on_connector_done(uselfu addr_record((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu_try_next_resolved_address)s    cCsd|_tjd|ƒt|tƒrö|jj|ƒt|tƒr„|j |j ksqt dj |j ƒƒ‚|j tƒƒq|jrét|tƒrétjdƒt|jtjjƒrÊt}nt|jƒ}|j |ƒq|jƒn |j |ƒdS(uÌHandle completion of connection attempt by `AMQPConnector`. :param pika.connection.Connection | BaseException conn_or_exc: See `AMQPConnector.start()` for exception details. u$Connection attempt completed with %ru&Expected _STATE_ABORTING, but got {!r}ucEnding AMQP connection workflow after first failed AMQP handshake due to _until_first_amqp_attempt.N(uNoneu _connectoru_LOGudebugu isinstanceu BaseExceptionu_connection_errorsuappenduAMQPConnectorAbortedu_stateu_STATE_ABORTINGuAssertionErroruformatu_report_completion_and_cleanupuAMQPConnectionWorkflowAbortedu_until_first_amqp_attemptuAMQPConnectorAMQPHandshakeErroru exceptionupikau exceptionsuConnectionOpenAborteduAMQPConnectionWorkflowFailedu_try_next_resolved_address(uselfu conn_or_excuerror((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyu_on_connector_done?s(        NF(u__name__u __module__u__doc__usocketu SOCK_STREAMu _SOCK_TYPEu IPPROTO_TCPu_IPPROTOu _STATE_INITu _STATE_ACTIVEu_STATE_ABORTINGu _STATE_DONEuFalseu__init__uset_io_servicesustartuabortu_closeu _deactivateu_report_completion_and_cleanupu_start_new_cycle_asyncu_try_next_config_asyncu_on_getaddrinfo_async_doneu_try_next_resolved_addressu_on_connector_done(u __locals__((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyuAMQPConnectionWorkflows&    *  ,        uAMQPConnectionWorkflow(u__doc__u functoolsuloggingusocketu pika.compatupikaupika.exceptionsupika.tcp_socket_optsu __version__u getLoggeru__name__u_LOGu ExceptionuAMQPConnectorExceptionuAMQPConnectorStackTimeoutuAMQPConnectorAborteduAMQPConnectorWrongStateuAMQPConnectorPhaseErrorBaseuAMQPConnectorSocketConnectErroru AMQPConnectorTransportSetupErroruAMQPConnectorAMQPHandshakeErroruAMQPConnectionWorkflowAbortedu AMQPConnectionWorkflowWrongStateuAMQPConnectionWorkflowFaileduobjectu AMQPConnectorucompatu AbstractBaseuAbstractAMQPConnectionWorkflowuAMQPConnectionWorkflow(((uL/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/connection_workflow.pyus.      ÿƒ-