)JfFdZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl m Z m Z ddlZddlZejejfZejejfZejeZejeZdZdZdZGddeZ Gd d eZ!Gd d e Z"Gd deZ#GddeZ$Gdde Z%Gdde%Z&Gdde%Z'dS)z^Utilities for implementing `nbio_interface.AbstractIOServices` for pika connection adapters. N)AbstractIOReferenceAbstractStreamTransportcjt|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)callbacknames h/home/alex/cs2snipeproduction/venv/lib/python3.11/site-packages/pika/adapters/utils/io_services_utils.pycheck_callback_argr ,sH H  ;BB ( c~t|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)fds r check_fd_argr9sL b'* + +K > E Eb I IKK KKKr cFtjfd}|S)z0Function decorator for retrying on SIGINT. c |i|S#tjj$r }|jtjkrYd}~8d}~wwxYw)zWrapper for decorated functionTN)pikacompat SOCKET_ERRORerrnoEINTR)argskwargserrorfuncs r retry_sigint_wrapz+_retry_on_sigint..retry_sigint_wrapJsc  tT,V,,,;+   ;%+--HHHH  s A;;A) functoolswraps)rrs` r _retry_on_sigintr"Es; _T      r ceZdZdZdZdS)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`. cLt||||S)z[Implement :py:meth:`.nbio_interface.AbstractIOServices.connect_socket()`. )nbiosock resolved_addron_done)_AsyncSocketConnectorstart)selfr'r(r)s r connect_socketz$SocketConnectionMixin.connect_socketbs. %D "UWW %r N)__name__ __module__ __qualname____doc__r-r r r$r$Ys-%%%%%r r$ceZdZdZ ddZdS)StreamingConnectionMixinzImplements `.nbio_interface.AbstractIOServices.create_streaming_connection()` on top of `.nbio_interface.AbstractFileDescriptorServices` and basic `nbio_interface.AbstractIOServices` services. NcB t||||||S#t$rk}td|| |n3#t$r&}td||Yd}~nd}~wwxYwd}~wwxYw)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,r6r'r)r7r8rs r create_streaming_connectionz4StreamingConnectionMixin.create_streaming_connectionts (!1' / !!! "'  )    MMF ! ! ! D  D D D 5tUCCCCCCCC  D  s>&) BBA%$B% B/B BBBB)NN)r.r/r0r1r=r2r r r4r4ls:1548 r r4ceZdZdZdZdZdS)_AsyncServiceAsyncHandlezGThis module's adaptation of `.nbio_interface.AbstractIOReference` c|j|_dS)zZ :param subject: subject of the reference containing a `cancel()` method N)cancel_cancel)r,subjects r __init__z!_AsyncServiceAsyncHandle.__init__s ~ r c*|S)zCancel pending operation :returns: False if was already done or cancelled; True otherwise :rtype: bool )rBr,s r rAz_AsyncServiceAsyncHandle.cancels||~~r N)r.r/r0r1rDrAr2r r r?r?s<&&&r r?ceZdZdZdZdZdZdZdZe dZ dZ d Z e d Z e d Ze d Zd S)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()`. rct|d tj|j|dn#t$rz}t tdst dn@d|||}t |t|Yd}~nd}~wwxYw||_ ||_ ||_ ||_|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 socketrLfamilyr:hasattrr;debugrr ValueError_nbio_sock_addr_on_done_STATE_NOT_STARTED_state_watching_socket_events)r,r&r'r(r)rmsgs r rDz_AsyncSocketConnector.__init__s 7I... &  T[-*: ; ; ; ; & & &6;// & NPPPP2396($4747 c""" oo%PPPPP &  "  - ',$$$s 3 B7A0B22B7c|jr:d|_|j|jdSdS)z'Remove socket watcher, if any FN)rXrR remove_writerrSfilenorFs r _cleanupz_AsyncSocketConnector._cleanupsK  ' :+0D ( J $ $TZ%6%6%8%8 9 9 9 9 9 : :r c|j|jksJd|jf|j|_|j|jt |S)zZStart asynchronous connection establishment. :rtype: AbstractIOReference z:_AsyncSocketConnector.start(): expected _STATE_NOT_STARTED)rWrV _STATE_ACTIVErRadd_callback_threadsafe _start_asyncr?rFs r r+z_AsyncSocketConnector.startsb {d5555 H K8555(  **4+<==='---r c|j|jkrH|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)rWr__STATE_CANCELEDr;rPrSrTr]rFs r rAz_AsyncSocketConnector.cancels} ;$, , ,.DK MMI*dj 2 2 2 MMOOO4   K 5 5 5ur cTtd||jt|tt dfs Jd|f|j|jksJd|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) r;rPrSr BaseExceptiontyperWr__STATE_COMPLETEDr]rUr,results r _report_completionz(_AsyncSocketConnector._report_completion s  Hdj * * *&=$t**"=>> ' ' %A' ' ' '{d0000 !"&+3/000+   fr cN|j|jkr.td|j|j|jdS |j|jn#ttj j f$rz}t|tj j r|j tvrnBtd|j|j|||Yd}~dSYd}~nd}~wwxYw |j|j|jd|_td|jdS#t$rA}td|j|||Yd}~dSd}~wwxYw)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: %rTz/Connection-establishment is in progress for %s.zasync.set_writer(%s) failed: %r)rWr_r;rPrSrTconnectr:rrrrr(_CONNECTION_IN_PROGRESS_SOCK_ERROR_CODESrrkrR set_writerr\ _on_writablerX exceptionr,rs r raz"_AsyncSocketConnector._start_async s ;$, , , MM+,0J DK Q Q Q F  J  tz * * * *4;34   5$+":;; K#KKK 94:"j%111''...   & J ! !$*"3"3"5"5t7H I I I,0D ( MMK* & & & & &      ?# % % %  # #E * * * FFFFF  s1A C5;A*C00C597E F$#6FF$c|j|jkr(td|j|jdS|jt jt j}|s#t d|jd}nVtj |}td|j||tj ||}||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))rWr_r;rrS getsockoptrM SOL_SOCKETSO_ERRORinfoosstrerrorrrrrk)r, error_coderj error_msgs r rpz"_AsyncSocketConnector._on_writableGs ;$, , , MMDEIZ     FZ**6+344 4  8        8 8 8##)6%#6#688 8 8  !1 ' / - %sB C#CCctd||jrtd||jd|_|j|j|j|j |rptd||j |jn4#t$r'}t d||jd}~wwxYwd|_d|_d|_ d|_ d|_ d|_dS#d|_d|_d|_ d|_ d|_ d|_wxYw)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)r;rPrrSrR remove_readerr\r[r<r:rqrrrrU)r,r<rs r r]z_AsyncStreamConnector._cleanups  :EBBB   : MMG    %*D ! J $ $TZ%6%6%8%8 9 9 9 J $ $TZ%6%6%8%8 9 9 9 !  L4:'''J$$&&&& %%&J&+TZ999 DJDJ%)D " $D $(D ! DMMM DJDJ%)D " $D $(D ! DM s0.#E C,+E , D6"DDE ,E8ctd|j|j|jksJd|jf|j|_|j|jt|S)zCKick off the workflow :rtype: AbstractIOReference z!_AsyncStreamConnector.start(); %sz9_AsyncStreamConnector.start() expected _STATE_NOT_STARTED) r;rPrSrWrVr_rRr`rar?rFs r r+z_AsyncStreamConnector.starts{  94:FFF{d5555 !"&+8/555(  **4+<==='---r c|j|jkrD|j|_td|j|ddStd|j|jdS)rcz%User canceled streaming linkup for %sTr<zD_AsyncStreamConnector cancel requested when not ACTIVE: state=%s; %sF)rWr_rdr;rPrSr]rFs r rAz_AsyncStreamConnector.cancelsy ;$, , ,.DK MMA4: N N N MMM % % %4   K 5 5 5ur c<td||jt|tt fsJd||jf|j|jksJd|jf|j|_ | |n0#t$r#t d|j |wxYw | t|tdS#| t|twxYw)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.rN)r;rPrSrrftuplerWr_rhrUr:rqrkr]ris r rkz(_AsyncStreamConnector._report_completionsF  Hdj * * *&=%"899 5 5  & <5 5 5 5{d0000 ![3*000+  C MM& ! ! ! !      8"5v ? ? ?   " MM 6= A AM B B B B BDMM 6= A AM B B B Bs>BC0-CC00+DcTtd|j|j|jkr(td|j|jdS|j|dStd|j |j|jddd|j|_nN#t$rA}t d|j|| |Yd}~dSd}~wwxYw| 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_eofsr8zSSL wrap_socket(%s) failed: %r) r;rPrSrWr_r_linkup wrap_socketrr:rqrk_do_ssl_handshakerrs r raz"_AsyncStreamConnector._start_async sD  @$*MMM ;$, , , MM./3{DJ H H H F   $ LLNNNNN MM8$* E E E !.::J %,1).$($9 ;;;    !!"BDJ"')))''...    " " $ $ $ $ $s.C D6D  Dctdd} |}n4#t$r'}td||jd}~wwxYw|jQ t|j||j}n#t$r'}td||jd}~wwxYw t|j||j}n4#t$r'}td||jd}~wwxYwtd| | |n5#t$r(}td|||jd}~wwxYwtd||||f}n#t$r }|}Yd}~nd}~wwxYw| |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) r;rPrr:rqrSr_AsyncPlaintextTransportrR_AsyncSSLTransportconnection_maderk)r, transportprotocolrrjs r rz_AsyncStreamConnector._linkup0sG  7888 , + 1133   !!"K"'555   ( 8 Hdj!:!:II %%&M&+TZ999  24:x37:!?!?II %%&K&+TZ999 MM;Y G G G ((3333   !!Gudj222   MMN#X / / /  *FF   FFFFFF  '''''s4F A%"A  A%% F0B F B="B88B==FCF D'"D  DF-EF E5 #E00E55F F/#F**F/ctd|j|jkr(td|j|jdSd} |jd}td|jnh#tj$rU}|j tj krtd|jd|_ |j |j|j|j |jn|j tjkrtd|jd|_ |j |j|j|j |jnYd}~nd}~wwxYwnN#t($rA}td ||j||Yd}~dSd}~wwxYw|rtd |j|j |j|j |jd|_ 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; .NFTz(SSL handshake completed successfully: %szSSL handshake wants read; %s.zSSL handshake wants write. %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)r;rPrWr_rS do_handshakerwrSSLErrorrSSL_ERROR_WANT_READrrR set_readerr\rr[SSL_ERROR_WANT_WRITErorr:rqrkr)r,doners r rz'_AsyncStreamConnector._do_ssl_handshakejs  ABBB ;$, , , MM@AE     F  ) '')))$ G!Z))))%<   ;#"999MM"A4:NNN,0D)J))$**;*;*=*=*.*@BBBJ,,TZ->->-@-@AAAA[C$<<<MM"A4:NNN,0D)J))$**;*;*=*=*.*@BBBJ,,TZ->->-@-@AAAA (      Eu"j * * *  # #E * * * FFFFF     MMJ     J $ $TZ%6%6%8%8 9 9 9 J $ $TZ%6%6%8%8 9 9 9%*D ! MMO     LLNNNNN  s=B0$G=G9$E G4/G=4G99G== I6IIN)r.r/r0r1rVr_rdrhrDr|r]r+rArkrarrr2r r r9r9ds MO.&.&.&` ! !_ !D...&(CC_C>#%#%_#%J7(7(_7(r::_:::r r9ceZdZdZdZdZdZdZdZdZ Gdd e Z d Z d Z d Zd ZdZdZdZeedZeedZedZedZedZedZdS)_AsyncTransportBasezIBase class for `_AsyncPlaintextTransport` and `_AsyncSSLTransport`. rHrIrJiic"eZdZdZfdZxZS)_AsyncTransportBase.RxEndOfFilezNWe raise this internally when EOF (empty read) is detected on input. cdttj|dddS)NzEnd of input stream (EOF))superr RxEndOfFilerD)r, __class__s r rDz(_AsyncTransportBase.RxEndOfFile.__init__s9 %14 8 8 A A/ 1 1 1 1 1r )r.r/r0r1rD __classcell__rs@r rrsB   1 1 1 1 1 1 1 1 1r rctd|||_||_||_|j|_tj|_ 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) r;rPrS _protocolrRr_rW collectionsdeque _tx_buffers_tx_buffered_byte_count)r,r'rr&s r rDz_AsyncTransportBase.__init__sW  8$??? ! ( &,..'($$$r c|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)r;rwrWrS_initiate_abortrFs r abortz_AsyncTransportBase.abortsB  BDKZ ! ! ! T"""""r c|jS)zReturn the protocol linked to this transport. :rtype: pika.adapters.utils.nbio_interface.AbstractStreamProtocol )rrFs r get_protocolz _AsyncTransportBase.get_protocols ~r c|jS)ze :returns: Current size of output data buffered by the transport :rtype: int )rrFs r get_write_buffer_sizez)_AsyncTransportBase.get_write_buffer_sizes ++r cx|sHtd|j|jt d||j|jkr(td|j|jdS|j ||xj t|z c_ 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) r;rrWrSrQrr_rPrappendrlen)r,datas r _buffer_tx_dataz#_AsyncTransportBase._buffer_tx_datas Q MMH+tz 3 3 3BII$OOPP P ;$, , , MM $ TZ 9 9 9 F %%% $$D 1$$$$r cd}|j|jkr||jkr||j|j}|t |z }|s4td|j|  |j |n4#t$r'}t d||jd}~wwxYw|j|jkr||jkdSdSdSdS)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)rWr__MAX_CONSUME_BYTES_sigint_safe_recvrS_MAX_RECV_BYTESrr;rrr data_receivedr:rq)r,bytes_consumedrrs r _consumez_AsyncTransportBase._consumes$ {d000 777))$*d6JKKD c$ii 'N ) . ;;;&&((( ,,T2222   !!CUJ     {d000 77777107710sB"" C,"CCc|jr||j|jd}|j}|t |krKt d|t ||j||d|xj|zc_|jdksJd|j|j f|jdSdS)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_sendrSpopleftrr;rP appendleftrrW)r,num_bytes_sentchunks r _producez_AsyncTransportBase._produce(s ;!33DJ484DQ4GIIN$,,..EE ** O,c%jj::: ++E.//,BCCC  ( (N : ( (/1444I,dk7;444 ; ; ; ; ;r c,||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_bytess r rz%_AsyncTransportBase._sigint_safe_recvAsyy###r c,||S)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'rs r rz%_AsyncTransportBase._sigint_safe_sendPsyyr cl|j|jkrtd|j|j|j|j|j|j|j dSdS)z2Unregister the transport from I/O events z$Deactivating transport: state=%s; %sN) rWr_r;rwrSrRrr\r[rclearrFs r _deactivatez_AsyncTransportBase._deactivate_s ;$, , , LL? % % % J $ $TZ%6%6%8%8 9 9 9 J $ $TZ%6%6%8%8 9 9 9   " " $ $ $ $ $ - ,r cj|j|jkrtd|j|j |jt jn#tj j $rYnwxYw|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)rWrhr;rwrSshutdownrM SHUT_RDWRrrrr<rrRrFs r _close_and_finalizez'_AsyncTransportBase._close_and_finalizeks ;$/ / / LLOdj 2 2 2  ##F$45555;+     J     DJ!DNDJ/DKKK 0 /s$AA43A4c:td|j||j|j|jksJd|jf|j|jkrdS||9|j|jkrtddS|j|_n?|j|jkr#|j|jksJd|jfdS|j |_|j tj |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)r;rwrWrSrhr_STATE_ABORTED_BY_USERrPr_ _STATE_FAILEDrRr`r partial_connection_lost_notify_asyncrrs r rz#_AsyncTransportBase._initiate_abort~sY   F K  , , , {d3333 #$(K61333 ;$/ / / F  ={d999 GHHH5DKK{d000{d&AAAA-.2kD;AAA,DK **  d@% H H J J J J Jr ctd|j||j|jkrdS|3|j|jkr#|j|jksJd|jfdS |j|n5#t$r(}t d|||j d}~wwxYw | dS#| wxYw)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) r;rPrWrhrrrconnection_lostr:rqrSr)r,rexcs r rz1_AsyncTransportBase._connection_lost_notify_asyncs!  Ik5 * * * ;$/ / / F  0B!B!B;$"====237;@@=== F ' N * *5 1 1 1 1      O#S$* 6 6 6    2  $ $ & & & & &D $ $ & & & &s**BC B7#B22B77CC'N)r.r/r0r1r_rrrhrrOSErrorrrDrrrrrr staticmethodr"rrr|rrrrr2r r rrsMMO$11111g111)))& # # #,,,222*###J;;;2 $ $\ $  \  % %_ %00_0$2J2J_2Jh ' '_ ' ' 'r rcTeZdZdZfdZdZedZedZxZ S)rz`Implementation of `nbio_interface.AbstractStreamTransport` for a plaintext connection. ctt|||||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)rrrDrRrrSr\_on_socket_readabler,r'rr&rs r rDz!_AsyncPlaintextTransport.__init__sY &--66tXtLLL dj//1143KLLLLLr c|j|jkr(td|j|jdS|sJd||jf|dk}|||rY|j|j |j td|jdSdS)rrNz7_AsyncPlaintextTransport.write(): empty data from user.r!Turned on writability watcher: %s) rWr_r;rPrSrrrRror\_on_socket_writabler,rtx_buffer_was_emptys r writez_AsyncPlaintextTransport.writes ;$, , , MM $ TZ 9 9 9 F ) )ODK) ) ) )#88::a? T"""  K J ! !$*"3"3"5"5t7O P P P MM=tz J J J J J K Kr c |j|jkr(td|j|jdS ||j|jkr(td|j|jdSdS#|j$r |j}|rTt d|j|j |j YdSt d|j| dYdS#t$rB}td||j| |Yd}~YdSd}~wwxYwtt jjf$r}t'|t jjr/|jt*vr!td|jnmtd||jd t/jt3j| |Yd}~dSYd}~dSd}~wwxYw) 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>Leaving Plaintext consumer due to inactive state: state=%s; %sz0protocol.eof_received() elected to keep open: %sz,protocol.eof_received() elected to close: %sz,protocol.eof_received() failed: error=%r; %szRecv would block on %sa_AsyncBaseTransport._consume() failed, aborting connection: error=%r; sock=%s; Caller's stack: %s)rWr_r;rPrSrrr eof_receivedrwrRrr\rr:rqrrrrr_TRY_IO_AGAIN_SOCK_ERROR_CODESjoin tracebackformat_exceptionsysexc_info)r, keep_openrs r rz,_AsyncPlaintextTransport._on_socket_readable s ;$, , , MM&'+{DJ @ @ @ F% D MMOOO>{d000 *+/; DDDDD10= / / / / N7799 /LLJ $$$J,,TZ->->-@-@AAAAAALL!O!%---((...... , , ,!!BEJ   $$U++++++++++  ,4;34 , , ,5$+":;; ,K#AAA 6 CCCC!!J4:rww!2CLNNC(E(EFFF $$U+++++++++DCCCCC ,sIB I)D:,AI)5I): F6F:I)FI)#B5I$$I)c L|j|jkr(td|j|jdS|jsJd|jf ||jsS|j|j td|jdSdS#ttj j f$r}t|tj j r/|jt vr!td|jnmtd||jdt'jt+j||Yd}~dSYd}~dSd}~wwxYw)-Handle writable socket notification EIgnoring writability notification due to inactive state: state=%s; %sNzP_AsyncPlaintextTransport._on_socket_writable() called, but _tx_buffers is empty.z"Turned off writability watcher: %szSend would block on %sa_AsyncBaseTransport._produce() failed, aborting connection: error=%r; sock=%s; Caller's stack: %sr)rWr_r;rPrSrrrRr[r\r:rrrrrrrqrrrrrrrrs r rz,_AsyncPlaintextTransport._on_socket_writable=s ;$, , , MM&'+{DJ @ @ @ F 6 6 ()-"6 6 6 6 P MMOOO# P (():):)<)<=== BDJOOOOO P P4;34 , , ,5$+":;; ,K#AAA 6 CCCC!!J4:rww!2CLNNC(E(EFFF $$U+++++++++DCCCCC ,sCF#B5FF#) r.r/r0r1rDrr|rrrrs@r rrs MMMMM KKK81D1D_1Df P P_ P P P P Pr rceZdZdZfdZdZedZedZefdZ efdZ xZ S)rz\Implementation of `.nbio_interface.AbstractStreamTransport` for an SSL connection. c.tt|||||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) rrrDr_ssl_readable_action_ssl_writable_actionrRrrSr\rr`rs r rDz_AsyncSSLTransport.__init__gs  $''00xFFF$(M!$(! dj//1143KLLL **4+CDDDDDr c|j|jkr(td|j|jdS|sJd||jf|dk}|||rl|jg|j|_|j |j |j td|jdSdSdS)rrNz1_AsyncSSLTransport.write(): empty data from user.rr) rWr_r;rPrSrrrrrRror\rrs r rz_AsyncSSLTransport.write{s  ;$, , , MM $ TZ 9 9 9 F ) )IDK) ) ) )#88::a? T"""  K4#<#D(, D % J ! !$*"3"3"5"5t7O P P P MM=tz J J J J J K K#D#Dr cX|j|jkr(td|j|jdS|jrD |dS#t $r }||Yd}~dSd}~wwxYwtd|j|jdS)z+Handle readable socket indication rNz>SSL readable action was suppressed: ssl_writable_action=%r; %s) rWr_r;rPrSrr:rrrrs r rz&_AsyncSSLTransport._on_socket_readable ;$, , , MM&'+{DJ @ @ @ F  $  ,))+++++ , , ,$$U+++++++++ , MM-.2.G      A B!A<<BcX|j|jkr(td|j|jdS|jrD |dS#t $r }||Yd}~dSd}~wwxYwtd|j|jdS)rrNz>SSL writable action was suppressed: ssl_readable_action=%r; %s) rWr_r;rPrSrr:rrrrs r rz&_AsyncSSLTransport._on_socket_writablerrc d} tt||j|jkr(t d|j|jdS|j |j n#tj $r}|j tjkr!t d|jn|j tjkr#t d|jd}nSt d||jdt%jt)jYd}~nd}~wwxYw|r|js7|j|j|j |j|_|j|jkr8|j|jd|_n|js7|j|j|j|j|_|jr8|j|jd|_|jrL|jsG|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 Tz8Leaving SSL consumer due to inactive state: state=%s; %sNzSSL ingester wants read: %szSSL ingester wants write: %sFrr) rrrrWr_r;rPrSrRr`rrrrrrrqrrrrrrrr\rr[rorrrr)r,next_consume_on_readablerrs r rz_AsyncSSLTransport._consumes$(  I $d + + 4 4 6 6 6 {d000 *+/; DDD J . .t/G H H H H9|   {c555 ;TZHHHH 888  $ 1, @ %%dj&7&7&9&9&*&>@@@(, D %(DM99 (():):)<)<===,0), @ %%dj&7&7&9&9&*&>@@@(, D %( 1 (():):)<)<===,0)   QD$= Q(, D % J ! !$*"3"3"5"5t7O P P P P P Q Q Q Qs'BECEEc d} tt||jrJdt |jfn#t j$r}|jt jkr#t d|j d}n|jt j kr#t d|j d}nSt d||j dtjt#jYd}~nd}~wwxYw|jr3|Jd |jf|r|js7|j|j |j|j|_|j|jkr8|j|j d|_n\|js7|j|j |j|j|_|jr8|j|j d|_n|j|jkrZ|j|j d|_|j|jksJd |jfnh|j|jks Jd d |jd |jd|jfd|_|j|j |jsd|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 Nz__AsyncSSLTransport._produce(): no exception from parent class, but data remains in _tx_buffers.zSSL emitter wants read: %sFzSSL emitter wants write: %sTrrzE_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:)!rrrrrrrrrr;rPrSrrqrrrrrrWrrRror\rrrrrr[rr`pending)r,next_produce_on_writablerrs r rz_AsyncSSLTransport._produce s $(  ' $d + + 4 4 6 6 6$' ' ':;>$<&<&*' ' ' ' '#|   {c555 :DJGGG+0(( 888 ;TZHHH+/((!!J4:rww!2CLNNC(E(EFFF  .  + >+77"k:+777( 50DJ))$**;*;*=*=*.*BDDD,0M), ==J,,TZ->->-@-@AAA04D-0DJ))$**;*;*=*=*.*BDDD,0M),5J,,TZ->->-@-@AAA04D-(DM99 (():):)<)<===,0)0DMAAA $ D-AAAA 0DMAAA&'9-/A-x DFAAA -1) (():):)<)<===( I(, D % J ! !$*"3"3"5"5t7O P P P J . .t/G H H H H H Z   ! ! I J . .t/G H H H H H I Is'AD.!CD))D.) r.r/r0r1rDrr|rrrrrrs@r rras EEEEE(KKK:_*_*FQFQFQFQ_FQPZIZIZIZI_ZIZIZIZIZIr r)(r1rrr loggingrrxrMrrr"pika.adapters.utils.nbio_interfacerr pika.compatrpika.diagnostic_utilsEAGAIN EWOULDBLOCKr EINPROGRESSrn getLoggerr.r;diagnostic_utilscreate_log_exception_decoratorr|r rr"objectr$r4r?r*r9rrrr2r r rs   IIIIIIII  L "  ,( ' H % %'FFwOO    K K K(%%%%%F%%%&&&&&&v&&&R2,v(v(v(v(v(Fv(v(v(rAAAAAFAAAH l'l'l'l'l'l'l'l'^ GPGPGPGPGP2GPGPGPTGIGIGIGIGI,GIGIGIGIGIr