如何制作rx多路分解器

Question

我是Rx语义的新手。特别是我正在用RxJava学习它。

我想写一个使用rx语义的多路复用器。

详细信息：
我有一个可观测到的发射类型为T的事件。
类型T有一个类型K的属性，可以通过对它执行一些T->K函数"getAttr"来获得。
我想注册观察者，每个观察者都对具有不同"getAttr"结果值的T事件感兴趣
应该使用equals（）来匹配"getAttr"的值（意思是互斥的观察者）。

在我看来，应该已经有一些基本的构建块可以做这种事情了，但我没有找到。

有什么办法吗？

Answer 1

给定Observable<T> observable，此代码为属性的特定K value提供了一个可观测值：

Observable<T> observable = ...
final K value = ...
final Function<T, K> toAttribute = ...
Observable<GroupedObservable<K, T>> grouped = 
    observable.groupBy(element -> toAttribute.apply(element))
Observable<T> filtered = 
    .filter(groupedObservable -> groupedObservable.getKey().equals(value))
    .flatMap(observable -> observable)