扁平化 Observable 的 Observable答案

【问题标题】：Flatten Observable of Observables扁平化 Observable 的 Observable
【发布时间】：2018-04-19 19:45:02
【问题描述】：

我想做的是创建一个每秒运行另一个函数的函数。第二个函数返回Observables<A>，我希望第一个函数也返回Observables<A>，而不是Observable<Observable<A>>

例如：

private A calcA(){
   ...
   return new A(...)
}

public Observable<A> getAs(){
   return Observable.create( subscriber -> {
      Bool condition = ...
      do {
         subscriber.onNext(calcA())
      } while (condition)
      subscriber.onComplete()
   })
}

public Observable<A> pollAs(){
   return Observable.create(subscriber -> {
      do {
         subscriber.onNext(getAs()) // Flatten here I guess
         Thread.sleep(1000)
      } while(true)
   })

所以我想做类似的事情（我尝试用 Java 式的方式来写，但我会使用 Kotlin）

【问题讨论】：

标签： rx-java observable reactivex rx-kotlin

【解决方案1】：

您不需要使用flatMap() 运算符来展平内部可观察对象，因为您只想重复订阅同一个可观察对象。

public Observable<A> getAs() {
   return Observable.fromCallable( () -> calcA() )
            .repeat()
            .takeWhile( v -> !condition( v );
}

getAs() 将发射项目，直到达到条件。然后它将完成。

public Observable<A> pollAs(){
   return getAs()
            .repeatWhen( completed -> completed.delay(1000, TimeUnit.MILLISECONDS) );

pollAs() 将不断重新订阅 getAs() observable，在每次订阅之间暂停一秒钟。

编辑：我已将一个为期 6 个月的示例上传到 https://pastebin.com/kSmi24GF 这表明您必须不断推进数据出来的时间。

【讨论】：

这看起来很漂亮，它温暖了我的心：D
还有一个问题，请。在另一种情况下，我调用calcAs，它返回A-s 的列表，我想一个一个地发出它们。我应该在repeat() 之前添加一个flatMap(items -> Observable.fromIterable(items)) 还是有更好的解决方案？
另外repeatWhen 只重复getAs() 一次，并且不会一遍又一遍地重复，我应该改用repeatUntil 吗？
repeatWhen() 接受一个返回 Observable 的函数。当原始 observable 上的 onComplete 发生时，它会向 observable 发出一个值，该值发出一个值，在这种情况下原始 observable 被重新订阅，或者 onComplete/onError 不会。换句话说，它将永远重复。
对我来说，这不会永远重复 :(

【解决方案2】：

我想出了这个解决方案：

public Observable<A> pollAs() {
   return Observable.create(subscriber -> {
       do {
           getAs().subscribe(
                   { subscriber.onNext(it) },
                   { subscriber.onError(it) },
                   { Thread.sleep(1000) }
           )
       } while (true)
   })
}

我不太喜欢这个，谁能给我一个更方便的方法？

【讨论】：