o eiF@s>dZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl m Z m Z ddlZddlZejejfZejejfZeeZejeZddZddZdd ZGd d d eZ Gd d d eZ!Gddde Z"GdddeZ#GdddeZ$Gddde Z%Gddde%Z&Gddde%Z'dS)z^Utilities for implementing `nbio_interface.AbstractIOServices` for pika connection adapters. N)AbstractIOReferenceAbstractStreamTransportcCst|s td||dS)zRaise TypeError if callback is not callable :param callback: callback to check :param name: Name to include in exception text :raises TypeError: z!{} must be callable, but got {!r}N)callable TypeErrorformat)callbacknamer d/var/www/html/pca-backend/venv/lib/python3.10/site-packages/pika/adapters/utils/io_services_utils.pycheck_callback_arg,s r cCst|tjs td|dS)zqRaise TypeError if file descriptor is not an integer :param fd: file descriptor :raises TypeError: z0Paramter must be a file descriptor, but got {!r}N) isinstancenumbersIntegralrr)fdr r r check_fd_arg9s rcstfdd}|S)z0Function decorator for retrying on SIGINT. c sJ z|i|WStjjy$}z |jtjkrWYd}~qd}~ww)zWrapper for decorated functionTN)pikacompat SOCKET_ERRORerrnoEINTR)argskwargserrorfuncr r retry_sigint_wrapJs  z+_retry_on_sigint..retry_sigint_wrap) functoolswraps)rrr rr _retry_on_sigintEs rc@seZdZdZddZdS)SocketConnectionMixinzImplements `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`. cCst||||dS)z[Implement :py:meth:`.nbio_interface.AbstractIOServices.connect_socket()`. )nbiosock resolved_addron_done)_AsyncSocketConnectorstart)selfr!r"r#r r r connect_socketbs z$SocketConnectionMixin.connect_socketN)__name__ __module__ __qualname____doc__r'r r r r rYs rc@seZdZdZ  dddZdS)StreamingConnectionMixinzImplements `.nbio_interface.AbstractIOServices.create_streaming_connection()` on top of `.nbio_interface.AbstractFileDescriptorServices` and basic `nbio_interface.AbstractIOServices` services. Nc Csz t||||||dWSty@}z'td||z|Wty;}z td||WYd}~d}~wwd}~ww)zhImplement :py:meth:`.nbio_interface.AbstractIOServices.create_streaming_connection()`. )r protocol_factoryr! ssl_contextserver_hostnamer#z*create_streaming_connection(%s) failed: %rz%s.close() failed: %rN)_AsyncStreamConnectorr% Exception_LOGGERrclose)r&r-r!r#r.r/rr r r create_streaming_connectionts2  z4StreamingConnectionMixin.create_streaming_connection)NN)r(r)r*r+r4r r r r r,ls  r,c@s eZdZdZddZddZdS)_AsyncServiceAsyncHandlezGThis module's adaptation of `.nbio_interface.AbstractIOReference` cCs |j|_dS)zZ :param subject: subject of the reference containing a `cancel()` method N)cancel_cancel)r&subjectr r r __init__s z!_AsyncServiceAsyncHandle.__init__cCs|S)zCancel pending operation :returns: False if was already done or cancelled; True otherwise :rtype: bool )r7r&r r r r6sz_AsyncServiceAsyncHandle.cancelN)r(r)r*r+r9r6r r r r r5s r5c@sheZdZdZdZdZdZdZddZe dd Z d d Z d d Z e ddZ e ddZe ddZdS)r$zConnects the given non-blocking socket asynchronously using `.nbio_interface.AbstractFileDescriptorServices` and basic `.nbio_interface.AbstractIOServices`. Used for implementing `.nbio_interface.AbstractIOServices.connect_socket()`. rc Cst|dz t|j|dWn-ty=}z!ttds#tdnd|||}t |t |WYd}~nd}~ww||_ ||_ ||_ ||_|j|_d|_dS)a :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()` r#r inet_ptonz8Unable to check resolved address: no socket.inet_pton().z9Invalid or unresolved IP address {!r} for socket {}: {!r}NF)r socketr>familyr1hasattrr2debugrr ValueError_nbio_sock_addr_on_done_STATE_NOT_STARTED_state_watching_socket_events)r&r r!r"r#rmsgr r r r9s0      z_AsyncSocketConnector.__init__cCs&|jrd|_|j|jdSdS)z'Remove socket watcher, if any FN)rJrD remove_writerrEfilenor:r r r _cleanupsz_AsyncSocketConnector._cleanupcCs8|j|jks Jd|jf|j|_|j|jt|S)zZStart asynchronous connection establishment. :rtype: AbstractIOReference z:_AsyncSocketConnector.start(): expected _STATE_NOT_STARTED)rIrH _STATE_ACTIVErDadd_callback_threadsafe _start_asyncr5r:r r r r%sz_AsyncSocketConnector.startcCsH|j|jkr|j|_td|j|j|dStd|j|jdS)Cancel pending connection request without calling user's completion callback. :returns: False if was already done or cancelled; True otherwise :rtype: bool z-User canceled connection request for %s to %sTzD_AsyncSocketConnector cancel requested when not ACTIVE: state=%s; %sF)rIrO_STATE_CANCELEDr2rBrErFrNr:r r r r6s z_AsyncSocketConnector.cancelcCsftd||jt|ttdfsJd|f|j|jks$Jd|jf|j|_| | |dS)zAdvance to COMPLETED state, remove socket watcher, and invoke user's completion callback. :param BaseException | None result: value to pass in user's callback z0_AsyncSocketConnector._report_completion(%r); %sNzP_AsyncSocketConnector._report_completion() expected exception or None as result.zF_AsyncSocketConnector._report_completion() expected _STATE_NOT_STARTED) r2rBrEr BaseExceptiontyperIrO_STATE_COMPLETEDrNrGr&resultr r r _report_completion sz(_AsyncSocketConnector._report_completionc Cs |j|jkrtd|j|j|jdSz |j|jWn9ttj j fyU}z)t |tj j r5|j t vr5ntd|j|j|||WYd}~dSWYd}~nd}~wwz |j|j|jWn ty}ztd|j|||WYd}~dSd}~wwd|_td|jdS)zCalled 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 zJAbandoning sock=%s connection establishment to %s due to inactive state=%sNz%s.connect(%s) failed: %rzasync.set_writer(%s) failed: %rTz/Connection-establishment is in progress for %s.)rIrOr2rBrErFconnectr1rrrr r(_CONNECTION_IN_PROGRESS_SOCK_ERROR_CODESrrYrD set_writerrM _on_writable exceptionrJr&rr r r rQ sD         z"_AsyncSocketConnector._start_asynccCs|j|jkrtd|j|jdS|jtjtj}|s&t d|jd}nt |}td|j||t j ||}||dS)zwCalled when socket connects or fails to. Check for predicament and invoke user's completion callback. z_Socket connection-establishment event watcher called in inactive state (ignoring): %s; state=%sNzSocket connected: %sz+Socket failed to connect: %s; error=%s (%s))rIrOr2rrE getsockoptr? SOL_SOCKETSO_ERRORinfoosstrerrorrrrrY)r& error_coderX error_msgr r r r]Gs"  z"_AsyncSocketConnector._on_writableN)r(r)r*r+rHrOrSrVr9_log_exceptionsrNr%r6rYrQr]r r r r r$s"$   &r$c@steZdZdZdZdZdZdZddZe dd Z d d Z d d Z e ddZ e ddZe ddZe ddZdS)r0zPerforms 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()`. rr;r<r=c Cst|dt|dt|tdtjfstd||dur'|dur'tdz|WntyA}ztd||d}~ww||_ ||_ ||_ ||_ ||_ ||_|j|_d|_dS)a 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-r#Nz8Expected ssl_context=None | ssl.SSLContext, but got {!r}z?Non-None server_hostname must not be passed without ssl contextzEExpected connected socket, but getpeername() failed: error={!r}; {}; F)r r rUssl SSLContextrCr getpeernamer1rD_protocol_factoryrE _ssl_context_server_hostnamerGrHrI_watching_socket)r&r r-r!r.r/r#rr r r r9ps2    z_AsyncStreamConnector.__init__c Cstd||jr&td||jd|_|j|j|j|jzQ|rNtd||jz|jWn*t yM}z t d||jd}~wwWd|_d|_d|_ d|_ d|_ d|_dSWd|_d|_d|_ d|_ d|_ d|_dSd|_d|_d|_ d|_ d|_ d|_w)zeCancel pending async operations, if any :param bool close: close the socket if true z"_AsyncStreamConnector._cleanup(%r)z5_AsyncStreamConnector._cleanup(%r): removing RdWr; %sFz6_AsyncStreamConnector._cleanup(%r): closing socket; %sz"_sock.close() failed: error=%r; %sN)r2rBrorErD remove_readerrMrLr3r1r^rlrmrnrG)r&r3rr r r rNsX    z_AsyncStreamConnector._cleanupcCsFtd|j|j|jksJd|jf|j|_|j|jt |S)zCKick off the workflow :rtype: AbstractIOReference z!_AsyncStreamConnector.start(); %sz9_AsyncStreamConnector.start() expected _STATE_NOT_STARTED) r2rBrErIrHrOrDrPrQr5r:r r r r%sz_AsyncStreamConnector.startcCsH|j|jkr|j|_td|j|jdddStd|j|jdS)rRz%User canceled streaming linkup for %sTr3zD_AsyncStreamConnector cancel requested when not ACTIVE: state=%s; %sF)rIrOrSr2rBrErNr:r r r r6s  z_AsyncStreamConnector.cancelc Cstd||jt|ttfsJd||jf|j|jks$Jd|jf|j|_z$z| |Wnt y@t d|j |wW|j t|tddS|j t|tdw)aAdvance 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 z0_AsyncStreamConnector._report_completion(%r); %szQ_AsyncStreamConnector._report_completion() expected exception or tuple as result.zA_AsyncStreamConnector._report_completion() expected _STATE_ACTIVEz%r: _on_done(%r) failed.rqN)r2rBrEr rTtuplerIrOrVrGr1r^rYrNrWr r r rYs,  *z(_AsyncStreamConnector._report_completionc Cstd|j|j|jkrtd|j|jdS|jdur#|dStd|jz|jj|jddd|jd|_Wn t yZ}zt d|j|| |WYd}~dSd}~ww| dS)zCalled 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 z(_AsyncStreamConnector._start_async(); %szMAbandoning streaming linkup due to inactive state transition; state=%s; %s; .NzStarting SSL handshake on %sF) server_sidedo_handshake_on_connectsuppress_ragged_eofsr/zSSL wrap_socket(%s) failed: %r) r2rBrErIrOrm_linkup wrap_socketrnr1r^rY_do_ssl_handshaker_r r r rQ s6       z"_AsyncStreamConnector._start_asyncc Csjtdd}zz|}Wnty#}z td||jd}~ww|jdurIz t|j||j}Wn5tyH}z td||jd}~wwz t |j||j}Wntyh}z td||jd}~wwtd|z| |Wnty}z td|||jd}~wwtd||Wnty}z|}WYd}~n d}~ww||f}| |dS) z}Connection is ready: instantiate and link up transport and protocol, and invoke user's completion callback. z_AsyncStreamConnector._linkup()Nz'protocol_factory() failed: error=%r; %sz%PlainTransport() failed: error=%r; %sz#SSLTransport() failed: error=%r; %sz_linkup(): created transport %rz1protocol.connection_made(%r) failed: error=%r; %sz2_linkup(): introduced transport to protocol %r; %r) r2rBrlr1r^rErm_AsyncPlaintextTransportrD_AsyncSSLTransportconnection_maderY)r& transportprotocolrrXr r r rv0sn     z_AsyncStreamConnector._linkupc Cstd|j|jkrtd|j|jdSd}zqz|jWn^tjy~}zQ|jtj krNtd|jd|_ |j |j |j|j |j n&|jtjkrstd|jd|_ |j |j |j|j |j nWYd}~nd}~wwd}td|jWn ty}ztd ||j||WYd}~dSd}~ww|rtd |j|j |j |j |j d|_ td |j|dSdS) zJPerform asynchronous SSL handshake on the already wrapped socket z)_AsyncStreamConnector._do_ssl_handshake()z`_do_ssl_handshake: Abandoning streaming linkup due to inactive state transition; state=%s; %s; .NFzSSL handshake wants read; %s.TzSSL handshake wants write. %sz(SSL handshake completed successfully: %sz%SSL do_handshake failed: error=%r; %sz8_do_ssl_handshake: removing watchers ahead of linkup: %sz=_do_ssl_handshake: pre-linkup removal of watchers is done; %s)r2rBrIrOrE do_handshakeriSSLErrorrSSL_ERROR_WANT_READrorD set_readerrMrxrLSSL_ERROR_WANT_WRITEr\rprcr1r^rYrv)r&donerr r r rxjsr      z'_AsyncStreamConnector._do_ssl_handshakeN)r(r)r*r+rHrOrSrVr9rhrNr%r6rYrQrvrxr r r r r0ds&0 "  % 9r0c@seZdZdZdZdZdZdZdZdZ Gdd d e Z d d Z d d Z ddZddZddZddZddZeeddZeeddZeddZeddZed d!Zed"d#Zd$S)%_AsyncTransportBasezIBase class for `_AsyncPlaintextTransport` and `_AsyncSSLTransport`. r;r<r=iics eZdZdZfddZZS)z_AsyncTransportBase.RxEndOfFilezNWe raise this internally when EOF (empty read) is detected on input. csttj|dddS)NzEnd of input stream (EOF))superr RxEndOfFiler9r: __class__r r r9s z(_AsyncTransportBase.RxEndOfFile.__init__)r(r)r*r+r9 __classcell__r r rr rsrcCs:td|||_||_||_|j|_t|_ d|_ dS)a~ :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: z _AsyncTransportBase.__init__: %srN) r2rBrE _protocolrDrOrI collectionsdeque _tx_buffers_tx_buffered_byte_countr&r!r}r r r r r9s   z_AsyncTransportBase.__init__cCs td|j|j|ddS)aClose 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 z+Aborting transport connection: state=%s; %sN)r2rcrIrE_initiate_abortr:r r r aborts z_AsyncTransportBase.abortcC|jS)zReturn the protocol linked to this transport. :rtype: pika.adapters.utils.nbio_interface.AbstractStreamProtocol )rr:r r r get_protocolz _AsyncTransportBase.get_protocolcCr)ze :returns: Current size of output data buffered by the transport :rtype: int )rr:r r r get_write_buffer_sizerz)_AsyncTransportBase.get_write_buffer_sizecCsh|std|j|jtd||j|jkr#td|j|jdS|j ||j t |7_ dS)Buffer the given data until it can be sent asynchronously. :param bytes data: :raises ValueError: if called with empty data z,write() called with empty data: state=%s; %sz#write() called with empty data {!r};Ignoring write() called during inactive state: state=%s; %sN) r2rrIrErCrrOrBrappendrlen)r&datar r r _buffer_tx_datas  z#_AsyncTransportBase._buffer_tx_datac Csd}|j|jkrS||jkrU||j|j}|t|7}|s(td|j| z|j |Wnt yE}z t d||jd}~ww|j|jkrW||jks dSdSdSdS)aUtility 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 rzSocket EOF; %sz-protocol.data_received() failed: error=%r; %sN)rIrO_MAX_CONSUME_BYTES_sigint_safe_recvrE_MAX_RECV_BYTESrr2rrr data_receivedr1r^)r&bytes_consumedrrr r r _consumes.    z_AsyncTransportBase._consumecCs|jrE||j|jd}|j}|t|kr+td|t||j||d|j|8_|jdks@Jd|j|j f|jsdSdS)aUtility 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 rz/Partial send, requeing remaining data; %s of %sNz7_AsyncTransportBase._produce() tx buffer size underflow) r_sigint_safe_sendrEpopleftrr2rB appendleftrrI)r&num_bytes_sentchunkr r r _produce(s     z_AsyncTransportBase._producecC ||S)amReceive 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 )recv)r! max_bytesr r r rA z%_AsyncTransportBase._sigint_safe_recvcCr)a@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 )send)r!rr r r rPrz%_AsyncTransportBase._sigint_safe_sendcCsT|j|jkr(td|j|j|j|j|j|j|j dSdS)z2Unregister the transport from I/O events z$Deactivating transport: state=%s; %sN) rIrOr2rcrErDrprMrLrclearr:r r r _deactivate_s  z_AsyncTransportBase._deactivatecCst|j|jkr8td|j|jz |jtjWn tj j y#Ynw|j d|_d|_ d|_ |j|_dSdS)z{Close the transport's socket and unlink the transport it from references to other assets (protocol, etc.) z4Closing transport socket and unlinking: state=%s; %sN)rIrVr2rcrEshutdownr? SHUT_RDWRrrrr3rrDr:r r r _close_and_finalizeks   z'_AsyncTransportBase._close_and_finalizecCstd|j||j|j|jksJd|jf|j|jkrdS||dur9|j|jkr4tddS|j|_n|j|jkrN|j|jksLJd|jfdS|j |_|j t |j|dS)aInitiate 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. zo_AsyncTransportBase._initate_abort(): Initiating abrupt asynchronous transport shutdown: state=%s; error=%r; %szB_AsyncTransportBase._initate_abort() expected non-_STATE_COMPLETEDNzM_AsyncTransportBase._initiate_abort(): ignoring - user-abort already pending.zD_AsyncTransportBase._initate_abort() expected _STATE_ABORTED_BY_USER)r2rcrIrErVr_STATE_ABORTED_BY_USERrBrO _STATE_FAILEDrDrPrpartial_connection_lost_notify_asyncr_r r r r~s4        z#_AsyncTransportBase._initiate_abortc Cstd|j||j|jkrdS|dur)|j|jkr)|j|jks'Jd|jfdSz&z|j|WntyH}z t d|||j d}~wwW| dS| w)aHandle 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. z1Concluding transport shutdown: state=%s; error=%rNzS_AsyncTransportBase._connection_lost_notify_async() expected _STATE_ABORTED_BY_USERz/protocol.connection_lost(%r) failed: exc=%r; %s) r2rBrIrVrrrconnection_lostr1r^rEr)r&rexcr r r rs,  z1_AsyncTransportBase._connection_lost_notify_asyncN)r(r)r*r+rOrrrVrrOSErrorrr9rrrrrr staticmethodrrrrhrrrrr r r r rs<  %      4rcs@eZdZdZfddZddZeddZedd ZZ S) ryz`Implementation of `nbio_interface.AbstractStreamTransport` for a plaintext connection. cs.tt|||||j|j|jdS)a{ :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)rryr9rDrrErM_on_socket_readablerrr r r9s z!_AsyncPlaintextTransport.__init__cCs||j|jkrtd|j|jdS|sJd||jf|dk}|||r<|j|j |j td|jdSdS)rrNz7_AsyncPlaintextTransport.write(): empty data from user.r!Turned on writability watcher: %s) rIrOr2rBrErrrDr\rM_on_socket_writabler&rtx_buffer_was_emptyr r r writes   z_AsyncPlaintextTransport.writec Cs|j|jkrtd|j|jdSz|Wn|jylz|j}Wn!t yG}zt d||j| |WYd}~YdSd}~ww|r]t d|j|j |jYdSt d|j| dYdSt tjjfy}z8t|tjjr|jtvrtd|jnt d||jdtjt| |WYd}~dSWYd}~dSd}~ww|j|jkrtd |j|jdSdS) zIngest 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. EIgnoring readability notification due to inactive state: state=%s; %sNz,protocol.eof_received() failed: error=%r; %sz0protocol.eof_received() elected to keep open: %sz,protocol.eof_received() elected to close: %szRecv would block on %sa_AsyncBaseTransport._consume() failed, aborting connection: error=%r; sock=%s; Caller's stack: %sz>Leaving Plaintext consumer due to inactive state: state=%s; %s)rIrOr2rBrErrr eof_receivedr1r^rrcrDrprMrrrr r_TRY_IO_AGAIN_SOCK_ERROR_CODESjoin tracebackformat_exceptionsysexc_info)r& keep_openrr r r r sb      z,_AsyncPlaintextTransport._on_socket_readablec Cs|j|jkrtd|j|jdS|jsJd|jfz|WnHttj j fyi}z8t |tj j rA|j t vrAtd|jntd||jdtjt||WYd}~dSWYd}~dSd}~ww|js|j|jtd|jdSdS)-Handle writable socket notification EIgnoring writability notification due to inactive state: state=%s; %sNzP_AsyncPlaintextTransport._on_socket_writable() called, but _tx_buffers is empty.zSend would block on %sa_AsyncBaseTransport._produce() failed, aborting connection: error=%r; sock=%s; Caller's stack: %srz"Turned off writability watcher: %s)rIrOr2rBrErrr1rrrr rrr^rrrrrrrDrLrMr_r r r r=s<      z,_AsyncPlaintextTransport._on_socket_writable) r(r)r*r+r9rrhrrrr r rr rys  3rycs`eZdZdZfddZddZeddZedd Zefd d Z efd d Z Z S)rzz\Implementation of `.nbio_interface.AbstractStreamTransport` for an SSL connection. csJtt|||||j|_d|_|j|j |j |j |j dS)a{ :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) rrzr9r_ssl_readable_action_ssl_writable_actionrDrrErMrrPrrr r r9gs  z_AsyncSSLTransport.__init__cCs|j|jkrtd|j|jdS|sJd||jf|dk}|||rE|jdurG|j|_|j |j |j td|jdSdSdS)rrNz1_AsyncSSLTransport.write(): empty data from user.rr) rIrOr2rBrErrrrrDr\rMrrr r r r{s   z_AsyncSSLTransport.writec C~|j|jkrtd|j|jdS|jr4z|WdSty3}z ||WYd}~dSd}~wwtd|j|jdS)z+Handle readable socket indication rNz>SSL readable action was suppressed: ssl_writable_action=%r; %s) rIrOr2rBrErr1rrr_r r r r$ z&_AsyncSSLTransport._on_socket_readablec Cr)rrNz>SSL writable action was suppressed: ssl_readable_action=%r; %s) rIrOr2rBrErr1rrr_r r r rrz&_AsyncSSLTransport._on_socket_writablec sd}z tt|WnCtjyN}z6|jtjkr"td|j n"|jtj kr2td|j d}nt d||j d t jtWYd}~nd}~ww|j|jkr`td|j|j dS|j|j|r|jsw|j|j |j|j|_|j|jkr|j|j d|_n!|js|j|j |j|j|_|jr|j|j d|_|jr|js|j|_|j|j |jdSdSdS) a[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 TzSSL ingester wants read: %szSSL ingester wants write: %sFrrNz8Leaving SSL consumer due to inactive state: state=%s; %s) rrzrrirrrr2rBrErr^rrrrrrIrOrDrPrrrrMrrLr\rrprr)r&next_consume_on_readablerrr r rs^        z_AsyncSSLTransport._consumec sBd}z tt|WnEtjyP}z8|jtjkr$td|j d}n"|jtj kr4td|j d}nt d||j d t jtWYd}~nd}~ww|jr]Jdt|jf|jr|duskJd |jf|r|js{|j|j |j|j|_|j|jkr|j|j d|_nb|js|j|j |j|j|_|jr|j|j d|_n@|j|jkr|j|j d|_|j|jksJd |jfn |j|jksJd d |jd |jd|jfd|_|j|j |js|j|_|j|j |j|j|jdS|j r|j|jdSdS)aI[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 NzSSL emitter wants read: %sFzSSL emitter wants write: %sTrrz__AsyncSSLTransport._produce(): no exception from parent class, but data remains in _tx_buffers.zE_AsyncSSLTransport._produce(): next_produce_on_writable is still Nonezr_AsyncSSLTransport._produce(): with empty tx_buffers, writable_action cannot be _produce when readable is _producez_AsyncSSLTransport._produce(): with empty tx_buffers, expected writable_action as _produce when readable_action is not _producezwritable_action:zreadable_action:zstate:)!rrzrrirrrr2rBrErr^rrrrrrrrIrrDr\rMrrrprrrLrrPpending)r&next_produce_on_writablerrr r r s          z_AsyncSSLTransport._produce) r(r)r*r+r9rrhrrrrrr r rr rzas   Hrz)(r+rrrloggingr rdr?rirr"pika.adapters.utils.nbio_interfacerr pika.compatrpika.diagnostic_utilsEAGAIN EWOULDBLOCKr EINPROGRESSr[ getLoggerr(r2diagnostic_utilscreate_log_exception_decoratorrhr rrobjectrr,r5r$r0rryrzr r r r sR     ): F1