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