ó Ì g]c@sÀdZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl m Z m Z ddlZddlZejejfZejejfZejeƒZejjeƒZd„Zd„Zd„Zdefd„ƒYZ d efd „ƒYZ!d e fd „ƒYZ"d efd„ƒYZ#defd„ƒYZ$de fd„ƒYZ%de%fd„ƒYZ&de%fd„ƒYZ'dS(s^Utilities for implementing `nbio_interface.AbstractIOServices` for pika connection adapters. iÿÿÿÿN(tAbstractIOReferencetAbstractStreamTransportcCs+t|ƒs'tdj||ƒƒ‚ndS(s£Raise TypeError if callback is not callable :param callback: callback to check :param name: Name to include in exception text :raises TypeError: s!{} must be callable, but got {!r}N(tcallablet TypeErrortformat(tcallbacktname((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pytcheck_callback_arg,s  cCs.t|tjƒs*tdj|ƒƒ‚ndS(sqRaise TypeError if file descriptor is not an integer :param fd: file descriptor :raises TypeError: s0Paramter must be a file descriptor, but got {!r}N(t isinstancetnumberstIntegralRR(tfd((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyt check_fd_arg9scs"tjˆƒ‡fd†ƒ}|S(s0Function decorator for retrying on SIGINT. csYxRtrTyˆ||ŽSWqtjjk rP}|jtjkrJqqQ‚qXqWdS(sWrapper for decorated functionN(tTruetpikatcompatt SOCKET_ERRORterrnotEINTR(targstkwargsterror(tfunc(sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pytretry_sigint_wrapJs (t functoolstwraps(RR((RsJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyt_retry_on_sigintEs tSocketConnectionMixincBseZdZd„ZRS(súImplements `pika.adapters.utils.nbio_interface.AbstractIOServices.connect_socket()` on top of `pika.adapters.utils.nbio_interface.AbstractFileDescriptorServices` and basic `pika.adapters.utils.nbio_interface.AbstractIOServices`. c Cs%td|d|d|d|ƒjƒS(s[Implement :py:meth:`.nbio_interface.AbstractIOServices.connect_socket()`. tnbiotsockt resolved_addrton_done(t_AsyncSocketConnectortstart(tselfRRR((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pytconnect_socketbs(t__name__t __module__t__doc__R#(((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyRYstStreamingConnectionMixincBseZdZddd„ZRS(sÒImplements `.nbio_interface.AbstractIOServices.create_streaming_connection()` on top of `.nbio_interface.AbstractFileDescriptorServices` and basic `nbio_interface.AbstractIOServices` services. c Csœy5td|d|d|d|d|d|ƒjƒSWn`tk r—}tjd||ƒy|jƒWn&tk r}tjd||ƒnX‚nXd S( shImplement :py:meth:`.nbio_interface.AbstractIOServices.create_streaming_connection()`. Rtprotocol_factoryRt ssl_contexttserver_hostnameRs*create_streaming_connection(%s) failed: %rs%s.close() failed: %rN(t_AsyncStreamConnectorR!t Exceptiont_LOGGERRtclose(R"R(RRR)R*R((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pytcreate_streaming_connectionts  N(R$R%R&tNoneR/(((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR'lst_AsyncServiceAsyncHandlecBs eZdZd„Zd„ZRS(sGThis module's adaptation of `.nbio_interface.AbstractIOReference` cCs|j|_dS(sZ :param subject: subject of the reference containing a `cancel()` method N(tcancelt_cancel(R"tsubject((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyt__init__šscCs |jƒS(sCancel pending operation :returns: False if was already done or cancelled; True otherwise :rtype: bool (R3(R"((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR2¡s(R$R%R&R5R2(((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR1•s R cBs}eZdZdZdZdZdZd„Zed„ƒZ d„Z d„Z ed „ƒZ ed „ƒZ ed „ƒZRS( súConnects the given non-blocking socket asynchronously using `.nbio_interface.AbstractFileDescriptorServices` and basic `.nbio_interface.AbstractIOServices`. Used for implementing `.nbio_interface.AbstractIOServices.connect_socket()`. iiiicCsÈt|dƒytj|j|dƒWn`tk rŠ}ttdƒsYtjdƒq‹dj|||ƒ}tj |ƒt |ƒ‚nX||_ ||_ ||_ ||_|j|_t|_dS(s :param AbstractIOServices | AbstractFileDescriptorServices nbio: :param socket.socket sock: non-blocking socket that needs to be connected via `socket.socket.connect()` :param tuple resolved_addr: resolved destination address/port two-tuple which is compatible with the given's socket's address family :param callable on_done: user callback that takes None upon successful completion or exception upon error (check for `BaseException`) as its only arg. It will not be called if the operation was cancelled. :raises ValueError: if host portion of `resolved_addr` is not an IP address or is inconsistent with the socket's address family as validated via `socket.inet_pton()` Rit inet_ptons8Unable to check resolved address: no socket.inet_pton().s9Invalid or unresolved IP address {!r} for socket {}: {!r}N(RtsocketR6tfamilyR,thasattrR-tdebugRRt ValueErrort_nbiot_sockt_addrt_on_donet_STATE_NOT_STARTEDt_statetFalset_watching_socket_events(R"RRRRRtmsg((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR5·s"        cCs2|jr.t|_|jj|jjƒƒndS(s'Remove socket watcher, if any N(RCRBR<t remove_writerR=tfileno(R"((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyt_cleanupÛs  cCsP|j|jks'td|jfƒ‚|j|_|jj|jƒt|ƒS(sZStart asynchronous connection establishment. :rtype: AbstractIOReference s:_AsyncSocketConnector.start(): expected _STATE_NOT_STARTED(RAR@tAssertionErrort _STATE_ACTIVER<tadd_callback_threadsafet _start_asyncR1(R"((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR!äs  cCsb|j|jkrE|j|_tjd|j|jƒ|jƒtStjd|j|jƒt S(s¾Cancel pending connection request without calling user's completion callback. :returns: False if was already done or cancelled; True otherwise :rtype: bool s-User canceled connection request for %s to %ssD_AsyncSocketConnector cancel requested when not ACTIVE: state=%s; %s( RARIt_STATE_CANCELEDR-R:R=R>RGR RB(R"((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR2õs   cCs‘tjd||jƒt|ttdƒfƒsCtd|fƒ‚|j|j ksjtd|jfƒ‚|j |_|j ƒ|j |ƒdS(s¹Advance to COMPLETED state, remove socket watcher, and invoke user's completion callback. :param BaseException | None result: value to pass in user's callback s0_AsyncSocketConnector._report_completion(%r); %ssP_AsyncSocketConnector._report_completion() expected exception or None as result.sF_AsyncSocketConnector._report_completion() expected _STATE_NOT_STARTEDN( R-R:R=Rt BaseExceptionttypeR0RHRARIt_STATE_COMPLETEDRGR?(R"tresult((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyt_report_completion s     cCs?|j|jkr5tjd|j|j|jƒdSy|jj|jƒWnsttj j fk rÁ}t |tj j ƒr‘|j t kr‘qÂtjd|j|j|ƒ|j|ƒdSnXy#|jj|jjƒ|jƒWn7tk r}tjd|j|ƒ|j|ƒdSXt|_tjd|jƒdS(s“Called as callback from I/O loop to kick-start the workflow, so it's safe to call user's completion callback from here, if needed sJAbandoning sock=%s connection establishment to %s due to inactive state=%sNs%s.connect(%s) failed: %rsasync.set_writer(%s) failed: %rs/Connection-establishment is in progress for %s.(RARIR-R:R=R>tconnectR,RRRRRt(_CONNECTION_IN_PROGRESS_SOCK_ERROR_CODESRRQR<t set_writerRFt _on_writablet exceptionR RC(R"R((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyRK s2  #   cCsº|j|jkr/tjd|j|jƒdS|jjtjtjƒ}|sltj d|jƒd}n=t j |ƒ}tjd|j||ƒt jj||ƒ}|j|ƒdS(swCalled when socket connects or fails to. Check for predicament and invoke user's completion callback. s_Socket connection-establishment event watcher called in inactive state (ignoring): %s; state=%sNsSocket connected: %ss+Socket failed to connect: %s; error=%s (%s)(RARIR-RR=t getsockoptR7t SOL_SOCKETtSO_ERRORtinfoR0toststrerrorRRRRQ(R"t error_codeRPt error_msg((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyRUGs   (R$R%R&R@RIRLROR5t_log_exceptionsRGR!R2RQRKRU(((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR «s $  'R+cBsŒeZdZdZdZdZdZd„Zed„ƒZ d„Z d„Z ed „ƒZ ed „ƒZ ed „ƒZed „ƒZRS( sùPerforms asynchronous SSL session establishment, if requested, on the already-connected socket and links the streaming transport to protocol. Used for implementing `.nbio_interface.AbstractIOServices.create_streaming_connection()`. iiiicCst|dƒt|dƒt|tdƒtjfƒsPtdj|ƒƒ‚n|dk rw|dkrwtdƒ‚ny|jƒWn+t k r²}tdj||ƒƒ‚nX||_ ||_ ||_ ||_ ||_||_|j|_t|_dS(s NOTE: We take ownership of the given socket upon successful completion of the constructor. See `AbstractIOServices.create_streaming_connection()` for detailed documentation of the corresponding args. :param AbstractIOServices | AbstractFileDescriptorServices nbio: :param callable protocol_factory: :param socket.socket sock: :param ssl.SSLContext | None ssl_context: :param str | None server_hostname: :param callable on_done: R(Rs8Expected ssl_context=None | ssl.SSLContext, but got {!r}s?Non-None server_hostname must not be passed without ssl contextsEExpected connected socket, but getpeername() failed: error={!r}; {}; N(RRRNR0tsslt SSLContextR;Rt getpeernameR,R<t_protocol_factoryR=t _ssl_contextt_server_hostnameR?R@RARBt_watching_socket(R"RR(RR)R*RR((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR5ps*          cCstjd|ƒ|jrmtjd||jƒt|_|jj|jjƒƒ|jj|jjƒƒnzc|rÏtjd||jƒy|jj ƒWqÏt k rË}tj d||jƒ‚qÏXnWdd|_d|_d|_ d|_d|_d|_XdS(seCancel pending async operations, if any :param bool close: close the socket if true s"_AsyncStreamConnector._cleanup(%r)s5_AsyncStreamConnector._cleanup(%r): removing RdWr; %ss6_AsyncStreamConnector._cleanup(%r): closing socket; %ss"_sock.close() failed: error=%r; %sN(R-R:RfR=RBR<t remove_readerRFRER.R,RVR0RcRdReR?(R"R.R((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyRG s2           cCsctjd|jƒ|j|jks:td|jfƒ‚|j|_|jj|j ƒt |ƒS(sCKick off the workflow :rtype: AbstractIOReference s!_AsyncStreamConnector.start(); %ss9_AsyncStreamConnector.start() expected _STATE_NOT_STARTED( R-R:R=RAR@RHRIR<RJRKR1(R"((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR!Ãs cCsb|j|jkrE|j|_tjd|jƒ|jdtƒtStjd|j|jƒtS(s¾Cancel pending connection request without calling user's completion callback. :returns: False if was already done or cancelled; True otherwise :rtype: bool s%User canceled streaming linkup for %sR.sD_AsyncStreamConnector cancel requested when not ACTIVE: state=%s; %s( RARIRLR-R:R=RGR RB(R"((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR2Ös cCsÙtjd||jƒt|ttfƒsCtd||jfƒ‚|j|jksjtd|jfƒ‚|j |_zBy|j |ƒWn*t k r¶tj d|j |ƒ‚nXWd|jdt|tƒƒXdS(sAdvance to COMPLETED state, cancel async operation(s), and invoke user's completion callback. :param BaseException | tuple result: value to pass in user's callback. `tuple(transport, protocol)` on success, exception on error s0_AsyncStreamConnector._report_completion(%r); %ssQ_AsyncStreamConnector._report_completion() expected exception or tuple as result.sA_AsyncStreamConnector._report_completion() expected _STATE_ACTIVEs%r: _on_done(%r) failed.NR.(R-R:R=RRMttupleRHRARIROR?R,RVRQRG(R"RP((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyRQês"       c Csðtjd|jƒ|j|jkrBtjd|j|jƒdS|jdkr^|jƒnŽtjd|jƒy7|jj|jdt dt dt d|j ƒ|_Wn7t k rá}tj d |j|ƒ|j |ƒdSX|jƒdS( s’Called as callback from I/O loop to kick-start the workflow, so it's safe to call user's completion callback from here if needed s(_AsyncStreamConnector._start_async(); %ssMAbandoning streaming linkup due to inactive state transition; state=%s; %s; .NsStarting SSL handshake on %st server_sidetdo_handshake_on_connecttsuppress_ragged_eofsR*sSSL wrap_socket(%s) failed: %r(R-R:R=RARIRdR0t_linkupt wrap_socketRBReR,RVRQt_do_ssl_handshake(R"R((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyRK s,    cCstjdƒd}yQy|jƒ}Wn,tk rT}tjd||jƒ‚nX|jdkr²yt|j||j ƒ}Wqýtk r®}tjd||jƒ‚qýXnKyt |j||j ƒ}Wn,tk rü}tjd||jƒ‚nXtjd|ƒy|j |ƒWn/tk rO}tjd|||jƒ‚nXtjd||ƒWntk r}|}n X||f}|j |ƒdS( s}Connection is ready: instantiate and link up transport and protocol, and invoke user's completion callback. s_AsyncStreamConnector._linkup()s'protocol_factory() failed: error=%r; %ss%PlainTransport() failed: error=%r; %ss#SSLTransport() failed: error=%r; %ss_linkup(): created transport %rs1protocol.connection_made(%r) failed: error=%r; %ss2_linkup(): introduced transport to protocol %r; %rN( R-R:R0RcR,RVR=Rdt_AsyncPlaintextTransportR<t_AsyncSSLTransporttconnection_madeRQ(R"t transporttprotocolRRP((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyRl0sL            cCstjdƒ|j|jkr<tjd|j|jƒdSt}yy|jjƒWnëtjk rC}|j tj krÔtjd|jƒt |_ |j j|jjƒ|jƒ|j j|jjƒƒq]|j tjkr=tjd|jƒt |_ |j j|jjƒ|jƒ|j j|jjƒƒq]‚nXt }tjd|jƒWn7tk r—}tjd||jƒ|j|ƒdSX|r tjd|jƒ|j j|jjƒƒ|j j|jjƒƒt|_ tjd |jƒ|jƒndS( sJPerform asynchronous SSL handshake on the already wrapped socket s)_AsyncStreamConnector._do_ssl_handshake()s`_do_ssl_handshake: Abandoning streaming linkup due to inactive state transition; state=%s; %s; .NsSSL handshake wants read; %s.sSSL handshake wants write. %ss(SSL handshake completed successfully: %ss%SSL do_handshake failed: error=%r; %ss8_do_ssl_handshake: removing watchers ahead of linkup: %ss=_do_ssl_handshake: pre-linkup removal of watchers is done; %s(R-R:RARIR=RBt do_handshakeR`tSSLErrorRtSSL_ERROR_WANT_READR RfR<t set_readerRFRnREtSSL_ERROR_WANT_WRITERTRgRZR,RVRQRl(R"tdoneR((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyRnjsX             (R$R%R&R@RIRLROR5R_RGR!R2RQRKRlRn(((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR+ds 0#   &:t_AsyncTransportBasecBsíeZdZdZdZdZdZdZdZde fd „ƒYZ d „Z d „Z d „Z d „Zd„Zd„Zd„Zeed„ƒƒZeed„ƒƒZed„ƒZed„ƒZed„ƒZed„ƒZRS(sIBase class for `_AsyncPlaintextTransport` and `_AsyncSSLTransport`. iiiiiiidt RxEndOfFilecBseZdZd„ZRS(sNWe raise this internally when EOF (empty read) is detected on input. cCs ttj|ƒjddƒdS(NiÿÿÿÿsEnd of input stream (EOF)(tsuperRzR{R5(R"((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR5½s(R$R%R&R5(((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR{¸scCsStjd|ƒ||_||_||_|j|_tjƒ|_ d|_ dS(s~ :param socket.socket | ssl.SSLSocket sock: connected socket :param pika.adapters.utils.nbio_interface.AbstractStreamProtocol protocol: corresponding protocol in this transport/protocol pairing; the protocol already had its `connection_made()` method called. :param AbstractIOServices | AbstractFileDescriptorServices nbio: s _AsyncTransportBase.__init__: %siN( R-R:R=t _protocolR<RIRAt collectionstdequet _tx_bufferst_tx_buffered_byte_count(R"RRsR((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR5Ás     cCs*tjd|j|jƒ|jdƒdS(sClose connection abruptly without waiting for pending I/O to complete. Will invoke the corresponding protocol's `connection_lost()` method asynchronously (not in context of the abort() call). :raises Exception: Exception-based exception on error s+Aborting transport connection: state=%s; %sN(R-RZRAR=t_initiate_abortR0(R"((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pytabortÔs cCs|jS(sReturn the protocol linked to this transport. :rtype: pika.adapters.utils.nbio_interface.AbstractStreamProtocol (R}(R"((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyt get_protocolàscCs|jS(se :returns: Current size of output data buffered by the transport :rtype: int (R(R"((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pytget_write_buffer_sizeçscCs|s7tjd|j|jƒtdj|ƒƒ‚n|j|jkrftjd|j|jƒdS|jj |ƒ|j t |ƒ7_ dS(s–Buffer the given data until it can be sent asynchronously. :param bytes data: :raises ValueError: if called with empty data s,write() called with empty data: state=%s; %ss#write() called with empty data {!r}s;Ignoring write() called during inactive state: state=%s; %sN( R-RRAR=R;RRIR:R€tappendRtlen(R"tdata((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyt_buffer_tx_dataîs cCsÅd}x¸|j|jkrÀ||jkrÀ|j|j|jƒ}|t|ƒ7}|sztjd|jƒ|j ƒ‚ny|j j |ƒWq t k r¼}tj d||jƒ‚q Xq WdS(sïUtility method for use by subclasses to ingest data from socket and dispatch it to protocol's `data_received()` method socket-specific "try again" exception, per-event data consumption limit is reached, transport becomes inactive, or a fatal failure. Consumes up to `self._MAX_CONSUME_BYTES` to prevent event starvation or until state becomes inactive (e.g., `protocol.data_received()` callback aborts the transport) :raises: Whatever the corresponding `sock.recv()` raises except the socket error with errno.EINTR :raises: Whatever the `protocol.data_received()` callback raises :raises _AsyncTransportBase.RxEndOfFile: upon shutdown of input stream isSocket EOF; %ss-protocol.data_received() failed: error=%r; %sN(RARIt_MAX_CONSUME_BYTESt_sigint_safe_recvR=t_MAX_RECV_BYTESR‡R-RR{R}t data_receivedR,RV(R"tbytes_consumedRˆR((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyt_consumes cCsºx³|jrµ|j|j|jdƒ}|jjƒ}|t|ƒkrytjd|t|ƒƒ|jj||ƒn|j|8_|jdkst d|j|j fƒ‚qWdS(sUtility method for use by subclasses to emit data from tx_buffers. This method sends chunks from `tx_buffers` until all chunks are exhausted or sending is interrupted by an exception. Maintains integrity of `self.tx_buffers`. :raises: whatever the corresponding `sock.send()` raises except the socket error with errno.EINTR is/Partial send, requeing remaining data; %s of %ss7_AsyncTransportBase._produce() tx buffer size underflowN( R€t_sigint_safe_sendR=tpopleftR‡R-R:t appendleftRRHRA(R"tnum_bytes_senttchunk((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyt_produce(s   cCs |j|ƒS(smReceive data from socket, retrying on SIGINT. :param sock: stream or SSL socket :param max_bytes: maximum number of bytes to receive :returns: received data or empty bytes uppon end of file :rtype: bytes :raises: whatever the corresponding `sock.recv()` raises except socket error with errno.EINTR (trecv(Rt max_bytes((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR‹As cCs |j|ƒS(s@Send data to socket, retrying on SIGINT. :param sock: stream or SSL socket :param data: data bytes to send :returns: number of bytes actually sent :rtype: int :raises: whatever the corresponding `sock.send()` raises except socket error with errno.EINTR (tsend(RRˆ((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyRPs cCsq|j|jkrmtjd|j|jƒ|jj|jjƒƒ|jj|jjƒƒ|j j ƒndS(s2Unregister the transport from I/O events s$Deactivating transport: state=%s; %sN( RARIR-RZR=R<RgRFRER€tclear(R"((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyt _deactivate_s  cCs—|j|jkr“tjd|j|jƒy|jjtjƒWntj j k r[nX|jj ƒd|_d|_ d|_|j|_ndS(s{Close the transport's socket and unlink the transport it from references to other assets (protocol, etc.) s4Closing transport socket and unlinking: state=%s; %sN(RAROR-RZR=tshutdownR7t SHUT_RDWRRRRR.R0R}R<(R"((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyt_close_and_finalizeks     cCs tjd|j||jƒ|j|jksCtd|jfƒ‚|j|jkrYdS|jƒ|dkr¡|j|jkr’tj dƒdS|j|_nI|j|j krÞ|j|jksÚtd|jfƒ‚dS|j |_|j j tj|j|ƒƒdS(s£Initiate asynchronous abort of the transport that concludes with a call to the protocol's `connection_lost()` method. No flushing of output buffers will take place. :param BaseException | None error: None if being canceled by user, including via falsie return value from protocol.eof_received; otherwise the exception corresponding to the the failed connection. so_AsyncTransportBase._initate_abort(): Initiating abrupt asynchronous transport shutdown: state=%s; error=%r; %ssB_AsyncTransportBase._initate_abort() expected non-_STATE_COMPLETEDNsM_AsyncTransportBase._initiate_abort(): ignoring - user-abort already pending.sD_AsyncTransportBase._initate_abort() expected _STATE_ABORTED_BY_USER(R-RZRAR=RORHRšR0t_STATE_ABORTED_BY_USERR:RIt _STATE_FAILEDR<RJRtpartialt_connection_lost_notify_async(R"R((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR‚~s,      cCsÑtjd|j|ƒ|j|jkr,dS|dk ru|j|jkru|j|jksqtd|jfƒ‚dSzJy|jj |ƒWn/t k r½}tj d|||j ƒ‚nXWd|j ƒXdS(sžHandle aborting of transport either due to socket error or user- initiated `abort()` call. Must be called from an I/O loop callback owned by us in order to avoid reentry into user code from user's API call into the transport. :param BaseException | None error: None if being canceled by user; otherwise the exception corresponding to the the failed connection. s1Concluding transport shutdown: state=%s; error=%rNsS_AsyncTransportBase._connection_lost_notify_async() expected _STATE_ABORTED_BY_USERs/protocol.connection_lost(%r) failed: exc=%r; %s(R-R:RAROR0RŸRžRHR}tconnection_lostR,RVR=R(R"Rtexc((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR¡³s"    i(R$R%R&RIRŸRžRORŒRŠtOSErrorR{R5RƒR„R…R‰RR•t staticmethodRR‹RR_RšRR‚R¡(((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyRz¨s.     %  5RocBs>eZdZd„Zd„Zed„ƒZed„ƒZRS(s`Implementation of `nbio_interface.AbstractStreamTransport` for a plaintext connection. cCs?tt|ƒj|||ƒ|jj|jjƒ|jƒdS(s{ :param socket.socket sock: non-blocking connected socket :param pika.adapters.utils.nbio_interface.AbstractStreamProtocol protocol: corresponding protocol in this transport/protocol pairing; the protocol already had its `connection_made()` method called. :param AbstractIOServices | AbstractFileDescriptorServices nbio: N(R|RoR5R<RwR=RFt_on_socket_readable(R"RRsR((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR5Ýs cCsŸ|j|jkr/tjd|j|jƒdS|sMtd||jfƒ‚|jƒsŽ|jj|jj ƒ|j ƒtjd|jƒn|j |ƒdS(s–Buffer the given data until it can be sent asynchronously. :param bytes data: :raises ValueError: if called with empty data s;Ignoring write() called during inactive state: state=%s; %sNs7_AsyncPlaintextTransport.write(): empty data from user.s!Turned on writability watcher: %s( RARIR-R:R=RHR…R<RTRFt_on_socket_writableR‰(R"Rˆ((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pytwriteís  cCs½|j|jkr/tjd|j|jƒdSy|jƒWnK|jk rôy|jjƒ}Wn6t k r›}tj d||jƒ|j |ƒq¹X|rÑtj d|jƒ|j j|jjƒƒq¹tj d|jƒ|j dƒnÅt tjjfk rŠ}t|tjjƒrI|jtkrItjd|jƒq¹tj d||jdjtjtjƒŒƒƒ|j |ƒn/X|j|jkr¹tjd |j|jƒndS( sÜIngest data from socket and dispatch it to protocol until exception occurs (typically EAGAIN or EWOULDBLOCK), per-event data consumption limit is reached, transport becomes inactive, or failure. sEIgnoring readability notification due to inactive state: state=%s; %sNs,protocol.eof_received() failed: error=%r; %ss0protocol.eof_received() elected to keep open: %ss,protocol.eof_received() elected to close: %ssRecv would block on %ssa_AsyncBaseTransport._consume() failed, aborting connection: error=%r; sock=%s; Caller's stack: %sts>Leaving Plaintext consumer due to inactive state: state=%s; %s(RARIR-R:R=RR{R}t eof_receivedR,RVR‚RZR<RgRFR0RRRRRt_TRY_IO_AGAIN_SOCK_ERROR_CODEStjoint tracebacktformat_exceptiontsystexc_info(R"t keep_openR((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR¦sH    cCs1|j|jkr/tjd|j|jƒdS|jsMtd|jfƒ‚y|jƒWn—tt j j fk rô}t |t j j ƒr³|j tkr³tjd|jƒq-tjd||jdjtjtjƒŒƒƒ|j|ƒn9X|js-|jj|jjƒƒtjd|jƒndS(s-Handle writable socket notification sEIgnoring writability notification due to inactive state: state=%s; %sNsP_AsyncPlaintextTransport._on_socket_writable() called, but _tx_buffers is empty.sSend would block on %ssa_AsyncBaseTransport._produce() failed, aborting connection: error=%r; sock=%s; Caller's stack: %sR©s"Turned off writability watcher: %s(RARIR-R:R=R€RHR•R,RRRRRR«RVR¬R­R®R¯R°R‚R<RERF(R"R((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR§7s,  (R$R%R&R5R¨R_R¦R§(((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyRo×s   4RpcBs\eZdZd„Zd„Zed„ƒZed„ƒZed„ƒZed„ƒZ RS(s\Implementation of `.nbio_interface.AbstractStreamTransport` for an SSL connection. cCsgtt|ƒj|||ƒ|j|_d|_|jj|j j ƒ|j ƒ|jj |j ƒdS(s{ :param ssl.SSLSocket sock: non-blocking connected socket :param pika.adapters.utils.nbio_interface.AbstractStreamProtocol protocol: corresponding protocol in this transport/protocol pairing; the protocol already had its `connection_made()` method called. :param AbstractIOServices | AbstractFileDescriptorServices nbio: N( R|RpR5Rt_ssl_readable_actionR0t_ssl_writable_actionR<RwR=RFR¦RJ(R"RRsR((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR5as   cCs¨|j|jkr/tjd|j|jƒdS|jƒdk}|j|ƒ|r¤|jdkr¤|j |_|j j |jj ƒ|j ƒtjd|jƒndS(s–Buffer the given data until it can be sent asynchronously. :param bytes data: :raises ValueError: if called with empty data s;Ignoring write() called during inactive state: state=%s; %sNis!Turned on writability watcher: %s(RARIR-R:R=R…R‰R³R0R•R<RTRFR§(R"Rˆttx_buffer_was_empty((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR¨us  cCs‰|j|jkr/tjd|j|jƒdS|jrly|jƒWq…tk rh}|j|ƒq…Xntjd|j|jƒdS(s+Handle readable socket indication sEIgnoring readability notification due to inactive state: state=%s; %sNs>SSL readable action was suppressed: ssl_writable_action=%r; %s( RARIR-R:R=R²R,R‚R³(R"R((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR¦‹s cCs‰|j|jkr/tjd|j|jƒdS|jrly|jƒWq…tk rh}|j|ƒq…Xntjd|j|jƒdS(s-Handle writable socket notification sEIgnoring writability notification due to inactive state: state=%s; %sNs>SSL writable action was suppressed: ssl_readable_action=%r; %s( RARIR-R:R=R³R,R‚R²(R"R((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR§¡s cCs#t}ytt|ƒjƒWn tjk r¿}|jtjkrZtj d|j ƒq|jtj krˆtj d|j ƒt }qtj d||j djtjtjƒŒƒƒ‚nCX|j|jkrïtj d|j|j ƒdS|jj|jƒ|ry|js3|jj|j jƒ|jƒn|j|_|j|jkrÞ|jj|j jƒƒd|_qÞne|js¤|jj|j jƒ|jƒn|j|_|jrÞ|jj |j jƒƒd|_n|j!r|j r|j"|_|jj|j jƒ|jƒndS(s“[override] Ingest data from socket and dispatch it to protocol until exception occurs (typically ssl.SSLError with SSL_ERROR_WANT_READ/WRITE), per-event data consumption limit is reached, transport becomes inactive, or failure. Update consumer/producer registration. :raises Exception: error that signals that connection needs to be aborted sSSL ingester wants read: %ssSSL ingester wants write: %ssa_AsyncBaseTransport._consume() failed, aborting connection: error=%r; sock=%s; Caller's stack: %sR©s8Leaving SSL consumer due to inactive state: state=%s; %sN(#R R|RpRR`RuRRvR-R:R=RxRBRVR¬R­R®R¯R°RARIR<RJR¦R²RwRFR³RER0RTR§RgR€R•(R"tnext_consume_on_readableR((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR·sL           cCs#d }ytt|ƒjƒWn¦tjk rÅ}|jtjkr`tj d|j ƒt }që|jtj krŽtj d|j ƒt }qëtjd||j djtjtjƒŒƒƒ‚n&X|j sëtdt|jƒfƒ‚|jrô|d k std|jfƒ‚|rŒ|jsF|jj|j jƒ|jƒn|j|_|j|jkrñ|jj|j jƒƒd |_qñq°|js·|jj|j jƒ|j ƒn|j|_|jr°|jj!|j jƒƒd |_q°n¼|j|jkrR|jj|j jƒƒd |_|j|jks°td|jfƒ‚n^|j|jksŽtdd |jd |jd |jfƒ‚d |_|jj!|j jƒƒ|jsú|j"|_|jj|j jƒ|j ƒ|jj#|j ƒn%|j j$ƒr|jj#|j ƒnd S( sI[override] Emit data from tx_buffers all chunks are exhausted or sending is interrupted by an exception (typically ssl.SSLError with SSL_ERROR_WANT_READ/WRITE). Update consumer/producer registration. :raises Exception: error that signals that connection needs to be aborted sSSL emitter wants read: %ssSSL emitter wants write: %ssa_AsyncBaseTransport._produce() failed, aborting connection: error=%r; sock=%s; Caller's stack: %sR©s__AsyncSSLTransport._produce(): no exception from parent class, but data remains in _tx_buffers.sE_AsyncSSLTransport._produce(): next_produce_on_writable is still Nonesr_AsyncSSLTransport._produce(): with empty tx_buffers, writable_action cannot be _produce when readable is _produces_AsyncSSLTransport._produce(): with empty tx_buffers, expected writable_action as _produce when readable_action is not _produceswritable_action:sreadable_action:sstate:N(%R0R|RpR•R`RuRRvR-R:R=RBRxR RVR¬R­R®R¯R°R€RHR‡RAR³R<RTRFR§R²RgRwR¦RERRJtpending(R"tnext_produce_on_writableR((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyR•sr                 ( R$R%R&R5R¨R_R¦R§RR•(((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyRp[s  I((R&R~RRtloggingR R[R7R`R¯R­t"pika.adapters.utils.nbio_interfaceRRt pika.compatRtpika.diagnostic_utilstEAGAINt EWOULDBLOCKR«t EINPROGRESSRSt getLoggerR$R-tdiagnostic_utilstcreate_log_exception_decoratorR_RR RtobjectRR'R1R R+RzRoRp(((sJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pytsB               )¹ÿEÿ/„