RxPY 可连接操作符
publish
此方法将可观测对象转换为可连接的可观测对象。
语法
publish(mapper=None)
参数
mapper: 可选参数。用于多次广播源值的函数,无需进行多次订阅。
示例
from rx import create, range, operators as op
import random
def test_observable(observer, scheduler):
observer.on_next(random.random())
observer.on_completed()
source = create(test_observable).pipe(op.publish())
test1 = source.subscribe(on_next = lambda i: print("From subscriber 1 - {0}".format(i)))
test2 = source.subscribe(on_next = lambda i: print("From subscriber 2 –
{0}".format(i)))
source.connect()
输出
E:\pyrx>python testrx.py
From subscriber 1 - 0.14751607273318490
From subscriber 2 - 0.1475160727331849
ref_count
该操作符将使得可观察对象成为普通的可观察对象。
语法
ref_count()
示例
from rx import create, operators as op
import random
def test_observable(observer, scheduler):
observer.on_next(random.random())
source = create(test_observable).pipe(op.publish(),op.ref_count())
test1 = source.subscribe(on_next = lambda i: print("From subscriber 1 - {0}".format(i)))
test2 = source.subscribe(on_next = lambda i: print("From subscriber 2 - {0}".format(i)))
输出
E:\pyrx>python testrx.py
From subscriber 1 - 0.8230640432381131
replay
该方法类似于replaySubject。即使observable已经发出了值,并且一些订阅者延迟订阅,该方法也会返回相同的值。
语法
replay()
示例
from rx import create, range, operators as op
import random
from threading import Timer
def test_observable(observer, scheduler):
observer.on_next(random.random())
observer.on_completed()
source = create(test_observable).pipe(op.replay())
test1 = source.subscribe(on_next = lambda i: print("From subscriber 1 - {0}".format(i)))
test2 = source.subscribe(on_next = lambda i: print("From subscriber 2 - {0}".format(i)))
source.connect()
print("subscriber called after delay ")
def last_subscriber():
test3 = source.subscribe(on_next = lambda i: print("From subscriber 3 - {0}".format(i)))
t = Timer(5.0, last_subscriber)
t.start()
输出
E:\pyrx>python testrx.py
From subscriber 1 - 0.8340998157725388
From subscriber 2 - 0.8340998157725388
subscriber called after delay
From subscriber 3 - 0.8340998157725388