o inX@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!F/var/www/edux/Edux_v2/venv/lib/python3.10/site-packages/zmq/_future.pyrs  rc@sReZdZUdZdZeed<eeed<defddZ defdd Z d d d d Z dS)_AsynczMixin for common async logicN _current_loop_FuturereturncCsL|jdur||_||j|jS|}||jur$||_|||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)self current_loopr!r!r" _get_loop6s     z_Async._get_loopcCstd)Nz!Must be implemented in a subclassNotImplementedErrorr)r!r!r"r'Gsz_Async._default_loopcCdSNr!r)loopr!r!r"r(Jsz_Async._init_io_stater0)r&N) rrr__doc__r$rrr rr+r'r(r!r!r!r"r#0s   r#c seZdZUdZeded<eed<eed<eeed<deded ed e d d f d dZ deded d fddZ dd e ee eefffdd ZZS) _AsyncPollerz:Poller that returns a Future on poll, instead of blocking. _AsyncSocket _socket_class_READ_WRITE raw_socketsr2socketevtfr&NcCt)z"Schedule callback for a raw socketr,)r)r2r:r;r<r!r!r"_watch_raw_socketVz_AsyncPoller._watch_raw_socketsocketscGr=)z$Unschedule callback for a raw socketr,)r)r2r@r!r!r"_unwatch_raw_socketsZr?z!_AsyncPoller._unwatch_raw_socketsc s|dkr0ztd}Wnty(}z |WYd}~Sd}~ww|Sgfdd}fddjD]R\}}t |t j r}t |j sdj |}|t j@rp|jdd|t j@r||jddqN|d}|t j@r|jO}|t j@r|jO}|||qNfd d }||dur|dkrчfd d } d || fdd} | fdd} | S)z Return a Future for a poll eventrNcs ddSdSr0done set_result)argswatcherr!r"wake_rawrz#_AsyncPoller.poll..wake_rawcsjgRSr0)rAr<)r2r9r)r!r"wsz#_AsyncPoller.poll..pollrc srdSrzWdStyYdSwr)dSz ttd}Wnt yK}z |WYd}~dSd}~ww |dSNr) rE cancelledcancel RuntimeError exception set_exceptionsuperr4rN ExceptionrF)r<resulte) __class__rr)rIr!r" on_poll_readys&  z(_AsyncPoller.poll..on_poll_readycrCr0rDr!rHr!r"trigger_timeoutrKz*_AsyncPoller.poll..trigger_timeoutMbP?cs$tdr dSdS)NrR)hasattrrRremove_timeoutrL)r2timeout_handler!r"cancel_timeouts  z)_AsyncPoller.poll..cancel_timeoutcss dSdSr0)rErRrLrHr!r"cancel_watchers z)_AsyncPoller.poll..cancel_watcher)r%rVrNrWrUrFr+add_done_callbackr@ isinstance_zmqSocketr6 from_socketr_add_recv_eventr_add_send_eventappendr7r8r> call_later) r)timeoutrXrYrJr:maskr;r[r\rarbrZ)rr2r9r)r`rIr"rN^sX                 z_AsyncPoller.poll)rB)rrrr3r rintr rrr>rArr rN __classcell__r!r!rnr"r4Ns   ,r4c@seZdZeddZdS)_NoTimercCr/r0r!r!r!r!r"rRsz_NoTimer.cancelN)rrr staticmethodrRr!r!r!r"rqsrqTr5)boundcseZdZUdZdZdZded<eZdZ    dPde dddffdd Z e dQd e ed dd edefd dZdQde eddffdd Zejjje_fddZejjje_e dRdddededeeefddZe dRdddededdedeeefddZe dRdddededdedeeejfddZe dSdedededeeeeeejffddZ dSdedededeeeeeejffddZ dSdedededeeeejffd d!Z dSd"edededee ej!fd#d$Z"   dSd%edededed&edee ej!f d'd(Z#d)d*Z$dej%fdeefd+d,Z&dee'ffd-d. Z(dTd0e'ded1e'dedffd2d3 Z)d4d5Z*d6d7Z+e,d8d9Z-dUd:d;Z.dVdd?Z0d@dAZ1dWdBdCZ2dQdDdEZ3dFdGZ4dHdIZ5dJdKZ6dQdLdMZ7dNdOZ8Z9S)Xr5Nrz _zmq.Socket _shadow_sockrB _from_socketr&c st|tjr d|}}|durtj|jd||_ntj||fi|tj|j|_|dur?tj |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)rdrerfrV__init__ underlyingrurwwarningswarnrZrDeprecationWarningr _recv_futures _send_futures_stateFD_fd)r)context socket_typeio_looprvrrnr!r"rzs"   z_AsyncSocket.__init__clsr:rcCs |||dS)z.Create an async socket from an existing Socket)rvrr!)rr:rr!r!r"rgs z_AsyncSocket.from_socketlingerc sz|js4|jdur4tt|jpg|jpg}|D]}|js/z|jWqt y.Yqwq| t j |ddS)N)r) closedrlistrrrrrErRrS_clear_io_staterVclose)r)r event_listeventrnr!r"rs  z_AsyncSocket.closecs"t|}|tkr|||Sr0)rVgetr_schedule_remaining_events)r)keyrXrnr!r"rs  z_AsyncSocket.getF)trackflagsrcCr/r0r!)r)rrr!r!r"recv_multipart z_AsyncSocket.recv_multipartcopyTcCr/r0r!r)rrrr!r!r"rrcCr/r0r!rr!r!r"rrcCr/r0r!rr!r!r"rrcC|dt|||dS)zvReceive a complete multipart zmq message. Returns a Future whose result will be a multipart message. rrrrrhdictrr!r!r"r!scCr)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_partscKs(||d<||d<||d<|jd||dS)zqSend a complete multipart zmq message. Returns a Future that resolves when sending is complete. rrrsend_multipartrr)ri)r)rrrrrr!r!r"r7sz_AsyncSocket.send_multipartdatarcKs<||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)updaterri)r)rrrrrr!r!r"rCs z_AsyncSocket.sendcs>|fdd}|fdd}|S)zDeserialize with Futuresc szrdSrdS}z|}Wnty5}z |WYd}~dSd}~ww|dS)z+Chain result through serialization to recvdN)rErTrUrXrWrF)_bufloadedrYr<loadrecvdr!r"_chain[s z)_AsyncSocket._deserialize.._chaincs$rdSrdSdS)z"Chain cancellation from f to recvdN)rErQrR)r)r<rr!r" _chain_cancells  z0_AsyncSocket._deserialize.._chain_cancel)r%rc)r)rrrrr!rr" _deserializeWs   z_AsyncSocket._deserializecsnjr ttj}||tt||} fdd}| r0||S| |S)zSpoll the socket for events returns a Future for the poll results. cstrdS|rzWdStyYdSw|r)|dSt|}| ddSrP) rErQrRrSrTrUrrXrFr)r<evtsrr)r!r" unwrap_results   z(_AsyncSocket.poll..unwrap_result) rreZMQErrorENOTSUP _poller_classregisterrrrNr%rErc)r)rlrpr<rr!rr"rNws   z_AsyncSocket.pollcstj|i|Sr0)rV recv_string)r)rGrrnr!r"rz_AsyncSocket.recv_stringutf-8sencodingcstj|||dS)N)rr)rV send_string)r)rrrrnr!r"rrz_AsyncSocket.send_stringcsfdd}|||S)z'Add a timeout for a send or recv FuturecsrdStdSr0)rErUreAgainr!rOr!r"future_timeoutsz1_AsyncSocket._add_timeout..future_timeout) _call_later)r)rrlrr!rOr" _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+rk)r)delaycallbackr!r!r"rsz_AsyncSocket._call_latercCs6t|D] \}}|j|urnqdS|||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)rrf_idxrr!r!r"_remove_finished_futures  z$_AsyncSocket._remove_finished_futurec s |p}|drC|ddtj@rCtj|}z |d i|}Wnty;}z ||WYd}~|Sd}~ww| ||St }t tdrZjj } | dkrZ || d}jt|||d|d|fdd jtt@r|jrt|S) z4Add a recv event, returning the corresponding FuturerrrNRCVTIMEOr])rrc|jSr0)rrrLr.r!r"rMz._AsyncSocket._add_recv_event..r!)r% startswithrreDONTWAITgetattrrurWrUrFrqr^rcvtimeorrrjrrcrr _handle_recv _add_io_state) r)rrrr<rrrYr timeout_msr!r.r"rhs4      z_AsyncSocket._add_recv_eventc s\|p}|dvrwjsw|dd}|}|tjB|d<tj|}d} z ||fi|} Wn6tjyP} z|tj@rD| | nd} WYd} ~ n d} ~ wt yf} z | | WYd} ~ n d} ~ ww| | | rwj ru |St} ttdrjtj} | dkr|| d} jt||||| d |fd d t|S) z4Add a send event, returning the corresponding Future)rrrrTFNSNDTIMEOr])rrrcrr0)rrrLr.r!r"rM!rz._AsyncSocket._add_send_event..)r%rrrrerrrurrUrWrFrrrqr^rrrjrrcrr)r)rrrrr<r nowait_kwargsr finish_earlyrrYrrr!r.r"risH         z_AsyncSocket._add_send_eventc Cs|jtt@s dSd}|jr$|j\}}}}}|r d}nn|js|js,|t|dur2dS||dkrA| ddS|dkrJ|jj }n|dkrS|jj }nt d||dt jO<z |di|}Wnty}z ||WYd}~dSd}~ww| |dS)zHandle recv eventsNrNrrzUnhandled recv event type: %rrr!)rurrrrpopleftrE_drop_io_staterRrFrr ValueErrorrerrWrU) r)r<rrrrrrXrYr!r!r"r's<     z_AsyncSocket._handle_recvc Cs|jtt@s dSd}|jr$|j\}}}}}|r d}nn|js|js,|t|dur2dS||dkrA| ddS|dkrJ|jj }n|dkrS|jj }nt d||dt jO<z ||fi|}Wnty}z ||WYd}~dSd}~ww| |dS)NrNrrzUnhandled send event type: %rr)rurrrrrrErrRrFrrrrerrWrU) r)r<rrrrrrXrYr!r!r" _handle_sendPs<     z_AsyncSocket._handle_sendcCsH|jjrdS|jt}|tj@r||tj@r|| dS)z(Dispatch IO events to _handle_recv, etc.N) rurrrrerrrrr)r)fdevents zmq_eventsr!r!r"_handle_eventsys    z_AsyncSocket._handle_eventscCsB|jdkrdS|dur|jt}||j@r|d|jdSdS)zkSchedule a call to handle_events next loop iteration If there are still events to handle. rN)rrurrrr)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"rs z_AsyncSocket._update_handlercCs6|dur|}||j|j|j|d|jdS)z#initialize the ioloop event handlerNr)r+ add_handlerrurr7rr1r!r!r"r(sz_AsyncSocket._init_io_statecCs2|j}|jjr |j}|jdur|j|dSdS)zNunregister the ioloop event handler called once during close N)rurrr$remove_handler)r)rr!r!r"rs  z_AsyncSocket._clear_io_state)NrBNNr0)r)rTF)rr)NN)NNN)rr):rrrrrrrr4rrr rz classmethodr rsrrgrorrerfr3rrboolrr bytesrrFramerrMessageTrackerrrrrrNr rrrrrrrrhrirrrrrrrr(rrpr!r!rnr"r5s "                   '$    $8) )  )%r3r|asyncior collectionsr itertoolsrtypingrrrrr r r r r rrrrzmqrerrr zmq._typingrrr#Pollerr4rqrsrfr5r!r!r!r"s   < m