RxPY - 变换运算符

---

---

buffer

该运算符将收集源可观察对象中的所有值，并在给定的边界条件满足时以常规时间间隔发出这些值。

语法

buffer(boundaries)

参数

boundaries：输入是可观察对象，它将决定何时停止，以便发出收集到的值。

返回值

返回值是可观察对象，它将具有从源可观察对象收集的所有值以及根据所采用的输入可观察对象确定的时间持续时间。

示例

from rx import of, interval, operators as op
from datetime import date
test = of(1, 2,3,4,5,6,7,8,9,10)
sub1 = test.pipe(
   op.buffer(interval(1.0))
)
sub1.subscribe(lambda x: print("The element is {0}".format(x)))

输出

E:\pyrx>python test1.py
The elements are [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

ground_by

该运算符将根据给定的 key_mapper 函数对源可观察对象中传出的值进行分组。

语法

group_by(key_mapper)

参数

key_mapper：此函数将负责从源可观察对象中提取键。

返回值

它返回一个具有根据 key_mapper 函数分组的值的可观察对象。

示例

from rx import from_, interval, operators as op
test = from_(["A", "B", "C", "D"])
sub1 = test.pipe(
   op.group_by(lambda v: v[0])
)
sub1.subscribe(lambda x: print("The element is {0}".format(x)))

输出

E:\pyrx>python testrx.py
The element is <rx.core.observable.groupedobservable.GroupedObservable object
at
 0x000000C99A2E6550>
The element is <rx.core.observable.groupedobservable.GroupedObservable object at
 0x000000C99A2E65C0>
The element is <rx.core.observable.groupedobservable.GroupedObservable object at
 0x000000C99A2E6588>
The element is <rx.core.observable.groupedobservable.GroupedObservable object at
 0x000000C99A2E6550>

map

该运算符将根据给定 mapper_func 的输出，将源可观察对象中的每个值更改为一个新值。

语法

map(mapper_func:None)

参数

mapper_func：（可选）它将根据该函数传出的输出更改源可观察对象中的值。

示例

from rx import of, interval, operators as op
test = of(1, 2,3,4,5,6,7,8,9,10)
sub1 = test.pipe(
   op.map(lambda x :x*x)
)
sub1.subscribe(lambda x: print("The element is {0}".format(x)))

输出

E:\pyrx>python testrx.py
The element is 1
The element is 4
The element is 9
The element is 16
The element is 25
The element is 36
The element is 49
The element is 64
The element is 81
The element is 100

scan

该运算符将累加器函数应用于源可观察对象中传出的值，并返回一个具有新值的可观察对象。

语法

scan(accumulator_func, seed=NotSet)

参数

accumulator_func：此函数应用于源可观察对象中的所有值。

seed：(可选) accumular_func 中要使用的初始值。

返回值

此运算符将返回一个可观察对象，该可观察对象将具有基于应用于源可观察对象的每个值的累加器函数的新值。

示例

from rx import of, interval, operators as op
test = of(1, 2,3,4,5,6,7,8,9,10)
sub1 = test.pipe(
   op.scan(lambda acc, a: acc + a, 0))
sub1.subscribe(lambda x: print("The element is {0}".format(x)))

输出

E:\pyrx>python testrx.py
The element is 1
The element is 3
The element is 6
The element is 10
The element is 15
The element is 21
The element is 28
The element is 36
The element is 45
The element is 55