U ējF@s8ddlZddlmZGdddeZGdddeZdS)N) IBatchedTimerc@s(eZdZdZddZddZddZdS) _BatchedCallz Wraps IDelayedCall-implementing objects, implementing only the API which txaio promised in the first place: .cancel Do not create these yourself; use _BatchedTimer.call_later() cCs||_||_||_dSN)_timer_index_call)selfZtimerindexZthe_callr 1/tmp/pip-unpacked-wheel-8qem9ucn/txaio/_common.py__init__'sz_BatchedCall.__init__cCs|j|j|d|_dSr)r _remove_callrrr r r cancel-sz_BatchedCall.cancelcCs|Sr)rrr r r __call__1sz_BatchedCall.__call__N)__name__ __module__ __qualname____doc__r rrr r r r rsrc@s2eZdZdZd ddZddZddZd d ZdS) _BatchedTimera= Internal helper. Instances of this are returned from :meth:`txaio.make_batched_timer` and that is the only way they should be instantiated. You may depend on methods from the interface class only (:class:`txaio.IBatchedTimer`) **NOTE** that the times are in milliseconds in this class! NcCs>|dkrtdt||_||_||_||_t|_||_dS)Nz!bucket_milliseconds must be > 0.0) ValueErrorfloat_bucket_milliseconds _chunk_size _get_seconds_create_delayed_calldict_bucketsZ_loop)rZbucket_milliseconds chunk_sizeZseconds_providerZdelayed_call_creatorZloopr r r r As z_BatchedTimer.__init__c s|}t||d}|t||j8}t||fdd}z|j|d|WnFtk r|d|}|td||j |} | |gf|j|<YnX|S)z# IBatchedTimer API ics Srr r argsfunckwargsr r Vz*_BatchedTimer.call_later..@@r) rintrrrappendKeyErrorrmax_notify_bucket) rdelayr"r!r#now real_timecallZdiff delayed_callr r r call_laterNs z_BatchedTimer.call_latercs`j|\}}j|=gfddjttt|j}|jtd|dS)z Internal helper. This 'does' the callbacks in a particular bucket. :param real_time: the bucket to do callbacks on c s|d|D]<}z |Wq tk rF}z|W5d}~XYq Xq ||d}|rr|d|||n,trd}D]}|d|7}qt|dS)Nr'z'Error(s) processing call_later bucket: z{} ) Exceptionr)rlenformat RuntimeError)callsrZchunk_delay_msr0emsgerrorsnotify_one_chunkrr r r<ps&  z6_BatchedTimer._notify_bucket..notify_one_chunkrN)rrmathceilrr4rr+)rr/r1r7Zdelay_msr r:r r,fs z_BatchedTimer._notify_bucketcCsLz|j|\}}Wntk r(YdSX|||sH|j|=|dS)z Internal helper. Removes a (possibly still pending) call from a bucket. It is *not* an error of the bucket is gone (e.g. the call has already happened). N)rr*remover)rr/r0r1r7r r r r s z_BatchedTimer._remove_call)N)rrrrr r2r,r r r r r r5s  $r)r=Ztxaio.interfacesrobjectrrr r r r s