)Jfj`dZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl mZddlmZmZejeZdZiZe jjrdee<deej<dee<d ee<eeZd Z Gd d eZ!Gd de"Z#Gdde"Z$Gdde"Z%GddeZ&Gdde jj'Z(Gdde(Z)Gdde(Z*Gdde(Z+Gdde+Z,dS)zeA connection adapter that tries to use the best polling method for the platform pika is running on. N)nbio_interface)BaseConnection)SelectorIOServicesAdapterAbstractSelectorIOLoopcdS)NTes b/home/alex/cs2snipeproduction/venv/lib/python3.11/site-packages/pika/adapters/select_connection.pyr !sc8|jdtjkSNr)argserrnoEINTRr s r r r #sek1Ir c,|jtjkSNrrr s r r r $AGu{,Br c,|jtjkSrrr s r r r %rr c`t|jd}| ||SdS)z{Check if caught exception represents EINTR error. :param exc: exception; must be one of classes in _SELECT_ERRORS NF)_SELECT_ERROR_CHECKERSget __class__)exccheckers r _is_resumabler1s4 %((==Gws||ur cReZdZdZ dfd Ze ddZdZxZS) SelectConnectionzxAn asynchronous connection adapter that attempts to use the fastest event loop adapter for the given platform. NTct|tjr|}nt|p t }t t |||||||dS)aCreate a new instance of the Connection object. :param pika.connection.Parameters parameters: Connection parameters :param callable on_open_callback: Method to call on connection open :param None | method on_open_error_callback: Called if the connection can't be established or connection establishment is interrupted by `Connection.close()`: on_open_error_callback(Connection, exception). :param None | method on_close_callback: Called when a previously fully open connection is closed: `on_close_callback(Connection, exception)`, where `exception` is either an instance of `exceptions.ConnectionClosed` if closed by user or broker or exception of another type that describes the cause of connection failure. :param None | IOLoop | nbio_interface.AbstractIOServices custom_ioloop: Provide a custom I/O Loop object. :param bool internal_connection_workflow: True for autonomous connection establishment which is default; False for externally-managed connection workflow via the `create_connection()` factory. :raises: RuntimeError )internal_connection_workflowN) isinstancerAbstractIOServicesrIOLoopsuperr __init__) self parameterson_open_callbackon_open_error_callbackon_close_callback custom_ioloopr"nbiors r r'zSelectConnection.__init__Cs: m^%F G G H DD,]-FfhhGGD %%..   "  )E / G G G G Gr ct|p tfd}||||S)z`Implement :py:classmethod::`pika.adapters.BaseConnection.create_connection()`. cB|td|dS)zConnection factory.NzIExpected pika.connection.Parameters instance, but got None in params arg.F)r)r-r") ValueError)paramsclsr.s r connection_factoryz>SelectConnection.create_connection..connection_factoryysE~ "IJJJ3!"-2444 4r )connection_configsr4r.workflowon_done)rr%_start_connection_workflow)r3r5r7r-r6r4r.s` @r create_connectionz"SelectConnection.create_connectionmsh))B&((CC 4 4 4 4 4 4--11 . r c4|jS)ze :returns: Current size of output data buffered by the transport :rtype: int ) _transportget_write_buffer_sizer(s r _get_write_buffer_sizez'SelectConnection._get_write_buffer_sizes 44666r )NNNNNT)NN) __name__ __module__ __qualname____doc__r' classmethodr9r> __classcell__rs@r r r =s!#'")-(G(G(G(G(G(GT)-#' [87777777r r c@eZdZdZdZdZdZdZdZdZ dZ d Z d S) _TimeoutzRepresents a timeout)deadlinecallbackc|dkrtd|t|std|||_||_dS)z :param float deadline: timer expiration as non-negative epoch number :param callable callback: callback to call when timeout expires :raises ValueError, TypeError: rz4deadline must be non-negative epoch number, but got %callback must be a callable, but got N)r1callable TypeErrorrHrI)r(rHrIs r r'z_Timeout.__init__sq a<<* !! I)=EXGII I!   r cZt|tr|j|jkStSz#NOTE: not supporting sort stabilityr#rGrHNotImplementedr(others r __eq__z_Timeout.__eq__) eX & & 3=EN2 2r cR||}|tur| StSrO)rTrQ)r(rSresults r __ne__z_Timeout.__ne__s,U##  ' ': r cZt|tr|j|jkStSrOrPrRs r __lt__z_Timeout.__lt__) eX & & 2=5>1 1r cZt|tr|j|jkStSrOrPrRs r __gt__z_Timeout.__gt__r[r cZt|tr|j|jkStSrOrPrRs r __le__z_Timeout.__le__rUr cZt|tr|j|jkStSrOrPrRs r __ge__z_Timeout.__ge__rUr N) r?r@rArB __slots__r'rTrXrZr]r_rarr r rGrGsI !!!&    r rGc:eZdZdZdZdZdZdZdZdZ dZ d S) _Timerz!Manage timeouts for use in ioloopic"g|_d|_dSr) _timeout_heap_num_cancellationsr=s r r'z_Timer.__init__s#$r cH|j|jD] }d|_ d|_dSdS)zTRelease resources. Don't use the `_Timer` instance after closing it N)rfrIr(timeouts r closez _Timer.closes@   )- ( (#'  !%D    * )r cJ|jtd|dkrtd|tj}t ||z|}t j|j|t d||j |j |||S)aSchedule a one-shot timeout given delay seconds. NOTE: you may cancel the timer before dispatch of the callback. Timer Manager cancels the timer upon dispatch of the callback. :param float delay: Non-negative number of seconds from now until expiration :param callable callback: The callback method, having the signature `callback()` :rtype: _Timeout :raises ValueError, TypeError NzTimeout closed before callrz0call_later: delay must be non-negative, but got zOcall_later: added timeout %r with deadline=%r and callback=%r; now=%s; delay=%s) rfr1pikacompattime_nowrGheapqheappushLOGGERdebugrHrI)r(delayrInowrjs r call_laterz_Timer.call_laters   %9:: : 199*HMOQQ Qk""$$3;11 t)7333  ,-4g6F  c5 * * * r c|jtd|dStd||j|jd|_|xjdz c_dS)zRCancel the timeout :param _Timeout timeout: The timer to cancel Nz8remove_timeout: timeout was already removed or called %rzDremove_timeout: removing timeout %r with deadline=%r and callback=%r)rIrrrsrHrgris r remove_timeoutz_Timer.remove_timeouts   # LLJ      LL"#*G,N P P P $G   # #q ( # # # #r c|jrBtj}t d|jdj|z }nd}|S)zGet the interval to the next timeout expiration :returns: non-negative number of seconds until next timer expiration; None if there are no timers :rtype: float rN)rfrmrnromaxrH)r(ruintervals r get_remaining_intervalz_Timer.get_remaining_interval"sM   +&&((C1d03z+_Timer.process_timeouts..Ss&&&&QZ5KA5K5K5Kr ) rfrmrnrorHrpheappoprIappendrg_GC_CANCELLATION_THRESHOLDlenheapify)r(ruready_timeoutsrjs r process_timeoutsz_Timer.process_timeouts2s   2+&&((C N$ 1);A)>)G3)N)N-(:;;#/"))'2222++q0++ $ 1);A)>)G3)N)N* ( (#+++q0++  """#'  '4+JJJ+s43E/F/F!/KLL*+'&&#1&&&" d011111? 2 22KJLLr N) r?r@rArBrr'rkrvryr}rrr r rdrds}++"&$$$&&&!!!F)))( $2$2$2$2$2r rdc`eZdZdZeeddZeeddZeeddZdS) PollEventszEvent flags for I/OPOLLINrxPOLLOUTPOLLERRN) r?r@rArBgetattrselectREADWRITEERRORrr r rrYsO 768T * *D GFIt , ,E GFIt , ,EEEr rceZdZdZejZejZejZdZdZ e dZ dZ dZ dZeZdZd Zd Zd Zd Zd ZdZdZdZdZdS)r%zI/O loop implementation that picks a suitable poller (`select`, `poll`, `epoll`, `kqueue`) to use based on platform. Implements the `pika.adapters.utils.selector_ioloop_adapter.AbstractSelectorIOLoop` interface. ct|_tj|_||j|j|_dSr) rd_timer collectionsdeque _callbacks _get_poller_get_remaining_intervalr_pollerr=s r r'zIOLoop.__init__psEhh &+--''(D(,(=?? r c|j;|j|jg|_dSdS)aRelease IOLoop's resources. `IOLoop.close` is intended to be called by the application or test code only after `IOLoop.start()` returns. After calling `close()`, no other interaction with the closed instance of `IOLoop` should be performed. N)rrrkrr=s r rkz IOLoop.closeysI ? & L   K     !DOOO ' &r cd}t||}ttdr8tr tdkr&tdt d i|}|sMttdr8tr tdkr&tdtd i|}|snttdrYttjdr8tr tdkr&td td i|}|s&td td i|}|S) aDetermine the best poller to use for this environment and instantiate it. :param get_wait_seconds: Function for getting the maximum number of seconds to wait for IO for use by the poller :param process_timeouts: Function for processing timeouts for use by the poller :returns: The instantiated poller instance supporting `_PollerBase` API :rtype: object N)get_wait_secondsrepollzUsing EPollPollerkqueuezUsing KQueuePollerpollmodifyzUsing PollPollerzUsing SelectPollerr) dicthasattrr SELECT_TYPErrrs EPollPoller KQueuePollerr PollPoller SelectPoller)rrpollerkwargss r rzIOLoop._get_pollers\--/// 67 # # / /+"8"8 0111$..v.. 0'&(33 0 0+"9"9 1222%//// .76622 . x00 . .+"7"7 /000#--f-- , LL- . . .!++F++F r c8|j||S)aAdd 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 )rrv)r(rtrIs r rvzIOLoop.call_laters{%%eX666r c:|j|dS)zVRemove a timeout :param timeout_handle: Handle of timeout to remove N)rry)r(timeout_handles r ryzIOLoop.remove_timeouts "">22222r ct|std||j||jt d|dS)aRequests 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 rKz*add_callback_threadsafe: added callback=%rN)rLrMrrrwake_threadsaferrrs)r(rIs r add_callback_threadsafezIOLoop.add_callback_threadsafes}!! I)=EXGII I x((( $$&&& A8LLLLLr ctjt|jD]@}|j}t d||A|j dS)z|[Extension] Process pending callbacks and timeouts, invoking those whose time has come. Internal use only. z&process_timeouts: invoking callback=%rN) rmrnxrangerrpopleftrrrsrr)r(_rIs r rzIOLoop.process_timeoutss| ##C$8$899  A..00H LLA8 L L L HJJJJ $$&&&&&r cF|jrdS|jS)aGet the remaining interval to the next callback or timeout expiration. :returns: non-negative number of seconds until next callback or timer expiration; None if there are no callbacks and timers :rtype: float r)rrr}r=s r rzIOLoop._get_remaining_intervals' ? 1{11333r c>|j|||dS)aStart 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)r add_handler)r(fdhandlereventss r rzIOLoop.add_handlers$   Wf55555r c<|j||dS)zChanges the events we watch for :param int fd: The file descriptor :param int events: The event mask using READ, WRITE, ERROR N)rupdate_handler)r(rrs r rzIOLoop.update_handler s" ##B/////r c:|j|dS)zhStop watching the given file descriptor for events :param int fd: The file descriptor N)rremove_handler)r(rs r rzIOLoop.remove_handlers ##B'''''r c8|jdS)zl[API] Start the main poller loop. It will loop until requested to exit. See `IOLoop.stop`. N)rstartr=s r rz IOLoop.starts r c8|jdS)a6[API] 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_threadsafe(ioloop.stop)` N)rstopr=s r rz IOLoop.stop$s r c8|jdS)z)[Extension] Activate the poller N)ractivate_pollerr=s r rzIOLoop.activate_poller0s $$&&&&&r c8|jdS)z+[Extension] Deactivate the poller N)rdeactivate_pollerr=s r rzIOLoop.deactivate_poller6s &&(((((r c8|jdS)a [Extension] Wait for events of interest on registered file descriptors until an event of interest occurs or next timer deadline or `_PollerBase._MAX_POLL_TIMEOUT`, whichever is sooner, and dispatch the corresponding event handlers. N)rrr=s r rz IOLoop.poll<s r N)r?r@rArBrrrrr'rk staticmethodrrvryr add_callbackrrrrrrrrrrrr r r%r%bs> ?D  E  E???!!! ''\'R 7 7 7333MMM6+L ' ' ' 4 4 4 6 6 6000(((   ''' ))) r r%cDeZdZdZdZdZdZdZdZdZ dZ d Z d Z d Z d Zd ZdZdZejdZejdZejdZejdZejdZejdZdZedZdZdS) _PollerBasez2Base class for select-based IOLoop implementationsrxc||_||_tj|_t |_tjttj ttj ti|_ i|_ d|_d|_|\|_|_||j|jtjdS)a :param get_wait_seconds: Function for getting the maximum number of seconds to wait for IO for use by the poller :param process_timeouts: Function for processing timeouts for use by the poller FN)_get_wait_seconds_process_timeouts threadingLock _waking_mutexr _fd_handlersrrsetrr _fd_events_processing_fd_event_map_running _stopping_get_interrupt_pair _r_interrupt _w_interruptrfileno_read_interrupt)r(rrs r r'z_PollerBase.__init__Qs"2!1'^--!FF OSUU  cee  cee  )+% 04/G/G/I/I,4, *1133T5I# * * * * *r c|jr Jd|j5|jl||j|jd|_|jd|_dddn #1swxYwY|d|_d|_ d|_ dS)zRelease poller's resources. `close()` is intended to be called after the poller's `start()` method returns. After calling `close()`, no other interaction with the closed poller instance should be performed. z+Cannot call close() before start() unwinds.N) rrrrrrrkrrrrr=s r rkz_PollerBase.closevs=OO"OOOO   ) ) ,##D$5$<$<$>$>???!'')))$(!!'')))$(!  ) ) ) ) ) ) ) ) ) ) ) ) ) ) )     (,%%%sA4BB Bch|j5|j ddddS |jdn`#tjj$r }|jt jkrYd}~n1d}~wt$r!}t d|d}~wwxYwddddS#1swxYwYdS)zlWake up the poller as soon as possible. As the name indicates, this method is thread-safe. NXz&Failed to send interrupt to poller: %s) rrsendrmrn SOCKET_ERRORr EWOULDBLOCK Exceptionrrwarning)r(errs r rz_PollerBase.wake_threadsafes>     (         !&&t,,,,;+   9 11121111   GMMM                    sC B':B'BA)$B') B6BBB''B+.B+c||}||j}nt||j}||jzS)zGet the interval to the next timeout event, or a default interval :returns: maximum number of self.POLL_TIMEOUT_MULT-scaled time units to wait for IO events :rtype: int )r_MAX_POLL_TIMEOUTminPOLL_TIMEOUT_MULT)r(rts r _get_max_waitz_PollerBase._get_max_waitsD&&(( =*EEt566Et---r cr||j|<||||||dS)zAdd a new fileno to the set to be monitored :param int fileno: The file descriptor :param callable handler: What is called when an event happens :param int events: The event mask using READ, WRITE, ERROR N)r_set_handler_events _register_fd)r(rrrs r rz_PollerBase.add_handlersE%,&!   000 &&)))))r cj|||\}}|||||dS)zSet the events to the current events :param int fileno: The file descriptor :param int events: The event mask using READ, WRITE, ERROR revents_to_clear events_to_setN)r_modify_fd_events)r(rrevents_cleared events_sets r rz_PollerBase.update_handlersS&*%=%=ff%M%M"   *$  & & & & &r c |j|=n#t$rYnwxYw||d\}}|j|=|||dS)z_Remove a file descriptor from the set :param int fileno: The file descriptor r)rN)rKeyErrorrr_unregister_fd)r(rrrs r rz_PollerBase.remove_handlers}  -f55    D !44VQ??  f % FNCCCCCs  c:d}d}tjtjtjfD]p}||zr5||j|vr%|j||||z}<||j|vr%|j||||z}q||fS)aSet the handler's events to the given events; internal to `_PollerBase`. :param int fileno: The file descriptor :param int events: The event mask (READ, WRITE, ERROR) :returns: a 2-tuple (events_cleared, events_set) :rtype: tuple r)rrrrradddiscard)r(rrrrevts r rz_PollerBase._set_handler_eventss OZ%5z7GH * *C| *!555OC(,,V444#%JT_S111OC(00888"c)Nz))r c(|tjt}|jD]\}}|D]}||xx|zcc<|D]\}}|||dS)zActivate the poller N) _init_pollerr defaultdictintritemsr)r( fd_to_eventseventfile_descriptorsrrs r rz_PollerBase.activate_pollers ".s33 '+'<'<'>'> . . #E#* . .V$$$-$$$$ .+0022 . .NFF   ff - - - - . .r c.|dS)zDeactivate the poller N)_uninit_pollerr=s r rz_PollerBase.deactivate_pollers r cV|jrtdtdd|_| |js/|||j/ td|d|_d|_dS#d|_d|_wxYw# td|d|_d|_n#d|_d|_wxYwwxYw)zStart the main poller loop. It will loop until requested to exit. This method is not reentrant and will raise an error if called recursively (pika/pika#1095) :raises: RuntimeError z.IOLoop is not reentrant and is already runningzEntering IOLoopTzDeactivating pollerFN) r RuntimeErrorrrrsrrrrrr=s r rz_PollerBase.startsA = QOPP P &'''   &n ) &&(((n ) & 2333&&(((!& % "' % %%%%  & 2333&&(((!& % "' % %%%%%%%%s0 6C.CCD(.DD(D$$D(cptdd|_|dS)znRequest exit from the ioloop. The loop is NOT guaranteed to stop before this method returns. zStopping IOLoopTN)rrrsrrr=s r rz_PollerBase.stop0s6  &''' r ct)z7Wait for events on interested filedescriptors. NotImplementedErrorr=s r rz_PollerBase.poll9s "!r ct)9Notify the implementation to allocate the poller resourcerr=s r rz_PollerBase._init_poller? "!r ct)8Notify the implementation to release the poller resourcerr=s r r z_PollerBase._uninit_pollerDrr ct)a<The base class invokes this method to notify the implementation to register the file descriptor with the polling object. The request must be ignored if the poller is not activated. :param int fileno: The file descriptor :param int events: The event mask (READ, WRITE, ERROR) rr(rrs r rz_PollerBase._register_fdI "!r ct)The base class invoikes this method to notify the implementation to modify an already registered file descriptor. The request must be ignored if the poller is not activated. :param int fileno: The file descriptor :param int events: absolute events (READ, WRITE, ERROR) :param int events_to_clear: The events to clear (READ, WRITE, ERROR) :param int events_to_set: The events to set (READ, WRITE, ERROR) rr(rrrrs r rz_PollerBase._modify_fd_eventsTs "!r ct)XThe base class invokes this method to notify the implementation to unregister the file descriptor being tracked by the polling object. The request must be ignored if the poller is not activated. :param int fileno: The file descriptor :param int events_to_clear: The events to clear (READ, WRITE, ERROR) rr(rrs r rz_PollerBase._unregister_fdarr cP|j||_tj|D]e}||vr||}t jt jt jfD]}||j |vr||z}|r|j |}|||fdS)a Helper to dispatch callbacks for file descriptors that received events. Before doing so we re-calculate the event mask based on what is currently set in case it has been changed under our feet by a previous callback. We also take a store a refernce to the fd_event_map so that we can detect removal of an fileno during processing of another callback and not generate spurious callbacks on it. :param dict fd_event_map: Map of fds to events received on them. N) rclearrmrndictkeysrrrrrr)r( fd_event_maprrrrs r _dispatch_fd_eventsz_PollerBase._dispatch_fd_eventsls %++---(4%k**<88 ( (F\))!&)F")9:;KL # #!555sdNF (+F3''' ( (r c>tjS)a& Use a socketpair to be able to interrupt the ioloop if called from another thread. Socketpair() is not supported on some OS (Win) so use a pair of simple TCP sockets instead. The sockets will be closed and garbage collected by python when the ioloop itself is. )rmrn_nonblocking_socketpairrr r rz_PollerBase._get_interrupt_pairs{22444r c |jddS#tjj$r!}|jt jkrYd}~dSd}~wwxYw)a Read the interrupt byte(s). We ignore the event mask as we can ony get here if there's data to be read on our fd. :param int _interrupt_fd: (unused) The file descriptor to read from :param int _events: (unused) The events generated for this fd iN)rrecvrmrnrrEAGAIN)r( _interrupt_fd_eventsrs r rz_PollerBase._read_interruptsn    " "3 ' ' ' ' '{'   yEL(()((((( sAAAN)r?r@rArBrrr'rkrrrrrrrrrrabcabstractmethodrrr rrrr#rrrrr r rrFs<< #*#*#*J---6,... * * *&&&"DDD"***2 . . . &&&8 """  """ """ """  " " " """(((B55\5     r rc:eZdZdZdZdZdZdZdZdZ dZ d S) rzDefault behavior is to use Select since it's the widest supported and has all of the methods we need for child classes as well. One should only need to override the update_handler and start methods for additional types. rxc |jtjs.|jtjs|jtjrmt j|jtj|jtj|jtj|\}}}n,tj|ggg}}}n'#t$r}t|rYd}~d}~wwxYwtj t}t|||ftjtjtjfD]\}}|D]}||xx|zcc<||dSWait for events of interest on registered file descriptors until an event of interest occurs or next timer deadline or _MAX_POLL_TIMEOUT, whichever is sooner, and dispatch the corresponding event handlers. TN)rrrrrrrtimesleep_SELECT_ERRORSrrrrzipr#)r(readwriteerrorr"fd_setrrs r rzSelectPoller.polls   OJO4 4 (89 4 (89 4*0 8 (89 (894;M;M;O;O*Q*Q&D%Jt1133444)+R%D!    ''HHHH  #.s33 ue$*"2J4DEGG , ,KFC! , ,V$$$+$$$$ ,   .....sCC"" D,DDDcdSrNrr=s r rzSelectPoller._init_pollerr cdSrNrr=s r r zSelectPoller._uninit_pollerr;r cdS@The base class invokes this method to notify the implementation to register the file descriptor with the polling object. The request must be ignored if the poller is not activated. :param int fileno: The file descriptor :param int events: The event mask using READ, WRITE, ERROR Nrrs r rzSelectPoller._register_fdr;r cdSrNrrs r rzSelectPoller._modify_fd_eventsr;r cdSrNrrs r rzSelectPoller._unregister_fdr;r N) r?r@rArBrrrr rrrrr r rrs $/$/$/LHHHGGG            r rc\eZdZdZfdZedZdZdZdZ dZ dZ d Z xZ S) rzAKQueuePoller works on BSD based systems and is faster than selectchd|_tt|||dS)z/Create an instance of the KQueuePoller N)_kqueuer&rr'r(rrrs r r'zKQueuePoller.__init__s4  lD!!**+;=MNNNNNr cPd}|jtjkr tj}n|jtjkr0tj}|jtjzr|tj z}n<|jtj zr tj }nt d||S)zreturn the event type associated with a kevent object :param kevent kevent: a kevent object as returned by kqueue.control() rzUnexpected kevent: %s) filterrKQ_FILTER_READrrKQ_FILTER_WRITErflags KQ_EV_EOFr KQ_EV_ERRORrrcritical)keventmasks r _map_eventzKQueuePoller._map_events =F1 1 1?DD ]f4 4 4#D|f.. ) (( \F. . =#DD OO3V < < < r cl |jdd|}n'#t$r}t |rYd}~Od}~wwxYwt jt}|D]*}||jxx| |zcc<+| |dS)r0TN) rGcontrolrr3rrrridentrSr#)r(keventsr7r"rs r rzKQueuePoller.poll s   ,..tT4;M;M;O;OPP!    ''HHHH   #.s33  @ @E  % % %)?)? ? % % % %   .....s.2 AAAAcH|jJtj|_dSr:)rGrrr=s r rzKQueuePoller._init_poller6s!|###} r cX|j"|jd|_dSdSr=)rGrkr=s r r zKQueuePoller._uninit_poller<s1 < # L   DLLL $ #r c8|||d|dS)r@rrNrrs r rzKQueuePoller._register_fdBs9  61F  L L L L Lr c|jdSt}|tjzr>|t j|t jt j|tjzr>|t j|t jt j |tj zr>|t j|t j t j|tj zr>|t j|t j t j |j |ddS)rN)rJrMr) rGlistrrrrrQrK KQ_EV_DELETE KQ_EV_ADDrrLrV)r(rrrrrXs r rzKQueuePoller._modify_fd_eventsMsb <  F&& Z_ , 0 NN !0 -/// 0 0 0 :? * - NN !0 *,,, - - - Z- - 0 NN !1 -/// 0 0 0 :+ + - NN !1 *,,, - - - Wa(((((r c8||d|ddS)rrrNr\rs r rzKQueuePoller._unregister_fdws9  1oQ  P P P P Pr )r?r@rArBr'rrSrrr rrrrDrEs@r rrsLKOOOOO \.///,'''    L L L()()()T P P P P P P Pr rc`eZdZdZdZfdZedZdZdZ dZ dZ d Z d Z xZS) rz|Poll works on Linux and can have better performance than EPoll in certain scenarios. Both are faster than select. rUchd|_tt|||dS)z0Create an instance of the KQueuePoller N)_pollr&rr'rHs r r'zPollPoller.__init__s4  j$(()9;KLLLLLr c(tjSz' :rtype: `select.poll` )rrrr r _create_pollerzPollPoller._create_pollers {}}r c |j|}n'#t$r}t |rYd}~Md}~wwxYwt jt}|D]D\}}|tj zr tj j r|tj z}||xx|zcc<E||dSr/)rdrrr3rrrrrPOLLHUPrmrnON_OSXrr#)r(rr7r"rrs r rzPollPoller.polls   ););)=)=>>!    ''HHHH   #.s33 # * *MFE & (DK,> ('  E )    .....s,0 AAAAcJ|jJ||_dSr:)rdrgr=s r rzPollPoller._init_pollers'z!!!((** r c|j7t|jdr|jd|_dSdS)rNrk)rdrrkr=s r r zPollPoller._uninit_pollersE : !tz7++ #   """DJJJ " !r cN|j|j||dSdSr?)rdregisterrs r rzPollPoller._register_fds2 : ! J   / / / / / " !r cN|j|j||dSdSrB)rdrrs r rzPollPoller._modify_fd_eventss2 : ! J  ff - - - - - " !r cL|j|j|dSdSrD)rd unregisterrs r rzPollPoller._unregister_fds0 : ! J ! !& ) ) ) ) ) " !r )r?r@rArBrr'rrgrrr rrrrDrEs@r rrsMMMMM\ ///:+++  0 0 0 . . . * * * * * * *r rc,eZdZdZdZedZdS)rz{EPoll works on Linux and can have better performance than Poll in certain scenarios. Both are faster than select. rxc(tjSrf)rrrr r rgzEPollPoller._create_pollers |~~r N)r?r@rArBrrrgrr r rrs?\r r)-rBr+rrrploggingrr1r pika.compatrmpika.adapters.utilsrpika.adapters.base_connectionr+pika.adapters.utils.selector_ioloop_adapterrr getLoggerr?rrrrrnPY3InterruptedErrorr7IOErrorOSErrortuplekeysr3rr objectrGrdrr% AbstractBaserrrrrrr r rsK   ......88888877777777  8 $ $ ;?>0>~+,'I'Iv|$"B"Bw"B"Bw-224455   R7R7R7R7R7~R7R7R7j>>>>>v>>>BC2C2C2C2C2VC2C2C2L--------aaaaa #aaaH\\\\\$+*\\\~ T T T T T ;T T T pCPCPCPCPCP;CPCPCPLa*a*a*a*a*a*a*a*H     *     r