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
|
from typing import Any, Callable, Optional, TypeVar
import reactivex
from reactivex import Observable, abc
from reactivex import operators as ops
from reactivex import typing
_T = TypeVar("_T")
def delay_subscription_(
duetime: typing.AbsoluteOrRelativeTime,
scheduler: Optional[abc.SchedulerBase] = None,
) -> Callable[[Observable[_T]], Observable[_T]]:
def delay_subscription(source: Observable[_T]) -> Observable[_T]:
"""Time shifts the observable sequence by delaying the subscription.
Exampeles.
>>> res = source.delay_subscription(5)
Args:
source: Source subscription to delay.
Returns:
Time-shifted sequence.
"""
def mapper(_: Any) -> Observable[_T]:
return reactivex.empty()
return source.pipe(
ops.delay_with_mapper(reactivex.timer(duetime, scheduler=scheduler), mapper)
)
return delay_subscription
__all__ = ["delay_subscription_"]
|
