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
|
from typing import Any, Callable, Optional, TypeVar, Union, cast
import reactivex
from reactivex import Observable, abc, typing
from reactivex.disposable import CompositeDisposable
_T = TypeVar("_T")
def sample_observable(
source: Observable[_T], sampler: Observable[Any]
) -> Observable[_T]:
def subscribe(
observer: abc.ObserverBase[_T], scheduler: Optional[abc.SchedulerBase] = None
):
at_end = False
has_value = False
value: _T = cast(_T, None)
def sample_subscribe(_: Any = None) -> None:
nonlocal has_value
if has_value:
has_value = False
observer.on_next(value)
if at_end:
observer.on_completed()
def on_next(new_value: _T):
nonlocal has_value, value
has_value = True
value = new_value
def on_completed():
nonlocal at_end
at_end = True
return CompositeDisposable(
source.subscribe(
on_next, observer.on_error, on_completed, scheduler=scheduler
),
sampler.subscribe(
sample_subscribe,
observer.on_error,
sample_subscribe,
scheduler=scheduler,
),
)
return Observable(subscribe)
def sample_(
sampler: Union[typing.RelativeTime, Observable[Any]],
scheduler: Optional[abc.SchedulerBase] = None,
) -> Callable[[Observable[_T]], Observable[_T]]:
def sample(source: Observable[_T]) -> Observable[_T]:
"""Samples the observable sequence at each interval.
Examples:
>>> res = sample(source)
Args:
source: Source sequence to sample.
Returns:
Sampled observable sequence.
"""
if isinstance(sampler, abc.ObservableBase):
return sample_observable(source, sampler)
else:
return sample_observable(
source, reactivex.interval(sampler, scheduler=scheduler)
)
return sample
__all__ = ["sample_"]
|
