)Jf-PdZddlZddlZddlZddlZddlZddlZddlm Z m Z ddlm Z ej e ZGdde jZGdde jZdS) zBase class extended by connection adapters. This extends the connection.Connection class to encapsulate connection behavior but still isolate socket and low level communication. N)connection_workflownbio_interface) connectionceZdZdZfdZfdZdZeej ddZ edZ e dZ d Zd Zd Zd Zed ZdZdZdZdZdZdZdZdZdZxZS)BaseConnectionzBaseConnection class that should be extended by connection adapters. This class abstracts I/O loop and transport services from pika core. c|r,t|tjstd|||_d|_d|_d|_tt| |||||dS)aCreate a new instance of the Connection object. :param None|pika.connection.Parameters parameters: Connection parameters :param None|method on_open_callback: Method to call on connection open :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 pika.adapters.utils.nbio_interface.AbstractIOServices nbio: asynchronous services :param bool internal_connection_workflow: True for autonomous connection establishment which is default; False for externally-managed connection workflow via the `create_connection()` factory. :raises: RuntimeError :raises: ValueError z%Expected instance of Parameters, not NF)internal_connection_workflow) isinstancer Parameters ValueError_nbio_connection_workflow _transport_got_eofsuperr__init__)self parameterson_open_callbackon_open_error_callbackon_close_callbacknbior __class__s `/home/alex/cs2snipeproduction/venv/lib/python3.11/site-packages/pika/adapters/base_connection.pyrzBaseConnection.__init__s0  KjZ5JKK K*=GZIKK K $(!  nd##,,   " )E - G G G G Gctt|d|_d|_d|_dS)zInitialize or reset all of our internal state variables for a given connection. If we disconnect and reconnect, all of our state needs to be wiped. NF)rr_init_connection_staterrr)rrs rrz%BaseConnection._init_connection_stateEs; nd##::<<<$(! rc hd|jjd|j|jd|jd|jd S)N< z transport=z params=>)r__name__ _STATE_NAMESconnection_staterparamsrs r__repr__zBaseConnection.__repr__QsB0 N # # #T%6t7L%M%M%M OOOT[[[* +rNct)aAsynchronously create a connection to an AMQP broker using the given configurations. Will attempt to connect using each config in the given order, including all compatible resolved IP addresses of the hostname supplied in each config, until one is established or all attempts fail. See also `_start_connection_workflow()`. :param sequence connection_configs: A sequence of one or more `pika.connection.Parameters`-based objects. :param callable on_done: as defined in `connection_workflow.AbstractAMQPConnectionWorkflow.start()`. :param object | None custom_ioloop: Provide a custom I/O loop that is native to the specific adapter implementation; if None, the adapter will use a default loop instance, which is typically a singleton. :param connection_workflow.AbstractAMQPConnectionWorkflow | None workflow: Pass an instance of an implementation of the `connection_workflow.AbstractAMQPConnectionWorkflow` interface; defaults to a `connection_workflow.AMQPConnectionWorkflow` instance with default values for optional args. :returns: Connection workflow instance in use. The user should limit their interaction with this object only to it's `abort()` method. :rtype: connection_workflow.AbstractAMQPConnectionWorkflow )NotImplementedError)clsconnection_configson_done custom_ioloopworkflows rcreate_connectionz BaseConnection.create_connectionls >"!rc Z|.tj}td|t |tjr|fd}|||tj |j ||S)a-Helper function for custom implementations of `create_connection()`. :param sequence connection_configs: A sequence of one or more `pika.connection.Parameters`-based objects. :param callable connection_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: :param connection_workflow.AbstractAMQPConnectionWorkflow | None workflow: Pass an instance of an implementation of the `connection_workflow.AbstractAMQPConnectionWorkflow` interface; defaults to a `connection_workflow.AMQPConnectionWorkflow` instance with default values for optional args. :param callable on_done: as defined in :py:meth:`connection_workflow.AbstractAMQPConnectionWorkflow.start()`. :returns: Connection workflow instance in use. The user should limit their interaction with this object only to it's `abort()` method. :rtype: connection_workflow.AbstractAMQPConnectionWorkflow Nz&Created default connection workflow %rc4tjfdS)z`AMQPConnector` factory.c4t|SN_StreamingProtocolShim)r%connection_factorys rzUBaseConnection._start_connection_workflow..create_connector..s 5&&v.. 0 0r)r AMQPConnector)r6rsrcreate_connectorzCBaseConnection._start_connection_workflow..create_connectors/&40000 r)r+connector_factory native_loopr,) rAMQPConnectionWorkflowLOGGERdebugr set_io_servicesstartget_native_ioloop functoolspartial$_unshim_connection_workflow_callback)r*r+r6rr.r,r9s `` r_start_connection_workflowz)BaseConnection._start_connection_workflows2  *ACCH LLA8 L L L h 3 J K K +  $ $T * * *       1...00%c&N&-//  0 0 0rc4|jS)a3 :returns: the native I/O loop instance underlying async services selected by user or the default selected by the specialized connection adapter (e.g., Twisted reactor, `asyncio.SelectorEventLoop`, `select_connection.IOLoop`, etc.) :rtype: object )r rAr&s rioloopzBaseConnection.ioloopsz++---rc8|j||S)zXImplement :py:meth:`pika.connection.Connection._adapter_call_later()`. )r call_later)rdelaycallbacks r_adapter_call_laterz"BaseConnection._adapter_call_laters z$$UH555rc.|dS)z\Implement :py:meth:`pika.connection.Connection._adapter_remove_timeout()`. N)cancel)r timeout_ids r_adapter_remove_timeoutz&BaseConnection._adapter_remove_timeouts rc|t|std||j|dS)zeImplement :py:meth:`pika.connection.Connection._adapter_add_callback_threadsafe()`. z%callback must be a callable, but got N)callable TypeErrorr add_callback_threadsafe)rrKs r _adapter_add_callback_threadsafez/BaseConnection._adapter_add_callback_threadsafesR !! I)=EXGII I **844444rc 4tjd_jjfd}jjg|jtj j j dS)a8Initiate full-stack connection establishment asynchronously for internally-initiated connection bring-up. Upon failed completion, we will invoke `Connection._on_stream_terminated()`. NOTE: On success, the stack will be up already, so there is no corresponding callback. T)_until_first_amqp_attemptc>tjfdjS)z`AMQPConnector` factoryc"tSr3r4)_paramsrs rr7zRBaseConnection._adapter_connect_stream..create_connector..s 6t < <r)rr8r r&srr9z@BaseConnection._adapter_connect_stream..create_connectors+&4<<<rr ExceptionrrAMQPConnectionWorkflowAbortedinfoerrorAMQPConnectionWorkflowFailed exceptionsAMQPConnectorSocketConnectErrorpikaAMQPConnectionError#_handle_connection_workflow_failureformat)r conn_or_excs rr[z+BaseConnection._on_connection_workflow_done/sR  C  " " "%)! k9 - - '"DO+-KMM % H')))#  H(***{2OQQ%&'2268K;==% #'/"E"E##%#%K  4 4[ A A A A A$&&&FMM+'''&&&&rc|tdntd||js||dStddS)aHandle failure of self-initiated stack bring-up and call `Connection._on_stream_terminated()` if connection is not in closed state yet. Called by adapter layer when the full-stack connection workflow fails. :param Exception | None error: exception instance describing the reason for failure or None if the connection workflow was aborted. Nz&Self-initiated stack bring-up aborted.z(Self-initiated stack bring-up failed: %rzO_handle_connection_workflow_failure(): suppressing - connection already closed.)r=rkrl is_closed_on_stream_terminatedr>rrls rrrz2BaseConnection._handle_connection_workflow_failureYs = KK@ A A A A LLCU K K K~ E  & &u - - - - - LLD E E E E Ercr|js|dS|jdS)zAsynchronously bring down the streaming transport layer and invoke `Connection._on_stream_terminated()` asynchronously when complete. N)rcrfrrer&s r_adapter_disconnect_streamz)BaseConnection._adapter_disconnect_streamosB | $  + + - - - - - O ! ! # # # # #rc:|j|dS)ztTake ownership of data and send it to AMQP server as soon as possible. :param bytes data: N)rwriterdatas r_adapter_emit_dataz!BaseConnection._adapter_emit_data{s  d#####rc<||_|dS)aIntroduces transport to protocol after transport is connected. :py:class:`.utils.nbio_interface.AbstractStreamProtocol` implementation. :param nbio_interface.AbstractStreamTransport transport: :raises Exception: Exception-based exception on error N)r_on_stream_connected)r transports r_proto_connection_madez%BaseConnection._proto_connection_mades%$ !!#####rc\d|_|'|jrtjd}n2tjd|}t | tj n tj d|| |dS)aCalled upon loss or closing of TCP connection. :py:class:`.utils.nbio_interface.AbstractStreamProtocol` implementation. NOTE: `connection_made()` and `connection_lost()` are each called just once and in that order. All other callbacks are called between them. :param BaseException | None error: An exception (check for `BaseException`) indicates connection failure. None indicates that connection was closed on this side, such as when it's aborted or when `AbstractStreamProtocol.eof_received()` returns a falsy result. :raises Exception: Exception-based exception on error NzTransport indicated EOFzStream connection lost: {!r}zconnection_lost: %r) rrrprnStreamLostErrorrsr=logloggingDEBUGERRORrwrxs r_proto_connection_lostz%BaseConnection._proto_connection_losts =} /77-//O33.55e<<>>E  EM7==w}(% 1 1 1 ""5)))))rcHtdd|_dS)a%Called after the remote peer shuts its write end of the connection. :py:class:`.utils.nbio_interface.AbstractStreamProtocol` implementation. :returns: A falsy value (including None) will cause the transport to close itself, resulting in an eventual `connection_lost()` call from the transport. If a truthy value is returned, it will be the protocol's responsibility to close/abort the transport. :rtype: falsy|truthy :raises Exception: Exception-based exception on error zTransport indicated EOF.TF)r=rlrr&s r_proto_eof_receivedz"BaseConnection._proto_eof_receiveds%  /000 urc0||dS)aCalled to deliver incoming data from the server to the protocol. :py:class:`.utils.nbio_interface.AbstractStreamProtocol` implementation. :param data: Non-empty data bytes. :raises Exception: Exception-based exception on error N)_on_data_availabler}s r_proto_data_receivedz#BaseConnection._proto_data_receiveds %%%%%r)NN)r" __module__ __qualname____doc__rrr' classmethodabcabstractmethodr/rEpropertyrGrLrPrUr\ staticmethodrDrfr[rrrzrrrrr __classcell__)rs@rrrs (G(G(G(G(GT     +++6)-#' """[">--[-^..X.666 5 5 5JJJ4  \ $$$$$$L('('('TEEE, $ $ $$$$ $ $ $***>. & & & & & & &rrc4eZdZdZdZdZdZdZdZdZ dZ dS)r5zShim for callbacks from transport so that we BaseConnection can delegate to private methods, thus avoiding contamination of API with methods that look public, but aren't. Nct||_|j|_|j|_|j|_|j|_dS)z- :param BaseConnection conn: N) r^rconnection_maderconnection_lostr eof_receivedr data_received)rr^s rrz_StreamingProtocolShim.__init__s= #:#: 4!6rc,t|j|S)zProxy inexistent attribute requests to our connection instance so that AMQPConnectionWorkflow/AMQPConnector may treat the shim as an actual connection. )getattrr^)rattrs r __getattr__z"_StreamingProtocolShim.__getattr__s ty$'''rcLd|jj|jS)Nz{}: {!r})rsrr"r^r&s rr'z_StreamingProtocolShim.__repr__s  !8$)DDDr) r"rrrrrrrrrr'rrr5r5sfOOLM 7 7 7(((EEEEErr5)rrrBr pika.compatrppika.exceptionspika.tcp_socket_optspika.adapters.utilsrrr getLoggerr"r= ConnectionrAbstractStreamProtocolr5rrrrs  CCCCCCCC  8 $ $}&}&}&}&}&Z*}&}&}&@!E!E!E!E!E^B!E!E!E!E!Er