java - 如何通过键连接两个 RxJava2 Obvervables？

Question

我有两个不同类型的未排序的可观察量。这两种类型共享一个公共(public) key 。我想将它们加入到新的可观察发射的相应元素对中，但我不知道该怎么做。

请注意，某些键可能会丢失。如果删除不完整的对也没关系，但用 null 代替丢失的部分会更好。

输入1:

Entity(id = 2),
Entity(id = 1),
Entity(id = 4)

输入2:

Dto(id = 3),
Dto(id = 2),
Dto(id = 1)

预期输出(以任何顺序):

Pair(Entity(id = 1), Dto(id = 1)),
Pair(Entity(id = 2), Dto(id = 2)),
Pair(null, Dto(id = 3)),
Pair(Entity(id = 4), null)

Answer 1

首先，Observable.merge 将流合并在一起:这将为您提供所有项目的流。 (在下面的代码中，我使用了自定义 Either 类来标记每个流。)

然后，对于流中的每个项目，尝试将其与先前观察到的其他类型的项目进行匹配，并输出该对。如果没有，请保存以供稍后匹配。

最后，一旦流完成，剩余的不匹配元素将不会与任何内容匹配，因此它们可以不配对地发出。

import io.reactivex.Observable

data class Entity(val id: Int)
data class Dto(val id: Int)

sealed class Either<out A, out B>
data class Left<A>(val value: A) : Either<A, Nothing>()
data class Right<B>(val value: B) : Either<Nothing, B>()

fun <A : Any, B : Any, C> joinById(a: Observable<A>, idA: (A) -> C, b: Observable<B>, idB : (B) -> C): Observable<Pair<A?, B?>> {
    val unmatchedA = mutableMapOf<C, A>()
    val unmatchedB = mutableMapOf<C, B>()
    val merged = Observable.mergeDelayError(a.map(::Left), b.map(::Right)).flatMap { latest ->
        when (latest) {
            is Left -> {
                val id = idA(latest.value)
                unmatchedB.remove(id)?.let { return@flatMap Observable.just(latest.value to it) }
                unmatchedA.put(id, latest.value)
            }
            is Right -> {
                val id = idB(latest.value)
                unmatchedA.remove(id)?.let { return@flatMap Observable.just(it to latest.value) }
                unmatchedB.put(id, latest.value)
            }
        }
        Observable.empty<Nothing>()
    }
    return Observable.concat(merged, Observable.create { emitter ->
        unmatchedA.values.forEach { emitter.onNext(it to null) }
        unmatchedB.values.forEach { emitter.onNext(null to it) }
        emitter.onComplete()
    })
}

fun main(args: Array<String>) {
    val entities = Observable.just(Entity(2), Entity(1), Entity(4))
    val dtos = Observable.just(Dto(3), Dto(2), Dto(1))
    joinById(entities, Entity::id, dtos, Dto::id).blockingForEach(::println)
}

(Entity(id=2), Dto(id=2))
(Entity(id=1), Dto(id=1))
(Entity(id=4), null)
(null, Dto(id=3))

请注意，如果 id 在流中重复，这可能会产生一些奇怪的行为，并且根据流的结构，这可能最终会在内存中缓冲大量元素。