B b ,c6X@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!*lib/python3.7/site-packages/zmq/_future.pyrs rc@sVeZdZUdZdZeed<eeed<edddZ eddd Z d ddd d Z dS) _AsynczMixin for common async logicN _current_loop_Future)returncCsL|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 subclass)NotImplementedError)r)r!r!r"r'Gsz_Async._default_loopcCsdS)Nr!)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)r,socketevtfr&cCs tdS)z"Schedule callback for a raw socketN)r+)r)r,r4r5r6r!r!r"_watch_raw_socketVsz_AsyncPoller._watch_raw_socket)r,socketsr&cGs tdS)z$Unschedule callback for a raw socketN)r+)r)r,r8r!r!r"_unwatch_raw_socketsZsz!_AsyncPoller._unwatch_raw_sockets)r&c s|dkr\ytd}Wn,tk rL}z|Wdd}~XYn X|Sgfdd}fddxjD]\}}t |t j rt |j sƈj |}|t j@r|jdd|t j@r|jddq|d}|t j@r|jO}|t j@r2|jO}|||qWfd d }||dk r|dkrfd d } d || fdd} | fdd} | S)z Return a Future for a poll eventrNcssddS)N)done set_result)args)watcherr!r"wake_rawrsz#_AsyncPoller.poll..wake_rawcsjfS)N)r9)r6)r,r3r)r!r"wz#_AsyncPoller.poll..poll)rc sr dSr:y Wntk r4YnXdSrRnLyttd}Wn,t k r}z|Wdd}~XYn X |dS)Nr) r; cancelledcancel RuntimeError exception set_exceptionsuperr.rB Exceptionr<)r6resulte) __class__rr)r>r!r" on_poll_readys z(_AsyncPoller.poll..on_poll_readycssddS)N)r;r<r!)r>r!r"trigger_timeoutsz*_AsyncPoller.poll..trigger_timeoutgMbP?cs"tdrn dS)NrD)hasattrrDZremove_timeout)r6)r,timeout_handler!r"cancel_timeouts  z)_AsyncPoller.poll..cancel_timeoutcssdS)N)r;rD)r6)r>r!r"cancel_watchersz)_AsyncPoller.poll..cancel_watcher)r%rHrBrIrGr<r*add_done_callbackr8 isinstance_zmqSocketr0 from_socketr_add_recv_eventr_add_send_eventappendr1r2r7 call_later) r)timeoutrJrKr?r4maskr5rMrNrQrR)rL)rr,r3r)rPr>r"rB^sN                 z_AsyncPoller.poll)r:)rrrr-r rintr rrr7r9rr rB __classcell__r!r!)rLr"r.Ns   r.c@seZdZeddZdS)_NoTimercCsdS)Nr!r!r!r!r"rDsz_NoTimer.cancelN)rrr staticmethodrDr!r!r!r"r`sr`Tr/)Zboundc 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)\r/Nrz _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)rTrUrVrH__init__Z underlyingrcrewarningswarnrLrDeprecationWarningr _recv_futures _send_futures_stateZFD_fd)r)contextZ socket_typeio_looprdr)rLr!r"rhs    z_AsyncSocket.__init__)clsr4rqr&cCs |||dS)z.Create an async socket from an existing Socket)rdrqr!)rrr4rqr!r!r"rWsz_AsyncSocket.from_socket)lingerr&c s|jsn|jdk rntt|jpg|jp$g}x:|D]2}|js0y|jWq0t k r`Yq0Xq0W| t j |ddS)N)rs) closedrolistrrlrmrr;rDrE_clear_io_staterHclose)r)rs event_listevent)rLr!r"rws   z_AsyncSocket.closecs"t|}|tkr|||S)N)rHgetr_schedule_remaining_events)r)keyrJ)rLr!r"rzs  z_AsyncSocket.getF)track)flagsr}r&cCsdS)Nr!)r)r~r}r!r!r"recv_multipart sz_AsyncSocket.recv_multipartT)r~copyr}r&cCsdS)Nr!)r)r~rr}r!r!r"rscCsdS)Nr!)r)r~rr}r!r!r"rscCsdS)Nr!)r)r~rr}r!r!r"rscCs|dt|||dS)zvReceive a complete multipart zmq message. Returns a Future whose result will be a multipart message. r)r~rr})rXdict)r)r~rr}r!r!r"r"scCs|dt|||dS)zReceive a single zmq frame. Returns a Future, whose result will be the received frame. Recommend using recv_multipart instead. recv)r~rr})rXr)r)r~rr}r!r!r"r-s z_AsyncSocket.recv) msg_partsr~rr&cKs(||d<||d<||d<|jd||dS)zqSend a complete multipart zmq message. Returns a Future that resolves when sending is complete. r~rr}send_multipart)rr)rY)r)rr~rr}rr!r!r"r8sz_AsyncSocket.send_multipart)datar~rr}rr&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. r~rr})r~rr}send)rr)updaterrY)r)rr~rr}rr!r!r"rDs z_AsyncSocket.sendcs>|fdd}|fdd}|S)zDeserialize with Futuresc str dSr$nL}y |}Wn,tk rd}z|Wdd}~XYn X|dS)z+Chain result through serialization to recvdN)r;rFrGrJrIr<)_ZbufZloadedrK)r6loadrecvdr!r"_chain\s z)_AsyncSocket._deserialize.._chaincs r dSrdS)z"Chain cancellation from f to recvdN)r;rCrD)r)r6rr!r" _chain_cancelmsz0_AsyncSocket._deserialize.._chain_cancel)r%rS)r)rrrrr!)r6rrr" _deserializeXs   z_AsyncSocket._deserialize)r&csljrttj}||tt||} fdd}| r^||n | |S)zSpoll the socket for events returns a Future for the poll results. cstr dS|r:y Wntk r4YnXdS|rR|nt|}| ddS)Nr) r;rCrDrErFrGrrJr<rz)r6Zevts)rr)r!r" unwrap_results  z(_AsyncSocket.poll..unwrap_result) rtrUZZMQErrorZENOTSUP _poller_classregisterrrrBr%r;rS)r)r\r~pr6rr!)rr)r"rBxs    z_AsyncSocket.pollcstj||S)N)rH recv_string)r)r=r)rLr!r"rsz_AsyncSocket.recv_stringutf-8)sr~encodingr&cstj|||dS)N)r~r)rH send_string)r)rr~r)rLr!r"rsz_AsyncSocket.send_stringcsfdd}|||S)z'Add a timeout for a send or recv Futurecsr dStdS)N)r;rGrUAgainr!)rr!r"future_timeoutsz1_AsyncSocket._add_timeout..future_timeout) _call_later)r)rr\rr!)rr" _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*r[)r)Zdelaycallbackr!r!r"rsz_AsyncSocket._call_latercCs8x$t|D]\}}|j|kr Pq WdS|||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)rrxZf_idxryr!r!r"_remove_finished_futures  z$_AsyncSocket._remove_finished_futurec s|p }|dr~|ddtj@r~tj|}y|f|}Wn,tk rn}z||Wdd}~XYn X| ||St }t tdrjj } | dkr || d}jt|||d|d|fdd jtt@rjrt|S) z4Add a recv event, returning the corresponding Futurerr~rNZRCVTIMEOgMbP?)rrcs|jS)N)rrl)r6)r)r!r"r@rAz._AsyncSocket._add_recv_event..)r% startswithrzrUDONTWAITgetattrrcrIrGr<r`rOZrcvtimeorrlrZrrSrr _handle_recv _add_io_state) r)rrrr6rrrKr timeout_msr!)r)r"rXs,     z_AsyncSocket._add_recv_eventc sX|p }|dkrjs|dd}|}|tjB|d<tj|}d} y||f|} Wnhtjk r} z|tj@r| | nd} Wdd} ~ XYn6t k r} z| | Wdd} ~ XYn X| | | rj r |St} ttdrjtj} | dkr|| d} jt||||| d |fd d t|S) z4Add a send event, returning the corresponding Future)rrr~rTFNSNDTIMEOgMbP?)rrrcs|jS)N)rrm)r6)r)r!r"r@"rAz._AsyncSocket._add_send_event..)r%rmrzrrUrrrcrrGrIr<rlr{r`rOrrrZrrSrr)r)rrrrr6r~Z nowait_kwargsrZ finish_earlyrrKrrr!)r)r"rYs>         z_AsyncSocket._add_send_eventc Cs|jtt@sdSd}x.|jrF|j\}}}}}|rBd}qPqW|jsX|t|dkrddS||dkr| ddS|dkr|jj }n|dkr|jj }n t d||dt jO<y|f|}Wn.tk r}z||Wdd}~XYn X| |dS)zHandle recv eventsNrBrrzUnhandled recv event type: %rr~)rcrzrrrlpopleftr;_drop_io_staterDr<rr ValueErrorrUrrIrG) r)r6rrrrrrJrKr!r!r"r(s6     z_AsyncSocket._handle_recvc Cs|jtt@sdSd}x.|jrF|j\}}}}}|rBd}qPqW|jsX|t|dkrddS||dkr| ddS|dkr|jj }n|dkr|jj }n t d||dt jO<y||f|}Wn.tk r}z||Wdd}~XYn X| |dS)NrBrrzUnhandled send event type: %rr~)rcrzrrrmrr;rrDr<rrrrUrrIrG) r)r6rrrrrrJrKr!r!r" _handle_sendQs6     z_AsyncSocket._handle_sendcCs<|jt}|tj@r||tj@r0||dS)z(Dispatch IO events to _handle_recv, etc.N) rcrzrrUrrrrr{)r)fdeventsZ zmq_eventsr!r!r"_handle_eventszs    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)rnrcrzrrr)r)rr!r!r"r{s    z'_AsyncSocket._schedule_remaining_eventscCs*|j|kr|j|B}|_||jdS)zAdd io_state to poller.N)rn_update_handler)r)stater!r!r"rs z_AsyncSocket._add_io_statecCs(|j|@r|j|@|_||jdS)z&Stop poller from watching an io_state.N)rnr)r)rr!r!r"rs z_AsyncSocket._drop_io_statecCs|r ||dS)zOUpdate IOLoop handler with state. zmq FD is always read-only. N)r*r{)r)rr!r!r"rsz_AsyncSocket._update_handlercCs6|dkr|}||j|j|j|d|jdS)z#initialize the ioloop event handlerNr)r*Z add_handlerrcrr1r)r)r,r!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)rcrtror$Zremove_handler)r)rr!r!r"rvs  z_AsyncSocket._clear_io_state)Nr:NN)N)N)r)r)r)rTF)rTF)rTF)rTF)rTF)rr)NN)NNN)rr)N)N):rrrrlrmrnrr.rror rh classmethodr rbrrWr^rwrUrVr-rzrboolrr bytesrrZFramerrZMessageTrackerrrrrrBr rrrrrarrXrYrrrr{rrrr(rvr_r!r!)rLr"r/st     &$      '   $ 8))  )%r-riZasyncior collectionsr itertoolsrtypingrrrrr r r r r rrrrZzmqrUrrrZ zmq._typingrrr#ZPollerr.r`rbrVr/r!r!r!r"s   < m