o h.@sUdZddlmZddlZddlZddlmZddlmZddlm Z ddl m Z ddl m Zdd lmZmZdd lmZdd lmZdd lmZmZmZmZdd lmZddlmZdZee Z!da"de#d<dZ$Gddde%Z&ddZ'eddZ(d"ddZ)d#ddZ*Gd d!d!Z+dS)$zEvent loop implementation.) annotationsN)contextmanager)copy)Empty)sleep) GeneratorType)Thenablepromise) get_logger)fileno)ERRREADWRITEpoll)cached_property)Timer)Hubget_event_loopset_event_loop 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|dur|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__Ns   z Hub.__init__cCs|js||jSr)_pollerr6r7rrrpollerisz Hub.pollercCs ||_dSr)r9)r7valuerrrr;o cCs||dSr)closer6r:rrrresetss z Hub.resetcCs t|_|jj|_|jj|_dSr)rr9register _register_fd unregister_unregister_fdr:rrrr6ws zHub._create_pollercCs.|jdur|jd|_d|_d|_dSdSr)r9r>rArCr:rrr _close_poller|s   zHub._close_pollercCs|tdSr) call_soonr r:rrrstopszHub.stopcCsdt|t|jt|jS)Nz)formatidlenr(r)r:rrr__repr__sz Hub.__repr__r rc Cs|j}d}|rm|jrmt|D]^}t|j\}}|durnPz|Wq|y*ttfy3tyS} z| jtj krAt j d| ddWYd} ~ qd} ~ wt yl} zt j d| ddWYd} ~ qd} ~ wwt |pq||S)NzError in timer: %rrexc_info)r'_queuerangenext scheduler MemoryErrorAssertionErrorOSErrorerrnoENOMEMloggererror Exceptionmin) r7 min_delay max_delay max_timers propagater'delayientryexcrrr fire_timerss.     zHub.fire_timersc Cs(z ||W||dS||wr) _unregister_discardr7fdrrr_remove_from_loops zHub._remove_from_loopFcCsvt|}z |j||Wn ty||w|t@r"|jn|j}|r3|j |d||<dS||f||<dSr) r r;r@ ValueErrorrhr r(r)r3add)r7rgcallbackflagsr!r3destrrrrjs    zHub.addcCst|}||dSr)r rhrfrrrremovesz Hub.removecCs@d|_z z|Wn tyYnwqWd|_dSd|_w)NTrF)r1run_oncerr:rrr run_forevers  zHub.run_forevercCs,zt|jWdStyd|_YdSwr)rPr& StopIterationr2r:rrrros   z Hub.run_oncecGsLt|ts t||}|j|j|Wd|S1swY|Sr) isinstancerr r0r-rj)r7rkr!rrrrEs   z Hub.call_sooncG|j|||Sr)r' call_afterr7r_rkr!rrr call_laterzHub.call_latercGrsr)r'call_at)r7whenrkr!rrrrxrwz Hub.call_atcGrsr)r'call_repeatedlyrurrrrzrwzHub.call_repeatedlycGs|||ttB|Sr)rjr r r7fdsrkr!rrr add_readerszHub.add_readercGs|||t|Sr)rjrr{rrr add_writerrwzHub.add_writerc Csh||jv}|j|}z||W|r"|\}}|||t|dSdS|r3|\}}|||t|wwr)r)getrhrjr)r7rgwritableon_writecbr!rrr remove_readers   zHub.remove_readerc Csp||jv}|j|}z||W|r$|\}}|||ttB|dSdS|r7|\}}|||ttB|wwr)r(rrhrjr r )r7rgreadableon_readrr!rrr remove_writers   zHub.remove_writerc Cs.z |j|WdStttfyYdSwr)r;rBAttributeErrorKeyErrorrTrfrrrrds zHub._unregistercCs<|j|j}t|_|WdS1swYdSr)r0r-r*)r7readyrrr _pop_readys $zHub._pop_readycs~fddjDjfddjDjjjD]}|q*}|D]}|q7dS)Ncg|]}|qSrrd.0rgr:rr zHub.close..crrrrr:rrrr)r(clearr)r3rDr,r)r7r!rktodositemrr:rr>s     z Hub.closecCs4t|}|j|d|j|d|j|dSr)r r(popr)r3discardrfrrrresz Hub._discardcCstjd||dddS)Nz Callback %r raised exception: %rrrL)rWrX)r7rkrbrrron_callback_errors zHub.on_callback_errorc cs|j|j} } |jj} |j}|j}|jj}|j}|j }|j } | }|D]}|r-|q&|r5||dnd}t |j D]}|q<| sG| rrerrrrZrPrrqrrrrrrQr&rrrrr9s^         _ r)r$r)r&rr$r),r __future__rrUr. contextlibrrqueuertimertypesrrvinerr kombu.logr kombu.utils.compatr kombu.utils.eventior r rrkombu.utils.objectsrr'r__all__rrWr__annotations__r BaseExceptionrr r#rrrrrrrs4