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
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
from datetime import datetime
from typing import Any, Optional
from reactivex import Observable, abc, typing
from reactivex.disposable import MultipleAssignmentDisposable
from reactivex.scheduler import TimeoutScheduler
from reactivex.scheduler.periodicscheduler import PeriodicScheduler
def observable_timer_date(
duetime: typing.AbsoluteTime, scheduler: Optional[abc.SchedulerBase] = None
) -> Observable[int]:
def subscribe(
observer: abc.ObserverBase[int], scheduler_: Optional[abc.SchedulerBase] = None
) -> abc.DisposableBase:
_scheduler: abc.SchedulerBase = (
scheduler or scheduler_ or TimeoutScheduler.singleton()
)
def action(scheduler: abc.SchedulerBase, state: Any) -> None:
observer.on_next(0)
observer.on_completed()
return _scheduler.schedule_absolute(duetime, action)
return Observable(subscribe)
def observable_timer_duetime_and_period(
duetime: typing.AbsoluteOrRelativeTime,
period: typing.AbsoluteOrRelativeTime,
scheduler: Optional[abc.SchedulerBase] = None,
) -> Observable[int]:
def subscribe(
observer: abc.ObserverBase[int], scheduler_: Optional[abc.SchedulerBase] = None
) -> abc.DisposableBase:
_scheduler = scheduler or scheduler_ or TimeoutScheduler.singleton()
nonlocal duetime
if not isinstance(duetime, datetime):
duetime = _scheduler.now + _scheduler.to_timedelta(duetime)
p = max(0.0, _scheduler.to_seconds(period))
mad = MultipleAssignmentDisposable()
dt = duetime
count = 0
def action(scheduler: abc.SchedulerBase, state: Any) -> None:
nonlocal dt
nonlocal count
if p > 0.0:
now = scheduler.now
dt = dt + scheduler.to_timedelta(p)
if dt <= now:
dt = now + scheduler.to_timedelta(p)
observer.on_next(count)
count += 1
mad.disposable = scheduler.schedule_absolute(dt, action)
mad.disposable = _scheduler.schedule_absolute(dt, action)
return mad
return Observable(subscribe)
def observable_timer_timespan(
duetime: typing.RelativeTime, scheduler: Optional[abc.SchedulerBase] = None
) -> Observable[int]:
def subscribe(
observer: abc.ObserverBase[int], scheduler_: Optional[abc.SchedulerBase] = None
) -> abc.DisposableBase:
_scheduler = scheduler or scheduler_ or TimeoutScheduler.singleton()
d = _scheduler.to_seconds(duetime)
def action(scheduler: abc.SchedulerBase, state: Any) -> None:
observer.on_next(0)
observer.on_completed()
if d <= 0.0:
return _scheduler.schedule(action)
return _scheduler.schedule_relative(d, action)
return Observable(subscribe)
def observable_timer_timespan_and_period(
duetime: typing.RelativeTime,
period: typing.RelativeTime,
scheduler: Optional[abc.SchedulerBase] = None,
) -> Observable[int]:
if duetime == period:
def subscribe(
observer: abc.ObserverBase[int],
scheduler_: Optional[abc.SchedulerBase] = None,
) -> abc.DisposableBase:
_scheduler: abc.SchedulerBase = (
scheduler or scheduler_ or TimeoutScheduler.singleton()
)
def action(count: Optional[int] = None) -> Optional[int]:
if count is not None:
observer.on_next(count)
return count + 1
return None
if not isinstance(_scheduler, PeriodicScheduler):
raise ValueError("Sceduler must be PeriodicScheduler")
return _scheduler.schedule_periodic(period, action, state=0)
return Observable(subscribe)
return observable_timer_duetime_and_period(duetime, period, scheduler)
def timer_(
duetime: typing.AbsoluteOrRelativeTime,
period: Optional[typing.RelativeTime] = None,
scheduler: Optional[abc.SchedulerBase] = None,
) -> Observable[int]:
if isinstance(duetime, datetime):
if period is None:
return observable_timer_date(duetime, scheduler)
else:
return observable_timer_duetime_and_period(duetime, period, scheduler)
if period is None:
return observable_timer_timespan(duetime, scheduler)
return observable_timer_timespan_and_period(duetime, period, scheduler)
__all__ = ["timer_"]
|
