通过属性匹配两个无序流，RxJava, RxAndroid

Question

有如下两个流

Stream 1: A -- B -- C -- D
Stream 2: 3 -- 1 -- 2 -- 4

我想输出

Output: A1 -- B2 -- C3 -- D4

或

Output: 3C -- 2B -- 1A -- 4D

每个字母与其在字母表上的位置相匹配。我试过使用操作符.zip和.combineLatest，但没有得到我要找的结果。

这实际上是我的问题在现实中的简化。A B和C是具有属性的对象，比如1 2或3。我可以输入A.position() = 1

是否有一个操作符可以专门解决我的问题，或者我必须使用哈希映射并自己匹配这些对象?

Answer 1

像这样使用Observable.toSortedList:

Observable<Data> s1 = Observable.from(new Data[]{new Data(3), new Data(1), new Data(2), new Data(4)});
Observable<String> s2 = Observable.from(new String[]{"A", "B", "C", "D"});
Observable.zip(
        s1.toSortedList(this::compare).flatMap(Observable::from),
        s2,
        (d, s) -> s + d.number
).subscribe(s -> {
    //Output: A1 -- B2 -- C3 -- D4
});

compare方法:

private Integer compare(Data a, Data b) {
    return a.number.equals(b.number) ? 0 : (a.number < b.number ? -1 : 1);
}

Data class:

class Data {
    private final Integer number;
    Data(Integer number) {
        this.number = number;
    }
}

Answer 2

我会尝试在一个流上使用.flatMap()，在另一个流上使用.groupBy(key)，这将生成Observable>

或在另一个流上使用.toMap()或.toMultimap()，这将生成Observable

不确定您是否期望一个特定的顺序

Answer 3

您可以使用rxjava-extras 0.8.0.3(或更高版本，如果可用)中的单元测试匹配运算符。该工件发布在Maven Central上。

添加到您的pom.xml:

<dependency>
  <groupId>com.github.davidmoten</groupId>
  <artifactId>rxjava-extras</artifactId>
  <version>0.8.0.3</version>
</dependency>

的例子:

Observable<Integer> a = Observable.just(1, 2, 4, 3);
Observable<Integer> b = Observable.just(1, 2, 3, 5, 6, 4);
Obs.match(a, b,
     x -> x, // key to match on for a
     x -> x, // key to match on for b
     (x, y) -> x // combiner
    )
   .forEach(System.out::println);

输出:

1
2
4
3