RxPY – 变换操作符

buffer

这个操作符将收集所有的值，从源观测点出发，一旦满足了给定的边界条件，就会以一定的时间间隔将它们发射出去。

语法

buffer(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:(optional) 在 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