l Ì g]c@s¨dZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl m Z m Z ddlZddlZejejfZejejfZejeƒZejjeƒZd„Zd„Zd„ZGd„deƒZ Gd „d eƒZ!Gd „d e ƒZ"Gd „deƒZ#Gd„deƒZ$Gd„de ƒZ%Gd„de%ƒZ&Gd„de%ƒZ'dS(u^Utilities for implementing `nbio_interface.AbstractIOServices` for pika connection adapters. iN(uAbstractIOReferenceuAbstractStreamTransportcCs+t|ƒs'tdj||ƒƒ‚ndS(u£Raise TypeError if callback is not callable :param callback: callback to check :param name: Name to include in exception text :raises TypeError: u!{} must be callable, but got {!r}N(ucallableu TypeErroruformat(ucallbackuname((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyucheck_callback_arg,s  cCs.t|tjƒs*tdj|ƒƒ‚ndS(uqRaise TypeError if file descriptor is not an integer :param fd: file descriptor :raises TypeError: u0Paramter must be a file descriptor, but got {!r}N(u isinstanceunumbersuIntegralu TypeErroruformat(ufd((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu check_fd_arg9scs"tjˆƒ‡fd†ƒ}|S(u0Function decorator for retrying on SIGINT. csdx]yˆ||ŽSWqtjjk r\}z |jtjkrGwn‚WYdd}~XqXqdS(uWrapper for decorated functionN(upikaucompatu SOCKET_ERRORuerrnouEINTR(uargsukwargsuerror(ufunc(uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyuretry_sigint_wrapJs(u functoolsuwraps(ufuncuretry_sigint_wrap((ufuncuJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_retry_on_sigintEs cBs|EeZdZd„ZdS(uú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(u[Implement :py:meth:`.nbio_interface.AbstractIOServices.connect_socket()`. unbiousocku resolved_addruon_done(u_AsyncSocketConnectorustart(uselfusocku resolved_addruon_done((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyuconnect_socketbsN(u__name__u __module__u__doc__uconnect_socket(u __locals__((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyuSocketConnectionMixinYs uSocketConnectionMixincBs#|EeZdZddd„ZdS(uÒImplements `.nbio_interface.AbstractIOServices.create_streaming_connection()` on top of `.nbio_interface.AbstractFileDescriptorServices` and basic `nbio_interface.AbstractIOServices` services. cCsÀy5td|d|d|d|d|d|ƒjƒSWn„tk r»}zdtjd||ƒy|jƒWn8tk r¥}ztjd||ƒWYd d }~XnX‚WYd d }~XnXd S( uhImplement :py:meth:`.nbio_interface.AbstractIOServices.create_streaming_connection()`. unbiouprotocol_factoryusocku ssl_contextuserver_hostnameuon_doneu*create_streaming_connection(%s) failed: %ru%s.close() failed: %rN(u_AsyncStreamConnectorustartu Exceptionu_LOGGERuerroruclose(uselfuprotocol_factoryusockuon_doneu ssl_contextuserver_hostnameuerror((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyucreate_streaming_connectionts  &N(u__name__u __module__u__doc__uNoneucreate_streaming_connection(u __locals__((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyuStreamingConnectionMixinls uStreamingConnectionMixincBs&|EeZdZd„Zd„ZdS(uGThis module's adaptation of `.nbio_interface.AbstractIOReference` cCs|j|_dS(uZ :param subject: subject of the reference containing a `cancel()` method N(ucancelu_cancel(uselfusubject((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu__init__šscCs |jƒS(uCancel pending operation :returns: False if was already done or cancelled; True otherwise :rtype: bool (u_cancel(uself((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyucancel¡sN(u__name__u __module__u__doc__u__init__ucancel(u __locals__((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_AsyncServiceAsyncHandle•s  u_AsyncServiceAsyncHandlecBsƒ|EeZdZdZdZdZdZd„Zed„ƒZ d„Z d„Z ed „ƒZ ed „ƒZ ed „ƒZd S( uú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ƒWnrtk rœ}zRttdƒs\tjdƒn.dj|||ƒ}tj |ƒt |ƒ‚WYdd}~XnX||_ ||_ ||_ ||_|j|_d|_dS(u :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()` uon_doneiu inet_ptonu8Unable to check resolved address: no socket.inet_pton().u9Invalid or unresolved IP address {!r} for socket {}: {!r}NF(ucheck_callback_argusocketu inet_ptonufamilyu Exceptionuhasattru_LOGGERudebuguformatuerroru ValueErroru_nbiou_socku_addru_on_doneu_STATE_NOT_STARTEDu_stateuFalseu_watching_socket_events(uselfunbiousocku resolved_addruon_doneuerrorumsg((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu__init__·s"        cCs2|jr.d|_|jj|jjƒƒndS(u'Remove socket watcher, if any NF(u_watching_socket_eventsuFalseu_nbiou remove_writeru_sockufileno(uself((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_cleanupÛs  cCsP|j|jks'td|jfƒ‚|j|_|jj|jƒt|ƒS(uZStart asynchronous connection establishment. :rtype: AbstractIOReference u:_AsyncSocketConnector.start(): expected _STATE_NOT_STARTED(u_stateu_STATE_NOT_STARTEDuAssertionErroru _STATE_ACTIVEu_nbiouadd_callback_threadsafeu _start_asyncu_AsyncServiceAsyncHandle(uself((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyustartäs  cCsb|j|jkrE|j|_tjd|j|jƒ|jƒdStjd|j|jƒdS(u¾Cancel pending connection request without calling user's completion callback. :returns: False if was already done or cancelled; True otherwise :rtype: bool u-User canceled connection request for %s to %suD_AsyncSocketConnector cancel requested when not ACTIVE: state=%s; %sTF( u_stateu _STATE_ACTIVEu_STATE_CANCELEDu_LOGGERudebugu_socku_addru_cleanupuTrueuFalse(uself((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyucancelõs   cCs‘tjd||jƒt|ttdƒfƒsCtd|fƒ‚|j|j ksjtd|jfƒ‚|j |_|j ƒ|j |ƒdS(u¹Advance to COMPLETED state, remove socket watcher, and invoke user's completion callback. :param BaseException | None result: value to pass in user's callback u0_AsyncSocketConnector._report_completion(%r); %suP_AsyncSocketConnector._report_completion() expected exception or None as result.uF_AsyncSocketConnector._report_completion() expected _STATE_NOT_STARTEDN( u_LOGGERudebugu_socku isinstanceu BaseExceptionutypeuNoneuAssertionErroru_stateu _STATE_ACTIVEu_STATE_COMPLETEDu_cleanupu_on_done(uselfuresult((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_report_completion s     cCsf|j|jkr5tjd|j|j|jƒdSy|jj|jƒWn…ttj j fk rÓ}zYt |tj j ƒr”|j t kr”n-tjd|j|j|ƒ|j|ƒdSWYdd}~XnXy#|jj|jjƒ|jƒWnLtk rE}z,tjd|j|ƒ|j|ƒdSWYdd}~XnXd|_tjd|jƒdS(u“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 uJAbandoning sock=%s connection establishment to %s due to inactive state=%sNu%s.connect(%s) failed: %ruasync.set_writer(%s) failed: %ru/Connection-establishment is in progress for %s.T(u_stateu _STATE_ACTIVEu_LOGGERudebugu_socku_addruconnectu Exceptionupikaucompatu SOCKET_ERRORu isinstanceuerrnou(_CONNECTION_IN_PROGRESS_SOCK_ERROR_CODESuerroru_report_completionu_nbiou set_writerufilenou _on_writableu exceptionuTrueu_watching_socket_events(uselfuerror((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu _start_async s2  #   cCsº|j|jkr/tjd|j|jƒdS|jjtjtjƒ}|sltj d|jƒd}n=t j |ƒ}tjd|j||ƒt jj||ƒ}|j|ƒdS(uwCalled when socket connects or fails to. Check for predicament and invoke user's completion callback. u_Socket connection-establishment event watcher called in inactive state (ignoring): %s; state=%sNuSocket connected: %su+Socket failed to connect: %s; error=%s (%s)(u_stateu _STATE_ACTIVEu_LOGGERuerroru_socku getsockoptusocketu SOL_SOCKETuSO_ERRORuinfouNoneuosustrerrorupikaucompatu SOCKET_ERRORu_report_completion(uselfu error_codeuresultu error_msg((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu _on_writableGs   N(u__name__u __module__u__doc__u_STATE_NOT_STARTEDu _STATE_ACTIVEu_STATE_CANCELEDu_STATE_COMPLETEDu__init__u_log_exceptionsu_cleanupustartucancelu_report_completionu _start_asyncu _on_writable(u __locals__((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_AsyncSocketConnector«s  $  'u_AsyncSocketConnectorcBs’|EeZdZdZdZdZdZd„Zed„ƒZ d„Z d„Z ed „ƒZ ed „ƒZ ed „ƒZed „ƒZd S(uù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Ä}ztdj||ƒƒ‚WYdd}~XnX||_ ||_ ||_ ||_ ||_||_|j|_d|_dS(u 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: uprotocol_factoryuon_doneu8Expected ssl_context=None | ssl.SSLContext, but got {!r}u?Non-None server_hostname must not be passed without ssl contextuEExpected connected socket, but getpeername() failed: error={!r}; {}; NF(ucheck_callback_argu isinstanceutypeuNoneusslu SSLContextu ValueErroruformatu getpeernameu Exceptionu_nbiou_protocol_factoryu_socku _ssl_contextu_server_hostnameu_on_doneu_STATE_NOT_STARTEDu_stateuFalseu_watching_socket(uselfunbiouprotocol_factoryusocku ssl_contextuserver_hostnameuon_doneuerror((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu__init__ps*   "       cCs tjd|ƒ|jrmtjd||jƒd|_|jj|jjƒƒ|jj|jjƒƒnzu|rátjd||jƒy|jj ƒWqát k rÝ}ztj d||jƒ‚WYdd}~XqáXnWdd|_d|_d|_ d|_d|_d|_XdS(ueCancel pending async operations, if any :param bool close: close the socket if true u"_AsyncStreamConnector._cleanup(%r)u5_AsyncStreamConnector._cleanup(%r): removing RdWr; %su6_AsyncStreamConnector._cleanup(%r): closing socket; %su"_sock.close() failed: error=%r; %sNF(u_LOGGERudebugu_watching_socketu_sockuFalseu_nbiou remove_readerufilenou remove_writerucloseu Exceptionu exceptionuNoneu_protocol_factoryu _ssl_contextu_server_hostnameu_on_done(uselfucloseuerror((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_cleanup s2           cCsctjd|jƒ|j|jks:td|jfƒ‚|j|_|jj|j ƒt |ƒS(uCKick off the workflow :rtype: AbstractIOReference u!_AsyncStreamConnector.start(); %su9_AsyncStreamConnector.start() expected _STATE_NOT_STARTED( u_LOGGERudebugu_socku_stateu_STATE_NOT_STARTEDuAssertionErroru _STATE_ACTIVEu_nbiouadd_callback_threadsafeu _start_asyncu_AsyncServiceAsyncHandle(uself((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyustartÃs cCsb|j|jkrE|j|_tjd|jƒ|jddƒdStjd|j|jƒdS(u¾Cancel pending connection request without calling user's completion callback. :returns: False if was already done or cancelled; True otherwise :rtype: bool u%User canceled streaming linkup for %sucloseuD_AsyncStreamConnector cancel requested when not ACTIVE: state=%s; %sTF( u_stateu _STATE_ACTIVEu_STATE_CANCELEDu_LOGGERudebugu_socku_cleanupuTrueuFalse(uself((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyucancelÖs cCsÚtjd||jƒt|ttfƒsCtd||jfƒ‚|j|jksjtd|jfƒ‚|j |_zCy|j |ƒWn+t k r·tj d|j |ƒ‚YnXWd|jdt|tƒƒXdS(uAdvance 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 u0_AsyncStreamConnector._report_completion(%r); %suQ_AsyncStreamConnector._report_completion() expected exception or tuple as result.uA_AsyncStreamConnector._report_completion() expected _STATE_ACTIVEu%r: _on_done(%r) failed.Nuclose(u_LOGGERudebugu_socku isinstanceu BaseExceptionutupleuAssertionErroru_stateu _STATE_ACTIVEu_STATE_COMPLETEDu_on_doneu Exceptionu exceptionu_report_completionu_cleanup(uselfuresult((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_report_completionês"       cCstjd|jƒ|j|jkrBtjd|j|jƒdS|jdkr^|jƒn£tjd|jƒy7|jj|jdd dd dd d|j ƒ|_WnLt k rö}z,tj d |j|ƒ|j |ƒdSWYdd}~XnX|jƒdS( u’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 u(_AsyncStreamConnector._start_async(); %suMAbandoning streaming linkup due to inactive state transition; state=%s; %s; .NuStarting SSL handshake on %su server_sideudo_handshake_on_connectusuppress_ragged_eofsuserver_hostnameuSSL wrap_socket(%s) failed: %rF(u_LOGGERudebugu_socku_stateu _STATE_ACTIVEu _ssl_contextuNoneu_linkupu wrap_socketuFalseu_server_hostnameu Exceptionu exceptionu_report_completionu_do_ssl_handshake(uselfuerror((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu _start_async s,    c4Cs÷tjdƒd}y™y|jƒ}Wn>tk rf}ztjd||jƒ‚WYdd}~XnX|jdkrÖyt|j||j ƒ}Wq3tk rÒ}ztjd||jƒ‚WYdd}~Xq3Xn]yt |j||j ƒ}Wn>tk r2}ztjd||jƒ‚WYdd}~XnXtjd|ƒy|j |ƒWnAtk r—}z!tjd|||jƒ‚WYdd}~XnXtjd||ƒWn+tk rÙ}z |}WYdd}~Xn X||f}|j |ƒdS( u}Connection is ready: instantiate and link up transport and protocol, and invoke user's completion callback. u_AsyncStreamConnector._linkup()u'protocol_factory() failed: error=%r; %sNu%PlainTransport() failed: error=%r; %su#SSLTransport() failed: error=%r; %su_linkup(): created transport %ru1protocol.connection_made(%r) failed: error=%r; %su2_linkup(): introduced transport to protocol %r; %r( u_LOGGERudebuguNoneu_protocol_factoryu Exceptionu exceptionu_socku _ssl_contextu_AsyncPlaintextTransportu_nbiou_AsyncSSLTransportuconnection_madeu_report_completion(uselfu transportuprotocoluerroruresult((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_linkup0sL          cCs7tjdƒ|j|jkr<tjd|j|jƒdSd }y.y|jjƒWnýtjk rU}zÚ|j tj kr×tjd|jƒd |_ |j j|jjƒ|jƒ|j j|jjƒƒnl|j tjkr@tjd|jƒd |_ |j j|jjƒ|jƒ|j j|jjƒƒn‚WYdd}~XnXd }tjd|jƒWnLtk r¾}z,tjd||jƒ|j|ƒdSWYdd}~XnX|r3tjd|jƒ|j j|jjƒƒ|j j|jjƒƒd |_ tjd |jƒ|jƒndS( uJPerform asynchronous SSL handshake on the already wrapped socket u)_AsyncStreamConnector._do_ssl_handshake()u`_do_ssl_handshake: Abandoning streaming linkup due to inactive state transition; state=%s; %s; .NuSSL handshake wants read; %s.uSSL handshake wants write. %su(SSL handshake completed successfully: %su%SSL do_handshake failed: error=%r; %su8_do_ssl_handshake: removing watchers ahead of linkup: %su=_do_ssl_handshake: pre-linkup removal of watchers is done; %sFT(u_LOGGERudebugu_stateu _STATE_ACTIVEu_sockuFalseu do_handshakeussluSSLErroruerrnouSSL_ERROR_WANT_READuTrueu_watching_socketu_nbiou set_readerufilenou_do_ssl_handshakeu remove_writeruSSL_ERROR_WANT_WRITEu set_writeru remove_readeruinfou Exceptionu exceptionu_report_completionu_linkup(uselfudoneuerror((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_do_ssl_handshakejsX             N(u__name__u __module__u__doc__u_STATE_NOT_STARTEDu _STATE_ACTIVEu_STATE_CANCELEDu_STATE_COMPLETEDu__init__u_log_exceptionsu_cleanupustartucancelu_report_completionu _start_asyncu_linkupu_do_ssl_handshake(u __locals__((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_AsyncStreamConnectords  0#   &:u_AsyncStreamConnectorcBsð|EeZdZdZdZdZdZdZdZGd„d e ƒZ d „Z d „Z d „Z d „Zd„Zd„Zd„Zeed„ƒƒZeed„ƒƒZed„ƒZed„ƒZed„ƒZed„ƒZdS(uIBase class for `_AsyncPlaintextTransport` and `_AsyncSSLTransport`. iiiiiiidcs#|EeZdZ‡fd†Z‡S(uNWe raise this internally when EOF (empty read) is detected on input. cs ttj|ƒjddƒdS(NiuEnd of input stream (EOF)iÿÿÿÿ(usuperu_AsyncTransportBaseu RxEndOfFileu__init__(uself(u __class__(uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu__init__½s(u__name__u __module__u__doc__u__init__(u __locals__((u __class__uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu RxEndOfFile¸s u RxEndOfFilecCsStjd|ƒ||_||_||_|j|_tjƒ|_ d|_ dS(u~ :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: u _AsyncTransportBase.__init__: %siN( u_LOGGERudebugu_socku _protocolu_nbiou _STATE_ACTIVEu_stateu collectionsudequeu _tx_buffersu_tx_buffered_byte_count(uselfusockuprotocolunbio((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu__init__Ás     cCs*tjd|j|jƒ|jdƒdS(uClose 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 u+Aborting transport connection: state=%s; %sN(u_LOGGERuinfou_stateu_socku_initiate_abortuNone(uself((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyuabortÔs cCs|jS(uReturn the protocol linked to this transport. :rtype: pika.adapters.utils.nbio_interface.AbstractStreamProtocol (u _protocol(uself((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu get_protocolàscCs|jS(ue :returns: Current size of output data buffered by the transport :rtype: int (u_tx_buffered_byte_count(uself((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyuget_write_buffer_sizeçscCs|s7tjd|j|jƒtdj|ƒƒ‚n|j|jkrftjd|j|jƒdS|jj |ƒ|j t |ƒ7_ dS(u–Buffer the given data until it can be sent asynchronously. :param bytes data: :raises ValueError: if called with empty data u,write() called with empty data: state=%s; %su#write() called with empty data {!r}u;Ignoring write() called during inactive state: state=%s; %sN( u_LOGGERuerroru_stateu_socku ValueErroruformatu _STATE_ACTIVEudebugu _tx_buffersuappendu_tx_buffered_byte_countulen(uselfudata((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_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Î}ztj d||jƒ‚WYdd}~Xq Xq WdS(uï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 iuSocket EOF; %su-protocol.data_received() failed: error=%r; %sN(u_stateu _STATE_ACTIVEu_MAX_CONSUME_BYTESu_sigint_safe_recvu_socku_MAX_RECV_BYTESulenu_LOGGERuerroru RxEndOfFileu _protocolu data_receivedu Exceptionu exception(uselfubytes_consumedudatauerror((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_consumes cCsÀx¹|jr»|j|j|jdƒ}|jjƒ}|t|ƒkrtjd|t|ƒƒ|jj||d…ƒn|j|8_|jdkst d|j|j fƒ‚qWdS(uUtility 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 iu/Partial send, requeing remaining data; %s of %sNu7_AsyncTransportBase._produce() tx buffer size underflow( u _tx_buffersu_sigint_safe_sendu_sockupopleftulenu_LOGGERudebugu appendleftu_tx_buffered_byte_countuAssertionErroru_state(uselfunum_bytes_sentuchunk((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_produce(s   cCs |j|ƒS(umReceive 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 (urecv(usocku max_bytes((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_sigint_safe_recvAs cCs |j|ƒS(u@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 (usend(usockudata((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_sigint_safe_sendPs cCsq|j|jkrmtjd|j|jƒ|jj|jjƒƒ|jj|jjƒƒ|j j ƒndS(u2Unregister the transport from I/O events u$Deactivating transport: state=%s; %sN( u_stateu _STATE_ACTIVEu_LOGGERuinfou_socku_nbiou remove_readerufilenou remove_writeru _tx_buffersuclear(uself((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu _deactivate_s  c Cs˜|j|jkr”tjd|j|jƒy|jjtjƒWntj j k r\YnX|jj ƒd|_d|_ d|_|j|_ndS(u{Close the transport's socket and unlink the transport it from references to other assets (protocol, etc.) u4Closing transport socket and unlinking: state=%s; %sN(u_stateu_STATE_COMPLETEDu_LOGGERuinfou_sockushutdownusocketu SHUT_RDWRupikaucompatu SOCKET_ERRORucloseuNoneu _protocolu_nbio(uself((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_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(u£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. uo_AsyncTransportBase._initate_abort(): Initiating abrupt asynchronous transport shutdown: state=%s; error=%r; %suB_AsyncTransportBase._initate_abort() expected non-_STATE_COMPLETEDNuM_AsyncTransportBase._initiate_abort(): ignoring - user-abort already pending.uD_AsyncTransportBase._initate_abort() expected _STATE_ABORTED_BY_USER(u_LOGGERuinfou_stateu_socku_STATE_COMPLETEDuAssertionErroru _deactivateuNoneu_STATE_ABORTED_BY_USERudebugu _STATE_ACTIVEu _STATE_FAILEDu_nbiouadd_callback_threadsafeu functoolsupartialu_connection_lost_notify_async(uselfuerror((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_initiate_abort~s,      cCsãtjd|j|ƒ|j|jkr,dS|dk ru|j|jkru|j|jksqtd|jfƒ‚dSz\y|jj |ƒWnAt k rÏ}z!tj d|||j ƒ‚WYdd}~XnXWd|j ƒXdS(už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. u1Concluding transport shutdown: state=%s; error=%rNuS_AsyncTransportBase._connection_lost_notify_async() expected _STATE_ABORTED_BY_USERu/protocol.connection_lost(%r) failed: exc=%r; %s(u_LOGGERudebugu_stateu_STATE_COMPLETEDuNoneu _STATE_FAILEDu_STATE_ABORTED_BY_USERuAssertionErroru _protocoluconnection_lostu Exceptionu exceptionu_socku_close_and_finalize(uselfuerroruexc((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_connection_lost_notify_async³s"   Ni(u__name__u __module__u__doc__u _STATE_ACTIVEu _STATE_FAILEDu_STATE_ABORTED_BY_USERu_STATE_COMPLETEDu_MAX_RECV_BYTESu_MAX_CONSUME_BYTESuOSErroru RxEndOfFileu__init__uabortu get_protocoluget_write_buffer_sizeu_buffer_tx_datau_consumeu_produceu staticmethodu_retry_on_sigintu_sigint_safe_recvu_sigint_safe_sendu_log_exceptionsu _deactivateu_close_and_finalizeu_initiate_abortu_connection_lost_notify_async(u __locals__((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_AsyncTransportBase¨s.      %  5u_AsyncTransportBasecsJ|EeZdZ‡fd†Zd„Zed„ƒZed„ƒZ‡S(u`Implementation of `nbio_interface.AbstractStreamTransport` for a plaintext connection. cs?tt|ƒj|||ƒ|jj|jjƒ|jƒdS(u{ :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(usuperu_AsyncPlaintextTransportu__init__u_nbiou set_readeru_sockufilenou_on_socket_readable(uselfusockuprotocolunbio(u __class__(uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu__init__Ýs cCsŸ|j|jkr/tjd|j|jƒdS|sMtd||jfƒ‚|jƒsŽ|jj|jj ƒ|j ƒtjd|jƒn|j |ƒdS(u–Buffer the given data until it can be sent asynchronously. :param bytes data: :raises ValueError: if called with empty data u;Ignoring write() called during inactive state: state=%s; %sNu7_AsyncPlaintextTransport.write(): empty data from user.u!Turned on writability watcher: %s( u_stateu _STATE_ACTIVEu_LOGGERudebugu_sockuAssertionErroruget_write_buffer_sizeu_nbiou set_writerufilenou_on_socket_writableu_buffer_tx_data(uselfudata((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyuwriteís  cCsâ|j|jkr/tjd|j|jƒdSy|jƒWnp|jk ry|jjƒ}WnHt k r­}z(tj d||jƒ|j |ƒWYdd}~XnVX|rãtj d|jƒ|j j|jjƒƒn tj d|jƒ|j dƒYn×t tjjfk r¯}z}t|tjjƒr_|jtkr_tjd|jƒn>tj d||jdjtjtjƒŒƒƒ|j |ƒWYdd}~Xn/X|j|jkrÞtjd |j|jƒndS( uÜ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. uEIgnoring readability notification due to inactive state: state=%s; %sNu,protocol.eof_received() failed: error=%r; %su0protocol.eof_received() elected to keep open: %su,protocol.eof_received() elected to close: %suRecv would block on %sua_AsyncBaseTransport._consume() failed, aborting connection: error=%r; sock=%s; Caller's stack: %suu>Leaving Plaintext consumer due to inactive state: state=%s; %s(u_stateu _STATE_ACTIVEu_LOGGERudebugu_socku_consumeu RxEndOfFileu _protocolu eof_receivedu Exceptionu exceptionu_initiate_abortuinfou_nbiou remove_readerufilenouNoneupikaucompatu SOCKET_ERRORu isinstanceuerrnou_TRY_IO_AGAIN_SOCK_ERROR_CODESujoinu tracebackuformat_exceptionusysuexc_info(uselfu keep_openuerror((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_on_socket_readablesH      cCsC|j|jkr/tjd|j|jƒdS|jsMtd|jfƒ‚y|jƒWn©tt j j fk r}z}t |t j j ƒr¶|j tkr¶tjd|jƒn>tjd||jdjtjtjƒŒƒƒ|j|ƒWYdd}~Xn9X|js?|jj|jjƒƒtjd|jƒndS(u-Handle writable socket notification uEIgnoring writability notification due to inactive state: state=%s; %sNuP_AsyncPlaintextTransport._on_socket_writable() called, but _tx_buffers is empty.uSend would block on %sua_AsyncBaseTransport._produce() failed, aborting connection: error=%r; sock=%s; Caller's stack: %suu"Turned off writability watcher: %s(u_stateu _STATE_ACTIVEu_LOGGERudebugu_socku _tx_buffersuAssertionErroru_produceu Exceptionupikaucompatu SOCKET_ERRORu isinstanceuerrnou_TRY_IO_AGAIN_SOCK_ERROR_CODESu exceptionujoinu tracebackuformat_exceptionusysuexc_infou_initiate_abortu_nbiou remove_writerufileno(uselfuerror((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_on_socket_writable7s,   (u__name__u __module__u__doc__u__init__uwriteu_log_exceptionsu_on_socket_readableu_on_socket_writable(u __locals__((u __class__uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_AsyncPlaintextTransport×s  4u_AsyncPlaintextTransportcst|EeZdZ‡fd†Zd„Zed„ƒZed„ƒZe‡fd†ƒZe‡fd†ƒZ ‡S(u\Implementation of `.nbio_interface.AbstractStreamTransport` for an SSL connection. csgtt|ƒj|||ƒ|j|_d|_|jj|j j ƒ|j ƒ|jj |j ƒdS(u{ :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( usuperu_AsyncSSLTransportu__init__u_consumeu_ssl_readable_actionuNoneu_ssl_writable_actionu_nbiou set_readeru_sockufilenou_on_socket_readableuadd_callback_threadsafe(uselfusockuprotocolunbio(u __class__(uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu__init__as   cCs¨|j|jkr/tjd|j|jƒdS|jƒdk}|j|ƒ|r¤|jdkr¤|j |_|j j |jj ƒ|j ƒtjd|jƒndS(u–Buffer the given data until it can be sent asynchronously. :param bytes data: :raises ValueError: if called with empty data u;Ignoring write() called during inactive state: state=%s; %sNiu!Turned on writability watcher: %s(u_stateu _STATE_ACTIVEu_LOGGERudebugu_sockuget_write_buffer_sizeu_buffer_tx_datau_ssl_writable_actionuNoneu_produceu_nbiou set_writerufilenou_on_socket_writable(uselfudatautx_buffer_was_empty((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyuwriteus  cCs›|j|jkr/tjd|j|jƒdS|jr~y|jƒWq—tk rz}z|j|ƒWYdd}~Xq—Xntjd|j|jƒdS(u+Handle readable socket indication uEIgnoring readability notification due to inactive state: state=%s; %sNu>SSL readable action was suppressed: ssl_writable_action=%r; %s( u_stateu _STATE_ACTIVEu_LOGGERudebugu_socku_ssl_readable_actionu Exceptionu_initiate_abortu_ssl_writable_action(uselfuerror((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_on_socket_readable‹s #cCs›|j|jkr/tjd|j|jƒdS|jr~y|jƒWq—tk rz}z|j|ƒWYdd}~Xq—Xntjd|j|jƒdS(u-Handle writable socket notification uEIgnoring writability notification due to inactive state: state=%s; %sNu>SSL writable action was suppressed: ssl_readable_action=%r; %s( u_stateu _STATE_ACTIVEu_LOGGERudebugu_socku_ssl_writable_actionu Exceptionu_initiate_abortu_ssl_readable_action(uselfuerror((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_on_socket_writable¡s #cs5d}ytt|ƒjƒWn²tjk rÑ}z|jtjkr]tj d|j ƒnb|jtj kr‹tj d|j ƒd}n4tj d||j djtjtjƒŒƒƒ‚WYdd}~XnCX|j|jkrtj d|j|j ƒdS|jj|jƒ|r‹|jsE|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!r1|j r1|j"|_|jj|j jƒ|jƒndS( u“[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 uSSL ingester wants read: %suSSL ingester wants write: %sua_AsyncBaseTransport._consume() failed, aborting connection: error=%r; sock=%s; Caller's stack: %suNu8Leaving SSL consumer due to inactive state: state=%s; %sTF(#uTrueusuperu_AsyncSSLTransportu_consumeussluSSLErroruerrnouSSL_ERROR_WANT_READu_LOGGERudebugu_sockuSSL_ERROR_WANT_WRITEuFalseu exceptionujoinu tracebackuformat_exceptionusysuexc_infou_stateu _STATE_ACTIVEu_nbiouadd_callback_threadsafeu_on_socket_readableu_ssl_readable_actionu set_readerufilenou_ssl_writable_actionu remove_writeruNoneu set_writeru_on_socket_writableu remove_readeru _tx_buffersu_produce(uselfunext_consume_on_readableuerror(u __class__(uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_consume·sL           cs5d}ytt|ƒjƒWn¸tjk r×}z•|jtjkrctj d|j ƒd }nb|jtj kr‘tj d|j ƒd}n4tjd||j djtjtjƒŒƒƒ‚WYdd}~Xn&X|j sýtdt|jƒfƒ‚|jr|dk s'td|jfƒ‚|rž|jsX|jj|j jƒ|jƒn|j|_|j|jkr|jj|j jƒƒd|_qqÂ|jsÉ|jj|j jƒ|j ƒn|j|_|jrÂ|jj!|j jƒƒd|_qÂn¼|j|jkrd|jj|j jƒƒd|_|j|jksÂtd|jfƒ‚n^|j|jks td d |jd |jd |jfƒ‚d|_|jj!|j jƒƒ|js |j"|_|jj|j jƒ|j ƒ|jj#|j ƒn%|j j$ƒr1|jj#|j ƒndS(uI[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 uSSL emitter wants read: %suSSL emitter wants write: %sua_AsyncBaseTransport._produce() failed, aborting connection: error=%r; sock=%s; Caller's stack: %suNu__AsyncSSLTransport._produce(): no exception from parent class, but data remains in _tx_buffers.uE_AsyncSSLTransport._produce(): next_produce_on_writable is still Noneur_AsyncSSLTransport._produce(): with empty tx_buffers, writable_action cannot be _produce when readable is _produceu_AsyncSSLTransport._produce(): with empty tx_buffers, expected writable_action as _produce when readable_action is not _produceuwritable_action:ureadable_action:ustate:FT(%uNoneusuperu_AsyncSSLTransportu_produceussluSSLErroruerrnouSSL_ERROR_WANT_READu_LOGGERudebugu_sockuFalseuSSL_ERROR_WANT_WRITEuTrueu exceptionujoinu tracebackuformat_exceptionusysuexc_infou _tx_buffersuAssertionErrorulenu_stateu_ssl_writable_actionu_nbiou set_writerufilenou_on_socket_writableu_ssl_readable_actionu remove_readeru set_readeru_on_socket_readableu remove_writeru_consumeuadd_callback_threadsafeupending(uselfunext_produce_on_writableuerror(u __class__(uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_producesr                 ( u__name__u __module__u__doc__u__init__uwriteu_log_exceptionsu_on_socket_readableu_on_socket_writableu_consumeu_produce(u __locals__((u __class__uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyu_AsyncSSLTransport[s  Iu_AsyncSSLTransport((u__doc__u collectionsuerrnou functoolsuloggingunumbersuosusocketusslusysu tracebacku"pika.adapters.utils.nbio_interfaceuAbstractIOReferenceuAbstractStreamTransportu pika.compatupikaupika.diagnostic_utilsuEAGAINu EWOULDBLOCKu_TRY_IO_AGAIN_SOCK_ERROR_CODESu EINPROGRESSu(_CONNECTION_IN_PROGRESS_SOCK_ERROR_CODESu getLoggeru__name__u_LOGGERudiagnostic_utilsucreate_log_exception_decoratoru_log_exceptionsucheck_callback_argu check_fd_argu_retry_on_sigintuobjectuSocketConnectionMixinuStreamingConnectionMixinu_AsyncServiceAsyncHandleu_AsyncSocketConnectoru_AsyncStreamConnectoru_AsyncTransportBaseu_AsyncPlaintextTransportu_AsyncSSLTransport(((uJ/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/io_services_utils.pyusB               )¹ÿE  ÿ/„