U .e.@sUdZddlmZddlZddlZddlmZddlmZddl m Z ddl m Z ddlmZmZdd lmZdd lmZdd lmZmZmZmZdd lmZd dlmZdZeeZ da!de"d<dZ#Gddde$Z%ddZ&eddZ'ddddZ(dddddZ)Gdd d Z*dS)!zEvent loop implementation.) annotationsN)contextmanager)Empty)sleep) GeneratorType)Thenablepromise) get_logger)fileno)ERRREADWRITEpoll)cached_property)Timer)Hubget_event_loopset_event_loopz Hub | None _current_loopzd?Z,d@dAZ-e.dBdCZ/e dDdEZ0dS)Irz~Event loop object. Arguments: --------- timer (kombu.asynchronous.Timer): Specify custom timer instance. NcCsn|dk r |nt|_i|_i|_t|_t|_t|_t |_ d|_ d|_ t|_ d|_d|_|dS)NFr)rtimerreaderswritersseton_tickon_close_ready threadingLock _ready_lock_running_loop consolidateconsolidate_callbackpropagate_errors_create_poller)selfr%rrr__init__Ms  z Hub.__init__cCs|js||jSr)_pollerr4r5rrrpollerhsz Hub.pollercCs ||_dSr)r7)r5valuerrrr9nscCs||dSr)closer4r8rrrresetrsz Hub.resetcCs t|_|jj|_|jj|_dSr)rr7register _register_fd unregister_unregister_fdr8rrrr4vs zHub._create_pollercCs*|jdk r&|jd|_d|_d|_dSr)r7r;r>r@r8rrr _close_poller{s   zHub._close_pollercCs|tdSr) call_soonrr8rrrstopszHub.stopcCsdt|t|jt|jS)Nz)formatidlenr&r'r8rrr__repr__s z Hub.__repr__r rc Cs|j}d}|r|jrt|D]}t|j\}}|dkr:qz |Wq|k rZYqttfk rrYqtk r} z"| jtj krt j d| ddW5d} ~ XYqt k r} zt j d| ddW5d} ~ XYqXqt |p||S)NzError in timer: %rrexc_info)r%_queuerangenext scheduler MemoryErrorAssertionErrorOSErrorerrnoZENOMEMloggererror Exceptionmin) r5Z min_delayZ max_delayZ max_timers propagater%delayientryexcrrr fire_timerss(     $zHub.fire_timersc Cs z||W5||XdSr)_discard _unregisterr5fdrrr_remove_from_loopszHub._remove_from_loopFcCszt|}z|j||Wn tk r:||YnszHub.close..csg|]}|qSrr~rr8rrrs)r&clearr'r1rAr*r})r5rrdZtodositemrr8rr;s     z Hub.closecCs4t|}|j|d|j|d|j|dSr)r r&popr'r1discardr_rrrr]sz Hub._discardcCstjd||dddS)Nz Callback %r raised exception: %rrrI)rSrT)r5rdr[rrron_callback_errors zHub.on_callback_errorc ! cs|j|j} } |jj} |j}|j}|jj}|j}|j }|j }|j }| }|D]}|rN|qN|rl||dnd}|D] }|qt| s| rg}z | |}Wnt k rYdSX|pdD]\}}d}||kr| |dkr||qd}}||@r4z| |\}}Wn$|k r0||YqYnXnd|| @rvz| |\}}Wn$|k rr||YqYnXn"|| @rd}ntt||d}|rz| |p| |\}}Wntk rYnX|dkr||qt||rtz ||Wnntk r>} z| jtjkr&||W5d} ~ XYn4|k rRYn tk rp||YnXqz ||Wq|k rYqXq|r||n|||ddVqBdS)N)rWrrFTg?)r&r'r9rr\rgr%rKr1r2r)r3r}rbrtappendrwryrSinfoW_UNKNOWN_EVENT TypeErrorrkrQrREBADFrU)!r5 generatorrrVrMrrjr{r r r r&r'rr\Z hub_removeZ scheduledr1r2r)rWtodorZ poll_timeoutZ tick_callbackZto_consolidateeventsr`eventZ general_errorrvZcbargsr[rrr create_loops                    zHub.create_loopcCsddlm}||S)Nr) repr_active)debugr)r5rrrrrs zHub.repr_activecCsddlm}|||pgS)Nr) repr_events)rr)r5rrrrrrs zHub.repr_eventscCs t|jSr)iterr%r8rrrrNsz Hub.schedulercCs|jdkr||_|jSr)r0rr8rrrr$s  zHub.loop)N)rrHrHr)rF)1rrrrr r r r*r6propertyr9setterr<r4rArCrGr\rarcrgrirhrBrmrnrprrrsrwryr^r}r;r]rrrrVrMrrjr{rrrrrNr$rrrrr8sj         ` r)+r __future__rrRr, contextlibrqueuertimertypesrrZvinerrZ kombu.logr Zkombu.utils.compatr Zkombu.utils.eventior r r rZkombu.utils.objectsrr%r__all__rrSr__annotations__r BaseExceptionrrr!rrrrrrrs0