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, List, Optional, TypeVar
from reactivex import Observable, compose
from reactivex import operators as ops
_T = TypeVar("_T")
def buffer_(
boundaries: Observable[Any],
) -> Callable[[Observable[_T]], Observable[List[_T]]]:
return compose(
ops.window(boundaries),
ops.flat_map(ops.to_list()),
)
def buffer_when_(
closing_mapper: Callable[[], Observable[Any]]
) -> Callable[[Observable[_T]], Observable[List[_T]]]:
return compose(
ops.window_when(closing_mapper),
ops.flat_map(ops.to_list()),
)
def buffer_toggle_(
openings: Observable[Any], closing_mapper: Callable[[Any], Observable[Any]]
) -> Callable[[Observable[_T]], Observable[List[_T]]]:
return compose(
ops.window_toggle(openings, closing_mapper),
ops.flat_map(ops.to_list()),
)
def buffer_with_count_(
count: int, skip: Optional[int] = None
) -> Callable[[Observable[_T]], Observable[List[_T]]]:
"""Projects each element of an observable sequence into zero or more
buffers which are produced based on element count information.
Examples:
>>> res = buffer_with_count(10)(xs)
>>> res = buffer_with_count(10, 1)(xs)
Args:
count: Length of each buffer.
skip: [Optional] Number of elements to skip between
creation of consecutive buffers. If not provided, defaults to
the count.
Returns:
A function that takes an observable source and returns an
observable sequence of buffers.
"""
def buffer_with_count(source: Observable[_T]) -> Observable[List[_T]]:
nonlocal skip
if skip is None:
skip = count
def mapper(value: Observable[_T]) -> Observable[List[_T]]:
return value.pipe(
ops.to_list(),
)
def predicate(value: List[_T]) -> bool:
return len(value) > 0
return source.pipe(
ops.window_with_count(count, skip),
ops.flat_map(mapper),
ops.filter(predicate),
)
return buffer_with_count
__all__ = ["buffer_", "buffer_with_count_", "buffer_when_", "buffer_toggle_"]
|
