U
vi: @ s d dl mZmZmZmZ d dlmZmZm Z d dlm Z d dl
m Z edZ d
e j
eej eee gee f ddd Zd gZdS ) )AnyCallableOptionalTypeVar)
Observableabctyping)CompositeDisposable)TimeoutScheduler_TN)duration schedulerreturnc s" t t t t d fdd }|S )N)sourcer c s2 dt jt tt j t jd fdd
}t|S )a Takes elements for the specified duration from the start of
the observable source sequence.
Example:
>>> res = take_with_time(source)
This operator accumulates a queue with a length enough to store
elements received during the initial duration window. As more
elements are received, elements older than the specified
duration are taken from the queue and produced on the result
sequence. This causes elements to be delayed with duration.
Args:
source: Source observable to take elements from.
Returns:
An observable sequence with the elements taken during the
specified duration from the start of the source sequence.
N)observer
scheduler_r c sH p|pt }dtjtd fdd
}||}t|j |dS )Nr
statec s d S )N)Z on_completedr r R/opt/alt/python38/lib/python3.8/site-packages/reactivex/operators/_takewithtime.pyaction( s zJtake_with_time_..take_with_time..subscribe..action)r
)N)r
Z singletonr
SchedulerBaser Zschedule_relativer subscribe)r r Z
_schedulerr Zdisp)r r
r r r r " s z:take_with_time_..take_with_time..subscribe)N)r Z ObserverBaser r r ZDisposableBaser )r r r r
)r r take_with_time
s z'take_with_time_..take_with_time)r r )r r
r r r r take_with_time_
s %r )N)r r r r r Z reactivexr r Zreactivex.disposabler Zreactivex.schedulerr
r Z RelativeTimer r __all__r r r r s +