U bh.@s&UdZddlmZddlZddlZddlmZddlmZddlm Z ddl m Z ddl m Zdd lmZmZdd lmZdd lmZdd lmZmZmZmZdd lmZddlmZdZee Z!da"de#d<dZ$Gddde%Z&ddZ'eddZ(ddddZ)dddddZ*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_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__Ns  z Hub.__init__cCs|js||jSr)_pollerr5r6rrrpollerisz Hub.pollercCs ||_dSr)r8)r6valuerrrr:oscCs||dSr)closer5r9rrrresetssz Hub.resetcCs t|_|jj|_|jj|_dSr)rr8register _register_fd unregister_unregister_fdr9rrrr5ws zHub._create_pollercCs*|jdk r&|jd|_d|_d|_dSr)r8r<r?rAr9rrr _close_poller|s   zHub._close_pollercCs|tdSr) call_soonrr9rrrstopszHub.stopcCsdt|t|jt|jS)Nz)formatidlenr'r(r9rrr__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) r6Z min_delayZ max_delayZ max_timers propagater&delayientryexcrrr fire_timerss(     $zHub.fire_timersc Cs z||W5||XdSr)_discard _unregisterr6fdrrr_remove_from_loopszHub._remove_from_loopFcCszt|}z|j||Wn tk r:||Yn ValueErrorrbr r'r(r2add)r6racallbackflagsr r2destrrrrds   zHub.addcCst|}||dSr)r rbr`rrrremovesz Hub.removecCs@d|_z,z |Wqtk r,Yq0YqXqW5d|_XdS)NTF)r0run_oncerr9rrr run_forevers zHub.run_forevercCs.zt|jWntk r(d|_YnXdSr)rNr% StopIterationr1r9rrrrisz Hub.run_oncec Gs6t|tst||}|j|j|W5QRX|Sr) isinstancerr r/r,rd)r6rer rrrrCs   z Hub.call_sooncGs|j|||Sr)r&Z call_afterr6rYrer rrr call_laterszHub.call_latercGs|j|||Sr)r&call_at)r6whenrer rrrrosz Hub.call_atcGs|j|||Sr)r&call_repeatedlyrmrrrrqszHub.call_repeatedlycGs|||ttB|Sr)rdr r r6Zfdsrer rrr add_readerszHub.add_readercGs|||t|Sr)rdrrrrrr add_writerszHub.add_writerc CsH||jk}|j|}z||W5|rB|\}}|||t|XdSr)r(getrdrrb)r6rawritableZon_writecbr rrr remove_readers  zHub.remove_readerc CsL||jk}|j|}z||W5|rF|\}}|||ttB|XdSr)r'rurdr r rb)r6rareadableZon_readrwr rrr remove_writers  zHub.remove_writerc Cs0z|j|Wntttfk r*YnXdSr)r:r@AttributeErrorKeyErrorrRr`rrrr_szHub._unregisterc Cs0|j |j}t|_|W5QRSQRXdSr)r/r,r))r6readyrrr _pop_readyszHub._pop_readycs~fddjDjfddjDjjjD] }|qT}|D] }|qndS)Ncsg|]}|qSrr_.0rar9rr szHub.close..csg|]}|qSrrrr9rrrs)r'clearr(r2rBr+r~)r6r reZtodositemrr9rr<s     z Hub.closecCs4t|}|j|d|j|d|j|dSr)r r'popr(r2discardr`rrrr^sz Hub._discardcCstjd||dddS)Nz Callback %r raised exception: %rrrJ)rTrU)r6rer\rrron_callback_errors zHub.on_callback_errorc cs|j|j} } |jj} |j}|j}|jj}|j}|j }|j }| }|D]}|rH|qH|rf||dnd}t |j D] }|qt| s| rg}z | |}Wntk 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|||ddVqs2