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
|
from typing import Any, Callable, Iterable, Optional, Tuple, TypeVar
import reactivex
from reactivex import Observable, abc
_T = TypeVar("_T")
_TOther = TypeVar("_TOther")
def zip_(
*args: Observable[Any],
) -> Callable[[Observable[Any]], Observable[Tuple[Any, ...]]]:
def _zip(source: Observable[Any]) -> Observable[Tuple[Any, ...]]:
"""Merges the specified observable sequences into one observable
sequence by creating a tuple whenever all of the
observable sequences have produced an element at a corresponding
index.
Example:
>>> res = zip(source)
Args:
source: Source observable to zip.
Returns:
An observable sequence containing the result of combining
elements of the sources as a tuple.
"""
return reactivex.zip(source, *args)
return _zip
def zip_with_iterable_(
seq: Iterable[_TOther],
) -> Callable[[Observable[_T]], Observable[Tuple[_T, _TOther]]]:
def zip_with_iterable(source: Observable[_T]) -> Observable[Tuple[_T, _TOther]]:
"""Merges the specified observable sequence and list into one
observable sequence by creating a tuple whenever all of
the observable sequences have produced an element at a
corresponding index.
Example
>>> res = zip(source)
Args:
source: Source observable to zip.
Returns:
An observable sequence containing the result of combining
elements of the sources as a tuple.
"""
first = source
second = iter(seq)
def subscribe(
observer: abc.ObserverBase[Tuple[_T, _TOther]],
scheduler: Optional[abc.SchedulerBase] = None,
):
index = 0
def on_next(left: _T) -> None:
nonlocal index
try:
right = next(second)
except StopIteration:
observer.on_completed()
else:
result = (left, right)
observer.on_next(result)
return first.subscribe(
on_next, observer.on_error, observer.on_completed, scheduler=scheduler
)
return Observable(subscribe)
return zip_with_iterable
__all__ = ["zip_", "zip_with_iterable_"]
|
