ó Ì g]c@sdZddlZddlZddlZddlZddlmZmZddlm Z m Z ej e ƒZ defd„ƒYZdejejejejfd„ƒYZd efd „ƒYZd ejfd „ƒYZd ejfd„ƒYZdefd„ƒYZdS(s¡ Implementation of `nbio_interface.AbstractIOServices` on top of a selector-based I/O loop, such as tornado's and our home-grown select_connection's I/O loops. iÿÿÿÿN(tnbio_interfacetio_services_utils(tcheck_callback_argt check_fd_argtAbstractSelectorIOLoopcBsøeZdZeejd„ƒƒZeejd„ƒƒZeejd„ƒƒZejd„ƒZ ejd„ƒZ ejd„ƒZ ejd„ƒZ ejd„ƒZ ejd „ƒZejd „ƒZejd „ƒZejd „ƒZRS( s(Selector-based I/O loop interface expected by `selector_ioloop_adapter.SelectorIOServicesAdapter` NOTE: this interface follows the corresponding methods and attributes of `tornado.ioloop.IOLoop` in order to avoid additional adapter layering when wrapping tornado's IOLoop. cCsdS(sThe value of the I/O loop's READ flag; READ/WRITE/ERROR may be used with bitwise operators as expected. Implementation note: the implementations can simply replace these READ/WRITE/ERROR properties with class-level attributes N((tself((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pytREADscCsdS(syThe value of the I/O loop's WRITE flag; READ/WRITE/ERROR may be used with bitwise operators as expected N((R((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pytWRITE'scCsdS(syThe value of the I/O loop's ERROR flag; READ/WRITE/ERROR may be used with bitwise operators as expected N((R((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pytERROR/scCsdS(sRelease IOLoop's resources. the `close()` method is intended to be called by the application or test code only after `start()` returns. After calling `close()`, no other interaction with the closed instance of `IOLoop` should be performed. N((R((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pytclose7scCsdS(sORun the I/O loop. It will loop until requested to exit. See `stop()`. N((R((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pytstartAscCsdS(s%Request exit from the ioloop. The loop is NOT guaranteed to stop before this method returns. To invoke `stop()` safely from a thread other than this IOLoop's thread, call it via `add_callback_threadsafe`; e.g., `ioloop.add_callback(ioloop.stop)` N((R((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pytstopGscCsdS(sªAdd the callback to the IOLoop timer to be called after delay seconds from the time of call on best-effort basis. Returns a handle to the timeout. :param float delay: The number of seconds to wait to call callback :param callable callback: The callback method :returns: handle to the created timeout that may be passed to `remove_timeout()` :rtype: object N((Rtdelaytcallback((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyt call_laterSscCsdS(sVRemove a timeout :param timeout_handle: Handle of timeout to remove N((Rttimeout_handle((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pytremove_timeoutascCsdS(súRequests a call to the given function as soon as possible in the context of this IOLoop's thread. NOTE: This is the only thread-safe method in IOLoop. All other manipulations of IOLoop must be performed from the IOLoop's thread. For example, a thread may request a call to the `stop` method of an ioloop that is running in a different thread via `ioloop.add_callback_threadsafe(ioloop.stop)` :param callable callback: The callback method N((RR ((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyt add_callbackiscCsdS(sStart watching the given file descriptor for events :param int fd: The file descriptor :param callable handler: When requested event(s) occur, `handler(fd, events)` will be called. :param int events: The event mask using READ, WRITE, ERROR. N((Rtfdthandlertevents((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyt add_handleryscCsdS(s˜Changes the events we watch for :param int fd: The file descriptor :param int events: The event mask using READ, WRITE, ERROR N((RRR((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pytupdate_handler„scCsdS(shStop watching the given file descriptor for events :param int fd: The file descriptor N((RR((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pytremove_handlers(t__name__t __module__t__doc__tpropertytabctabstractmethodRRRR R R RRRRRR(((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyRs      tSelectorIOServicesAdaptercBseZdZd„Zd„Zd„Zd„Zd„Zd„Zd„Z ddddd „Z d „Z d „Z d „Z d „Zd„ZRS(sˆImplements the :py:class:`.nbio_interface.AbstractIOServices` interface on top of selector-style native loop having the :py:class:`AbstractSelectorIOLoop` interface, such as :py:class:`pika.selection_connection.IOLoop` and :py:class:`tornado.IOLoop`. NOTE: :py:class:`.nbio_interface.AbstractFileDescriptorServices` interface is only required by the mixins. cCsA||_tƒ|_|jj|_|jj|jjB|_dS(s½ :param AbstractSelectorIOLoop native_loop: An instance compatible with the `AbstractSelectorIOLoop` interface, but not necessarily derived from it. N(t_looptdictt _watchersRt_readable_maskRRt_writable_mask(Rt native_loop((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyt__init__¦s  cCs|jS(s^Implement :py:meth:`.nbio_interface.AbstractIOServices.get_native_ioloop()`. (R(R((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pytget_native_ioloopºscCs|jjƒdS(sJImplement :py:meth:`.nbio_interface.AbstractIOServices.close()`. N(RR (R((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyR ÁscCs|jjƒdS(sHImplement :py:meth:`.nbio_interface.AbstractIOServices.run()`. N(RR (R((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pytrunÇscCs|jjƒdS(sIImplement :py:meth:`.nbio_interface.AbstractIOServices.stop()`. N(RR (R((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyR ÍscCs|jj|ƒdS(sdImplement :py:meth:`.nbio_interface.AbstractIOServices.add_callback_threadsafe()`. N(RR(RR ((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pytadd_callback_threadsafeÓscCst|jj||ƒ|jƒS(sOImplement :py:meth:`.nbio_interface.AbstractIOServices.call_later()`. (t _TimerHandleRR(RR R ((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyRÚsicCsFttd|jd|d|d|d|d|d|d|ƒjƒƒS( sPImplement :py:meth:`.nbio_interface.AbstractIOServices.getaddrinfo()`. R$thosttporttfamilytsocktypetprototflagston_done(t_SelectorIOLoopIOHandlet_AddressResolverRR (RR*R+R0R,R-R.R/((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyt getaddrinfoàs  cCstjd||ƒt|ƒt|dƒy|j|}WnStk r|jj||j|j ƒt d|ƒ|j|((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyt remove_readers&    cCstjd||ƒt|ƒt|dƒy|j|}WnStk r|jj||j|j ƒt d|ƒ|j|((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyt set_writer3s$      cCs÷tjd|ƒt|ƒy|j|}Wn"tk rOtjd|ƒtSX|jdkrˆ|jdk stt ‚tjd|ƒtSd|_|jdkrÍ|j|=|j j |ƒtjd|ƒn&|j j ||j ƒtjd|ƒtS(sfImplement :py:meth:`.nbio_interface.AbstractFileDescriptorServices.remove_writer()`. s+SelectorIOServicesAdapter.remove_writer(%s)s"remove_writer(%s) neither was set.s&remove_writer(%s) writer wasn't set Rds!remove_writer(%s) removed handlers$remove_writer(%s) updated handler RdN(R6R7RR!R8R@R<R;R5R=RRRR"RA(RRR>((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyt remove_writerPs&    cCsº|j|}||j@rE|jdkrEtjd|t|ƒƒn||j@rŠ|jdk rn|jƒqŠtjd|t|ƒƒn||j@r¶|jdk r¶|jƒq¶ndS(säHandle indicated file descriptor events requested via `set_reader()` and `set_writer()`. :param fd: file descriptor :param events: event mask using native loop's READ/WRITE/ERROR. NOTE: depending on the underlying poller mechanism, ERROR may be indicated upon certain file description state even though we don't request it. We ignore ERROR here since `set_reader()`/`set_writer()` don't request for it. s?READ indicated on fd=%s, but reader callback is None; events=%ss@WRITE indicated on fd=%s, but writer callback is None; events=%sN( R!R"R5R;R6twarningtbinR#R<(RRRR>((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyR9ps     (RRRR%R&R R'R R(RR3R?RBRDRER9(((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyR–s"           R:cBs#eZdZdZddd„ZRS(s7Holds reader and writer callbacks for a file descriptorR5R<cCs||_||_dS(N(R5R<(RR5R<((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyR%›s (sreaderswriterN(RRRt __slots__R;R%(((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyR:–sR)cBs eZdZd„Zd„ZRS(sJThis module's adaptation of `nbio_interface.AbstractTimerReference`. cCs||_||_dS(s :param opaque handle: timer handle from the underlying loop implementation that may be passed to its `remove_timeout()` method :param AbstractSelectorIOLoop loop: the I/O loop instance that created the timeout. N(t_handleR(Rthandletloop((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyR%¦s cCs;|jdk r7|jj|jƒd|_d|_ndS(N(RR;RRI(R((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pytcancel±s (RRRR%RL(((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyR)¡s R1cBs eZdZd„Zd„ZRS(sFThis module's adaptation of `nbio_interface.AbstractIOReference` cCs|j|_dS(sZ :param subject: subject of the reference containing a `cancel()` method N(RLt_cancel(Rtsubject((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyR%½scCs |jƒS(sCancel pending operation :returns: False if was already done or cancelled; True otherwise :rtype: bool (RM(R((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyRLÄs(RRRR%RL(((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyR1¸s R2cBs\eZdZdZdZdZdZd„Zd„Zd„Z d„Z d „Z d „Z RS( s&Performs getaddrinfo asynchronously using a thread, then reports result via callback from the given I/O loop. NOTE: at this stage, we're using a thread per request, which may prove inefficient and even prohibitive if the app performs many of these operations concurrently. iiiic Cs†t|dƒ|j|_d|_||_||_||_||_||_ ||_ ||_ ||_ t jƒ|_d|_dS(sŠ :param AbstractSelectorIOLoop native_loop: :param host: `see socket.getaddrinfo()` :param port: `see socket.getaddrinfo()` :param family: `see socket.getaddrinfo()` :param socktype: `see socket.getaddrinfo()` :param proto: `see socket.getaddrinfo()` :param flags: `see socket.getaddrinfo()` :param on_done: on_done(records|BaseException) callback for reporting result from the given I/O loop. The single arg will be either an exception object (check for `BaseException`) in case of failure or the result returned by `socket.getaddrinfo()`. R0N(Rt NOT_STARTEDt_stateR;t_resultRt_hostt_portt_familyt _socktypet_protot_flagst_on_donet threadingtLockt_mutext_threading_timer( RR$R*R+R,R-R.R/R0((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyR%Ûs           cCsd|_d|_d|_dS(sRelease resources N(R;RR\RX(R((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyt_cleanupûs  cCs\|j|jks!t|jƒ‚|j|_tjd|jƒ|_|jjƒt |ƒS(s\Start asynchronous DNS lookup. :rtype: nbio_interface.AbstractIOReference i( RPROR=tACTIVERYtTimert_resolveR\R R1(R((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyR s !  c Cs‰|jz|j|jkr\tjd|j|jƒ|j|_|jj ƒ|j ƒt Stjd|j|j|jƒt SWdQXdS(s„Cancel the pending resolver :returns: False if was already done or cancelled; True otherwise :rtype: bool sCanceling resolver for %s:%ssKIgnoring _AddressResolver cancel request when not ACTIVE; (%s:%s); state=%sN( R[RPR^R6R7RRRStCANCELEDR\RLR]RAR@(R((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyRLs     cCs¸y4tj|j|j|j|j|j|jƒ}Wn)tk r_}t j d|ƒ|}nX||_ |j @|j |jkr›|jj|jƒnt jd|jƒWdQXdS(srCall `socket.getaddrinfo()` and return result via user's callback function on the given I/O loop sAddress resolution failed: %rsBAsynchronous getaddrinfo cancellation detected; in thread; host=%rN(tsocketR3RRRSRTRURVRWt ExceptionR6terrorRQR[RPR^RRt_dispatch_resultR7(Rtresulttexc((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyR`+s    cCsm|j|jkrV|j|_z'tjd|jƒ|j|jƒWd|jƒXntjd|jƒdS(syThis is called from the user's I/O loop to pass the result to the user via the user's on_done callback s@Invoking asynchronous getaddrinfo() completion callback; host=%rNsLAsynchronous getaddrinfo cancellation detected; in I/O loop context; host=%r( RPR^t COMPLETEDR6R7RRRXRQR](R((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyReDs  ( RRRROR^RaRhR%R]R RLR`Re(((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyR2Îs    (RRtloggingRbRYtpika.adapters.utilsRRt%pika.adapters.utils.io_services_utilsRRt getLoggerRR6tobjectRtSocketConnectionMixintStreamingConnectionMixintAbstractIOServicestAbstractFileDescriptorServicesRR:tAbstractTimerReferenceR)tAbstractIOReferenceR1R2(((sP/srv/kernel/kteam-tools/dashboard/pika/adapters/utils/selector_ioloop_adapter.pyts     ƒ ý