以递归方式将Rx选项组合成Obserbles

比方说，我有一个名为s_0的Single ，它可以从Ttypes发出一个元素t_0 ，也可以失败（在某些语言中，这可能是Single ）。 那是：

 s_0: -- t_0 // Success OR s_0: -- X // Failure 

Ttypes的实例有一个next()方法，它返回Ttypes的一个可选的Single （ Kotlin中的Single? ）。 这种行为导致一个能够发射一个T实例链的Single实例链，其中每个单独的s_i可以发出一个能够返回下一个单独的s_i+1的元素t_i+1 ，这将发出一个元素t_i+1等等，直到最后一个元素t_n-1没有返回单个或任何单数失败：

 s_0: -- t_0 ↓ s_1: -- t_1 ↓ s_2: -- t_2 ... ↓ s_n-1: -- t_n-1 ↓ null OR s_0: -- t_0 ↓ s_1: -- t_1 ↓ s_2: -- t_2 ... ↓ s_i: -- X 

我正在寻找一种优雅的方式来获得一个T型的Observable o ，它能够在链上没有更多的单链或者失败的情况下成功地从s_0开始，

 o: -- t_0 -- t_1 -- t_2 -- ... -- t_n-1 --o // Success OR o: -- t_0 -- t_1 -- t_2 -- ... --X // Failure 

通过优雅 ，我的意思是这样简单（在科特林 ）：

 // Get single somehow (out of the scope of this question) val s0: Single = provideSingle() // Get observable val o: Observable = s0.chain() // Define extension method fun Single.chain(): Observable { /* Implement here */ } // Element interface interface T { fun next(): Single? } 

这是什么适用性？

当使用分页的REST API时，可以发现这种情况，其中可以使用Single实例来检索单个页面，从而可以提供能够发送后续页面的Single实例。

Answer 1

我没有测试过这个，但是基于我前一段时间编写的解决方案，我把它翻译成Kotlin的类似的分页问题

 fun Single.chain(): Observable = toObservable() .concatWith { it.next()?.chain() ?: Observable.empty() } 

获得“递归”链接的关键是concatWith运算符递归调用chain方法

Answer 2

 public class Q44535765 { public static void main(String[] args) { Maybe first = get(); first.toObservable() .compose(o -> chain(o)) .doOnError(e -> System.out.println(e)) .subscribe( e -> System.out.println(e), e -> System.out.println("fail"), () -> System.out.println("complete")); } static Maybe get() { return Maybe.just( () -> If.> that(Math.random() > 0.1) .tobe(() -> get()) .orbe(() -> If.> that(Math.random() > 0.5) .tobe(() -> Maybe.empty()) .orbe(() -> null) .result()) .result()); } static Observable chain(Observable s) { return s.concatMap( e -> Observable.just(e) .concatWith(e.next() .toObservable() .compose(o -> chain(o)))); } interface Element { Maybe next(); } } 

If是我的util类，你可以通过if...else...