1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
|
from datetime import datetime
from typing import Any, Callable, Optional, TypeVar
from reactivex import Observable, abc, typing
from reactivex.disposable import CompositeDisposable
from reactivex.scheduler import TimeoutScheduler
_T = TypeVar("_T")
def take_until_with_time_(
end_time: typing.AbsoluteOrRelativeTime,
scheduler: Optional[abc.SchedulerBase] = None,
) -> Callable[[Observable[_T]], Observable[_T]]:
def take_until_with_time(source: Observable[_T]) -> Observable[_T]:
"""Takes elements for the specified duration until the specified end
time, using the specified scheduler to run timers.
Examples:
>>> res = take_until_with_time(source)
Args:
source: Source observale to take elements from.
Returns:
An observable sequence with the elements taken
until the specified end time.
"""
def subscribe(
observer: abc.ObserverBase[_T],
scheduler_: Optional[abc.SchedulerBase] = None,
) -> abc.DisposableBase:
_scheduler = scheduler or scheduler_ or TimeoutScheduler.singleton()
def action(scheduler: abc.SchedulerBase, state: Any = None):
observer.on_completed()
if isinstance(end_time, datetime):
task = _scheduler.schedule_absolute(end_time, action)
else:
task = _scheduler.schedule_relative(end_time, action)
return CompositeDisposable(
task, source.subscribe(observer, scheduler=scheduler_)
)
return Observable(subscribe)
return take_until_with_time
__all__ = ["take_until_with_time_"]
|
