U lu f>Z@sdZddlZddlmZddlmZddlmZddlm Z m Z m Z m Z m Z mZmZmZmZmZmZmZmZddlZddlmZmZmZddlmZGd d d eZGd d d ZGd ddeejZ GdddZ!edddZ"Gdddeej#eZ$dS)z%Future-returning APIs for coroutines.N)Future)deque)chain) Any AwaitableCallableDictList NamedTupleOptionalTupleTypeTypeVarUnioncastoverload)EVENTSPOLLINPOLLOUT)Literalc@s6eZdZUeed<eed<eed<eed<eed<dS) _FutureEventfuturekindkwargsmsgtimerN)__name__ __module__ __qualname__r__annotations__strrrr!r!//tmp/pip-unpacked-wheel-h6ekxre8/zmq/_future.pyrs rc@sVeZdZUdZdZeed<eeed<edddZ eddd Z d ddd d Z dS) _AsynczMixin for common async logicN _current_loop_FuturereturncCsL|jdkr&||_||j|jS|}||jk rH||_|||S)zGet event loop Notice if event loop has changed, and register init_io_state on activation of a new event loop N)r$ _default_loop_init_io_state)selfZ current_loopr!r!r" _get_loop6s     z_Async._get_loopcCs tddS)Nz!Must be implemented in a subclassNotImplementedErrorr*r!r!r"r(Gsz_Async._default_loopcCsdSNr!r*loopr!r!r"r)Jsz_Async._init_io_state)N) rrr__doc__r$rrr rr+r(r)r!r!r!r"r#0s   r#cseZdZUdZeded<eed<eed<eeed<eeee ddd d Z eedd d d Z de ee eefdfdd ZZS) _AsyncPollerz:Poller that returns a Future on poll, instead of blocking. _AsyncSocket _socket_class_READ_WRITE raw_socketsN)r1socketevtfr'cCs tdS)z"Schedule callback for a raw socketNr,)r*r1r9r:r;r!r!r"_watch_raw_socketVsz_AsyncPoller._watch_raw_socket)r1socketsr'cGs tdS)z$Unschedule callback for a raw socketNr,)r*r1r=r!r!r"_unwatch_raw_socketsZsz!_AsyncPoller._unwatch_raw_socketsr&c s|dkr\ztd}Wn,tk rL}z|W5d}~XYn X|Sgfdd}fddjD]\}}t |t j rt |j sĈj |}|t j@r|jdd|t j@r|jddq|d}|t j@r|jO}|t j@r0|jO}|||qfd d }||dk r|dkrfd d } d || fdd} | fdd} | S)z Return a Future for a poll eventrNcssddSr/done set_result)argswatcherr!r"wake_rawrsz#_AsyncPoller.poll..wake_rawcsjfSr/)r>r;)r1r8r*r!r"wz#_AsyncPoller.poll..pollrc sr dSr:z Wntk r4YnXdSrRnLzttd}Wn,t k r}z|W5d}~XYn X |dSNr) rA cancelledcancel RuntimeError exception set_exceptionsuperr3rJ ExceptionrB)r;resulte) __class__rr*rEr!r" on_poll_readys z(_AsyncPoller.poll..on_poll_readycssddSr/r@r!rDr!r"trigger_timeoutsz*_AsyncPoller.poll..trigger_timeoutMbP?cs"tdrn dS)NrN)hasattrrNZremove_timeoutrG)r1timeout_handler!r"cancel_timeouts  z)_AsyncPoller.poll..cancel_timeoutcssdSr/)rArNrGrDr!r"cancel_watchersz)_AsyncPoller.poll..cancel_watcher)r%rRrJrSrQrBr+add_done_callbackr= isinstance_zmqSocketr5 from_socketr_add_recv_eventr_add_send_eventappendr6r7r< call_later) r*timeoutrTrUrFr9maskr:rWrXr\r]rV)rr1r8r*r[rEr"rJ^sP                 z_AsyncPoller.poll)r?)rrrr2r rintr rrr<r>rr rJ __classcell__r!r!rir"r3Ns   r3c@seZdZeddZdS)_NoTimercCsdSr/r!r!r!r!r"rNsz_NoTimer.cancelN)rrr staticmethodrNr!r!r!r"rlsrlTr4)boundc seZdZUdZdZdZded<eZdZ dKe dddfdd Z e dLe edeed d d ZdMe edd fd d Zejjje_fddZejjje_edNddeeeeedddZedOddeedeeeedddZedPddeedeeeejdddZedQeeeeeeeeejfdddZdReeeeeeeeejfdddZdSeeeeeeejfdddZ dTeeeee ej!ddd Z"dUeeeeeee ej!d!d"d#Z#d$d%Z$dej%feed&d'd(Z&ee'd&fd)d* Z(dVe'ee'edd,fd-d. Z)d/d0Z*d1d2Z+e,d3d4Z-dWd5d6Z.dXd7d8Z/d9d:Z0d;d<Z1dYd=d>Z2dZd?d@Z3dAdBZ4dCdDZ5dEdFZ6d[dGdHZ7dIdJZ8Z9S)\r4Nrz _zmq.Socket _shadow_sockr?) _from_socketr'c st|tjrd|}}|dk r6tj|jd||_n"tj||f|tj|j|_|dk rztj |j j dt ddt |_t |_d|_|jj|_dS)N)shadowz^(io_loop) argument is deprecated in pyzmq 22.2. The currently active loop will always be used.) stacklevelr)r_r`rarR__init__Z underlyingrprrwarningswarnrVrDeprecationWarningr _recv_futures _send_futures_stateZFD_fd)r*context socket_typeio_looprqrrir!r"rus"   z_AsyncSocket.__init__)clsr9rr'cCs |||dS)z.Create an async socket from an existing Socket)rqrr!)rr9rr!r!r"rbsz_AsyncSocket.from_socket)lingerr'c s||jsj|jdk rjtt|jpg|jp$g}|D]2}|js.z|jWq.t k r^Yq.Xq.| t j |ddS)N)r) closedr|listrryrzrrArNrO_clear_io_staterRclose)r*r event_listeventrir!r"rs z_AsyncSocket.closecs"t|}|tkr|||Sr/)rRgetr_schedule_remaining_events)r*keyrTrir!r"rs  z_AsyncSocket.getF)track)flagsrr'cCsdSr/r!)r*rrr!r!r"recv_multipart sz_AsyncSocket.recv_multipartT)rcopyrr'cCsdSr/r!r*rrrr!r!r"rscCsdSr/r!rr!r!r"rscCsdSr/r!rr!r!r"rscCs|dt|||dS)zvReceive a complete multipart zmq message. Returns a Future whose result will be a multipart message. rrrrrcdictrr!r!r"r!s cCs|dt|||dS)zReceive a single zmq frame. Returns a Future, whose result will be the received frame. Recommend using recv_multipart instead. recvrrrr!r!r"r,s z_AsyncSocket.recv) msg_partsrrr'cKs(||d<||d<||d<|jd||dS)zqSend a complete multipart zmq message. Returns a Future that resolves when sending is complete. rrrsend_multipartrr)rd)r*rrrrrr!r!r"r7sz_AsyncSocket.send_multipart)datarrrrr'cKs<||d<||d<||d<|t|||d|jd||dS)zSend a single zmq frame. Returns a Future that resolves when sending is complete. Recommend using send_multipart instead. rrrrsendr)updaterrd)r*rrrrrr!r!r"rCs z_AsyncSocket.sendcs>|fdd}|fdd}|S)zDeserialize with Futuresc str dSr$nL}z |}Wn,tk rd}z|W5d}~XYn X|dS)z+Chain result through serialization to recvdN)rArPrQrTrSrB)_bufZloadedrUr;loadrecvdr!r"_chain[s z)_AsyncSocket._deserialize.._chaincs r dSrdS)z"Chain cancellation from f to recvdN)rArMrN)r)r;rr!r" _chain_cancellsz0_AsyncSocket._deserialize.._chain_cancel)r%r^)r*rrrrr!rr" _deserializeWs   z_AsyncSocket._deserializer&csjrttj}||tt|| fdd} r`|n  |fdd} |S)zSpoll the socket for events returns a Future for the poll results. cstr dSr:z Wntk r4YnXdS|rRnt}| ddSrL) rArMrNrOrPrQrrTrBr)r;Zevtsr poll_futurer*r!r" unwrap_results  z(_AsyncSocket.poll..unwrap_resultcs.s*z Wntk r(YnXdS)z4Cancel underlying poll if request has been cancelledN)rArNrOrK)rr!r" cancel_polls  z&_AsyncSocket.poll..cancel_poll) rr`ZZMQErrorENOTSUP _poller_classregisterrrrJr%rAr^)r*rgrprrr!rr"rJws     z_AsyncSocket.pollcstj||Sr/)rR recv_string)r*rCrrir!r"rsz_AsyncSocket.recv_stringutf-8)srencodingr'cstj|||dS)N)rr)rR send_string)r*rrrrir!r"rsz_AsyncSocket.send_stringcsfdd}|||S)z'Add a timeout for a send or recv Futurecsr dStdSr/)rArQr`Againr!rKr!r"future_timeoutsz1_AsyncSocket._add_timeout..future_timeout) _call_later)r*rrgrr!rKr" _add_timeouts z_AsyncSocket._add_timeoutcCs|||S)zSchedule a function to be called later Override for different IOLoop implementations Tornado and asyncio happen to both have ioloop.call_later with the same signature. )r+rf)r*delaycallbackr!r!r"rsz_AsyncSocket._call_latercCs6t|D]\}}|j|krq$qdS|||dS)zMake sure that futures are removed from the event list when they resolve Avoids delaying cleanup until the next send/recv event, which may never come. N) enumeraterremove)rrZf_idxrr!r!r"_remove_finished_futures  z$_AsyncSocket._remove_finished_futurec s|p }|dr~|ddtj@r~tj|}z|f|}Wn,tk rn}z||W5d}~XYn X| ||St }t tdrjj } | dkr || d}jt|||d|d|fdd jtt@rjrt|S) z4Add a recv event, returning the corresponding FuturerrrNZRCVTIMEOrY)rrcs|jSr/)rryrGr.r!r"rHrIz._AsyncSocket._add_recv_event..)r% startswithrr`DONTWAITgetattrrprSrQrBrlrZZrcvtimeorryrerr^rr _handle_recv _add_io_state) r*rrrr;rrrUr timeout_msr!r.r"rcs.      z_AsyncSocket._add_recv_eventc sX|p }|dkrjs|dd}|}|tjB|d<tj|}d} z||f|} Wnhtjk r} z|tj@r| | nd} W5d} ~ XYn6t k r} z| | W5d} ~ XYn X| | | rj r |St} ttdrjtj} | dkr|| d} jt||||| d |fd d t|S) z4Add a send event, returning the corresponding Future)rrrrTFNSNDTIMEOrY)rrrcs|jSr/)rrzrGr.r!r"rH-rIz._AsyncSocket._add_send_event..)r%rzrrr`rrrprrQrSrBryrrlrZrrrerr^rr)r*rrrrr;rZ nowait_kwargsrZ finish_earlyrrUrrr!r.r"rdsB          z_AsyncSocket._add_send_eventc Cs |jtt@sdSd}|jrD|j\}}}}}|rDd}qqDq|jsT|t|dkr`dS||dkr~| ddS|dkr|jj }n|dkr|jj }n t d||dt jO<z|f|}Wn,tk r}z||W5d}~XYn X| |dS)zHandle recv eventsNrJrrzUnhandled recv event type: %rr)rprrrrypopleftrA_drop_io_staterNrBrr ValueErrorr`rrSrQ) r*r;rrrrrrTrUr!r!r"r3s6     z_AsyncSocket._handle_recvc Cs |jtt@sdSd}|jrD|j\}}}}}|rDd}qqDq|jsT|t|dkr`dS||dkr~| ddS|dkr|jj }n|dkr|jj }n t d||dt jO<z||f|}Wn,tk r}z||W5d}~XYn X| |dS)NrJrrzUnhandled send event type: %rr)rprrrrzrrArrNrBrrrr`rrSrQ) r*r;rrrrrrTrUr!r!r" _handle_send\s6     z_AsyncSocket._handle_sendcCsH|jjr dS|jt}|tj@r*||tj@r<|| dS)z(Dispatch IO events to _handle_recv, etc.N) rprrrr`rrrrr)r*fdeventsZ zmq_eventsr!r!r"_handle_eventss   z_AsyncSocket._handle_eventscCs>|jdkrdS|dkr"|jt}||j@r:|d|jdS)zkSchedule a call to handle_events next loop iteration If there are still events to handle. rN)r{rprrrr)r*rr!r!r"rs    z'_AsyncSocket._schedule_remaining_eventscCs*|j|kr|j|B}|_||jdS)zAdd io_state to poller.Nr{_update_handlerr*stater!r!r"rs z_AsyncSocket._add_io_statecCs(|j|@r|j|@|_||jdS)z&Stop poller from watching an io_state.Nrrr!r!r"rs z_AsyncSocket._drop_io_statecCs|r ||dS)zOUpdate IOLoop handler with state. zmq FD is always read-only. N)r+rrr!r!r"rsz_AsyncSocket._update_handlercCs6|dkr|}||j|j|j|d|jdS)z#initialize the ioloop event handlerNr)r+ add_handlerrprr6rr0r!r!r"r)sz_AsyncSocket._init_io_statecCs.|j}|jjr|j}|jdk r*|j|dS)zNunregister the ioloop event handler called once during close N)rprr|r$Zremove_handler)r*rr!r!r"rs  z_AsyncSocket._clear_io_state)Nr?NN)N)N)r)r)r)rTF)rTF)rTF)rTF)rTF)rr)NN)NNN)rr)N)N):rrrryrzr{rr3rr|r ru classmethodr rnrrbrjrr`rar2rrboolrr bytesrrZFramerrZMessageTrackerrrrrrJr rrrrrmrrcrdrrrrrrrr)rrkr!r!rir"r4s            3    $ 8))  )%r2rvZasyncior collectionsr itertoolsrtypingrrrrr r r r r rrrrZzmqr`rrrZ zmq._typingrrr#ZPollerr3rlrnrar4r!r!r!r"s   < m