python - 为什么repeat(n)不能与Reactive扩展中的create一起使用

Question

from rx import Observable, Observer
from __future__ import print_function
import random

def create_observable(observer):
    while True:
        observer.on_next(random.randint(1,100))    

Observable.create(create_observable).take_while(lambda x: x>50).repeat(6).subscribe(print)

给出

74 78 94 59 79 76

序列，而我预计每个数字都会重复 6 次

因此“重复”永远不适用于使用 create 方法创建的可观察量。

Answer 1

发布的代码获取 [51, 100] 范围内的随机整数序列的 6 倍。

尝试使用

(Observable.create(create_observable)
    .take_while(lambda x: x > 50)
    .select_many(lambda x: Observable.just(x).repeat(6))
    .subscribe(print))

或者只是

(Observable.create(create_observable)
    .take_while(lambda x: x > 50)
    .select_many(lambda v: [v] * 6)
    .subscribe(print))