如何使用来自另一个可观察 rx kotlin 的结果过滤可观察发出的列表？答案

【问题标题】：How to filter list emitted by observable using result from another observable rx kotlin?如何使用来自另一个可观察 rx kotlin 的结果过滤可观察发出的列表？
【发布时间】：2019-09-28 23:07:14
【问题描述】：

假设我有一个数字列表和 observable 的来源，它可以评估该列表中的每个数字。（需要使用以下方法。它们是我必须使用的 API 的一部分）

fun getNumberListSource() : Observable<List<Int>> {...}
fun getOddChecker(value: Int): Observable<Boolean> {...}

假设来源可能如下所示：

class NumberSource {

    private val source = BehaviorSubject.create<List<Int>>()

    fun getNumberListSource(): Observable<List<Int>> = source.doOnNext { println("doOnNext :)") }

    fun start(){
        Thread {
            Thread.sleep(1000)
            source.onNext(listOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12))
            Thread.sleep(1000)
            source.onNext(listOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12))
            Thread.sleep(1000)

            source.onComplete()
        }.start()

    }
}

fun getOddChecker(i: Int) : Observable = Observable.just(i%2== 0)

在流的末尾，我需要有奇数列表。所以我完成了那个代码：

fun main() {
    val numberSource = NumberSource()
    numberSource.start()
    println("== started ==")

    numberSource.getNumberListSource()
        .flatMap {
            Observable.fromIterable(it)
                .flatMap (
                    {getOddChecker(it)},
                    {a, b -> a to b}
                )
                .filter { it.second }
                .map { it.first }
                .toList()
                .toObservable()
        }
        .subscribe({ println("onNext: $it")},{},{ print("action:onComplete")})

}

以上代码的输出：

== started == 
doOnNext :) 
onNext: [2, 4, 6, 8, 10, 12] 
doOnNext :) 
onNext: [2, 4, 6, 8, 10, 12] 
action:onComplete

上面的代码可以满足我的要求，但我必须使用 Pair 类。部分

        .filter { it.second }
        .map { it.first }

似乎不太可读。那么是否有任何运营商女巫可以简化该代码？

【问题讨论】：

标签： rx-java2 rx-kotlin2

【解决方案1】：

你可以把它简化成这样的

   getNumberListSource()
           .flatMap { Observable.fromIterable(it)
                    .flatMap { item -> getOddChecker(item).filter {it}.map { item } }
                    .toList().toObservable()
            }
            .subscribe({ println("onNext: $it")},{},{ print("action:onComplete")})

【讨论】：

getOddChecker 是必需的。它是域 api 的一部分。
getOddChecker 到底是做什么的？
它发出 http 请求。这个例子简化了我的实际问题：我得到了产品列表和特殊端点，它根据产品 ID 进行分类。如果端点返回 false，则应用不应显示产品（从列表中删除）。
明白。我已根据您的要求更新了答案
几乎。现在，如果 observable 发出多于一个列表，那么结果将是连续的。它也会等到可观察完成。