>iFdZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl m Z m Z ddlZddlZej"ej$fZej(ej$fZej,eZej2j5eZdZdZdZGddeZ Gd d eZ!Gd d e Z"Gd deZ#GddeZ$Gdde Z%Gdde%Z&Gdde%Z'y)z^Utilities for implementing `nbio_interface.AbstractIOServices` for pika connection adapters. N)AbstractIOReferenceAbstractStreamTransportcPt|stdj||y)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 _/var/www/html/aiagents/fe/lib/python3.12/site-packages/pika/adapters/utils/io_services_utils.pycheck_callback_argr ,s1 H ;BB (  clt|tjstdj |y)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_argr9s6 b'** + > E Eb IK K ,r cBtjfd}|S)z0Function decorator for retrying on SIGINT. c |i|S#tjj$r(}|jtjk(rYd}~Id}~wwxYw)zWrapper for decorated functionN)pikacompat SOCKET_ERRORerrnoEINTR)argskwargserrorfuncs r retry_sigint_wrapz+_retry_on_sigint..retry_sigint_wrapJsM T,V,,;;++ ;;%++-  s AA  A  A) functoolswraps)rrs` r _retry_on_sigintr"Es( __T   r ceZdZdZdZy)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`. c<t||||jS)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 "UW %r N)__name__ __module__ __qualname____doc__r-r r r$r$Ys %r r$ceZdZdZ ddZy)StreamingConnectionMixinzImplements `.nbio_interface.AbstractIOServices.create_streaming_connection()` on top of `.nbio_interface.AbstractFileDescriptorServices` and basic `nbio_interface.AbstractIOServices` services. Nc t||||||jS#t$r[}tj d|| |j #t$r!}tj d||Yd}~d}~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 5tUCC   D s>! BBAB A=A83B8A==BB)NN)r.r/r0r1r=r2r r r4r4ls1548 r r4ceZdZdZdZdZy)_AsyncServiceAsyncHandlezGThis module's adaptation of `.nbio_interface.AbstractIOReference` c&|j|_y)zZ :param subject: subject of the reference containing a `cancel()` method N)cancel_cancel)r,subjects r __init__z!_AsyncServiceAsyncHandle.__init__s ~~ r c"|jS)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?creZdZdZdZdZdZdZdZe dZ dZ d Z e d Z e d Ze d Zy )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|d||_ ||_ ||_ ||_|j|_d|_y#t$rc}t tdst jdn3dj|||}t j|t|Yd}~d}~wwxYw)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;/ NP2396($47 c" o%P &s#A&& C/AC  Cc|jr;d|_|jj|jj yy)z'Remove socket watcher, if any FN)rXrR remove_writerrSfilenorFs r _cleanupz_AsyncSocketConnector._cleanups9  ' '+0D ( JJ $ $TZZ%6%6%8 9 (r c|j|jk(sJd|jf|j|_|jj |j t |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.startsf {{d555  H KK8 5((  **4+<+<='--r c&|j|jk(rM|j|_tj d|j |j |jytj d|j|j y)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.cancelsn ;;$,, ,..DK MMI**djj 2 MMO   KK 5r cJtjd||jt|tt dfs Jd|f|j |jk(sJd|j f|j|_|j|j|y)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  Hdjj *&=$t*"=> ' %A' '>{{d000 / !"&++3/ /0++   fr c|j|jk7r7tjd|j|j |jy |jj |j  |j j#|jj%|j&d|_tjd|jy#ttjjf$r}t|tjjr|jtvrnBtjd|j|j ||j|Yd}~yYd}~ d}~wwxYw#t$r<}tj+d|j||j|Yd}~yd}~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+,0JJ DKK Q   JJ  tzz * & JJ ! !$**"3"3"5t7H7H I,0D ( MMK** &+4;;334 5$++":":;KK#KK 94::"jj%1''.      ?# %  # #E *   s1%C8>F#FA4FF G 2GG c|j|jk7r,tjd|j|jy|jj t jt j}|s#tjd|jd}nWtj|}tjd|j||tjj||}|j|y)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 ;;$,, , MMDEIZZ   ZZ**6+<+34 4  8      !1 ' / -- % 8##)6%#68 8 8s(B;; C$CC$ctjd||jrtjd||jd|_|jj |jj |jj|jj  |r->-@A[[C$<$<<MM"A4::N,0D)JJ))$***;*;*=*.*@*@BJJ,,TZZ->->-@A (    Eu"jj *  # #E *   s=E9"K+K(0E-K#K+#K((K++ L042L++L0N)r.r/r0r1rVr_rdrhrDr|r]r+rArkrarrr2r r r9r9ds MO.&` ! !D.&(CC>#%#%J7(7(r::r r9ceZdZdZdZdZdZdZdZdZ Gdd e Z d Z d Z d Zd ZdZdZdZeedZeedZedZedZedZedZy)_AsyncTransportBasezIBase class for `_AsyncPlaintextTransport` and `_AsyncSSLTransport`. rHrIrJiic"eZdZdZfdZxZS)_AsyncTransportBase.RxEndOfFilezNWe raise this internally when EOF (empty read) is detected on input. cBttj|ddy)NzEnd of input stream (EOF))superr RxEndOfFilerD)r, __class__s r rDz(_AsyncTransportBase.RxEndOfFile.__init__s %114 A/ 1r )r.r/r0r1rD __classcell__rs@r rrs  1 1r rctjd|||_||_||_|j |_tj|_ d|_ y)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__sN  8$? ! (( &,,.'($r c|tjd|j|j|j dy)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.aborts0  BDKKZZ ! 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 c|sEtjd|j|jt dj ||j|j k7r,tjd|j|jy|jj||xjt|z c_ y)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 MMH++tzz 3BII$OP P ;;$,, , MM $ TZZ 9  % $$D 1$r cd}|j|jk(r||jkr|j|j|j }|t |z }|s0tjd|j|j |jj||j|jk(r||jkryyyy#t$r'}tjd||jd}~wwxYw)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 7 77))$**d6J6JKD c$i 'N . ;&&(( ,,T2{{d000 7 77171 !!CUJJ   sC D #"DD c|jr|j|j|jd}|jj}|t |kr>t j d|t ||jj||d|xj|zc_|jdk\sJd|j|jf|jryy)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!33DJJ484D4DQ4GIN$$,,.EE * O,c%j:  ++E./,BC  ( (N : (//14 ;I,,dkk7; ;4r c$|j|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$|j|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 c|j|jk(rtjd|j|j|j j |jj|j j|jj|jjyy)z2Unregister the transport from I/O events z$Deactivating transport: state=%s; %sN) rWr_r;rwrSrRrr\r[rclearrFs r _deactivatez_AsyncTransportBase._deactivate_s ;;$,, , LL? % JJ $ $TZZ%6%6%8 9 JJ $ $TZZ%6%6%8 9    " " $ -r c|j|jk7rtjd|j|j |jj t j|jjd|_d|_ d|_ |j|_yy#tjj$rYawxYw)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 ;;$// / LLOdjj 2  ##F$4$45 JJ   DJ!DNDJ//DK 0 ;;++  s)B11CCctjd|j||j|j|jk7sJd|jf|j|jk(ry|j |A|j|j k(rtjdy|j |_nW|j|jk7r-|j|j k(sJd|jfy|j|_|jjtj|j|y)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~s<   F KK  , {{d333 1 #$(KK61 13 ;;$// /   ={{d999 GH55DK{{d000{{d&A&AA;-.2kkD;;A,,DK **   d@@% H Jr ctjd|j||j|jk(ry|F|j|jk7r-|j|j k(sJd|jfy |j j| |jy#t$r(}tjd|||jd}~wwxYw#|jwxYw)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  Ikk5 * ;;$// /   0B0B!B;;$"="== @237;;@@ @=  ' NN * *5 1  $ $ &    O#S$** 6     $ $ &s$B22 C#;#CC##C&&C8N)r.r/r0r1r_rrrhrrOSErrorrrDrrrrrr staticmethodr"rrr|rrrrr2r r rrsMMO$1g1)& #,2*#J;2 $ $   % %00$2J2Jh ' 'r rcHeZdZdZfdZdZedZedZxZ S)rz`Implementation of `nbio_interface.AbstractStreamTransport` for a plaintext connection. ctt| ||||jj |j j |jy)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__sB &6tXtL djj//143K3KLr c|j|jk7r,tjd|j|jy|sJd||jf|j dk(}|j ||r_|jj|jj|jtjd|jyy)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 $ TZZ 9  )ODKK) )t#88:a? T"  JJ ! !$**"3"3"5t7O7O P MM=tzz J r c |j|jk7r,tjd|j|jy |j |j|jk7r,tjd|j|jyy#|j $r |jj}|rUtjd|j|jj|jjYytjd|j|jdYy#t$r=}tjd||j|j|Yd}~Yyd}~wwxYwtt j"j$f$r}t'|t j"j$r3|j(t*vr!tjd|jnitjd||jd j-t/j0t3j4|j|Yd}~yYd}~yd}~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&'+{{DJJ @ % D MMO>{{d000 *+/;; D1= / / NN779 LLJ $JJ,,TZZ->->-@ALL!O!%-((. ,!!BEJJ $$U++  ,4;;334 ,5$++":":;KK#AA 6 C!!J4::rww!22CLLNC(EF $$U++D ,sJBJ-EAJ1J F2F JF%JSSL readable action was suppressed: ssl_writable_action=%r; %s) rWr_r;rPrSrr:rrrrs r rz&_AsyncSSLTransport._on_socket_readable ;;$,, , MM&'+{{DJJ @   $ $ ,))+ MM-.2.G.G   ,$$U++ ,B B4B//B4cn|j|jk7r,tjd|j|jy|j r |j ytjd|j|jy#t $r}|j|Yd}~yd}~wwxYw)rrNz>SSL writable action was suppressed: ssl_readable_action=%r; %s) rWr_r;rPrSrr:rrrrs r rz&_AsyncSSLTransport._on_socket_writablerrc `d} tt| |j|jk7r,t j d|j|jy|jj|j|r|j,s>|jj/|jj1|j|j|_|j2|jk(r|jj5|jj1d|_n|j2s>|jj7|jj1|j8|j|_|j,r:|jj;|jj1d|_|j<r]|j2sP|j>|_|jj7|jj1|j8yyy#tj$r}|jtjk(r!t j d|jn|jtjk(r#t j d|jd}nTt j!d||jdj#t%j&t)j*Yd}~d}~wwxYw) 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 6 {{d000 *+/;; D JJ . .t/G/G H $,, %%djj&7&7&9&*&>&>@(, D %((DMM9 (():):)<=,0),, %%djj&7&7&9&*&>&>@(, D %(( (():):)<=,0)   D$=$=(, D % JJ ! !$**"3"3"5t7O7O P%> k|| {{c555 ;TZZH 8 88 |j*j-|jj/|j0|j|_|j2|jk(r|j*j5|jj/d|_n|j2s>|j*j7|jj/|j8|j|_|j(r9|j*j;|jj/d|_n|j2|jk(rf|j*j5|jj/d|_|j(|jk7sJd |j&f|j(|jk(s,Jd d |j(d |j2d|j&fd|_|j*j;|jj/|j2su|j<|_|j*j7|jj/|j8|j*j?|j8y|jjAr&|j*j?|j8yy#t j $r}|jt jk(r#tjd|jd}n|jt jk(r#tjd|jd}nTtjd||jdjtj t#j$Yd}~%d}~wwxYw)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 6$'' ':;>$$<&*' ''   +7 +"kk:+ +7(00JJ))$***;*;*=*.*B*BD,0MM),, =JJ,,TZZ->->-@A04D-00JJ))$***;*;*=*.*B*BD,0MM),,JJ,,TZZ->->-@A04D-((DMM9 (():):)<=,0)00DMMA- $ D--A 00DMMAF&'9--/A--x DFFA -1) (():):)<=(((, D % JJ ! !$**"3"3"5t7O7O P JJ . .t/G/G H ZZ   ! JJ . .t/G/G H"U|| {{c555 :DJJG+0( 8 88 ;TZZH+/(!!J4::rww!22CLLNC(EF  sM77Q) CQ$$Q)) r.r/r0r1rDrr|rrrrrrs@r rrasm E(K:**FQFQPZIZIr 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   I  LL "  ,( '  H %''FFwO  K(%F%&&v&R2,v(Fv(rAFAH l'l'^ GP2GPTGI,GIr