如何访问上游对象，然后在 RxJava2 中访问下游对象

Question

假设我有这种 Observable 情况：

public void main() {
    Observable.fromIterable(Arrays.asList(1, 2, 3, 4, 5))
            .flatMap(id -> getEvenOdd(id))
            .map(string -> {
                // I now want to join string 
                // AND the last emitted ID integer
                return null;
            });
}
    
private Observable<String> getEvenOdd(Integer id) {
    if (id % 2 == 0) {
        return Observable.just("even");
    } else {
        return Observable.just("odd");
    }
}

flatmap 已将 Integer 转换为 String。我现在如何才能访问 map 中的整数？

我知道我可以添加一个 doOnNext 并缓存 Integer:

private Integer intCache; 
    
public void main() {
    Observable.fromIterable(Arrays.asList(1, 2, 3, 4, 5))
            .doOnNext(integer -> intCache = integer)
            .flatMap(id -> getEvenOdd(id))
            .map(string -> {
                return intCache.toString() + " " + string;
            });

但这似乎有点老套，并且将 Integer 的范围扩展到我的可观察链之外。

Answer 1

这个用例有一个专门的flatMap：

http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Observable.html#flatMap(io.reactivex.functions.Function,%20io.reactivex.functions.BiFunction)

第二个参数将平面图的结果与导致它们被发射的项目相结合。

所以更新后的例子是

public void main() {
    Observable.fromIterable(Arrays.asList(1, 2, 3, 4, 5))
            .flatMap(id -> getEvenOdd(id),
                    (BiFunction<Integer, String, String>) (integer, string) -> { // LOOK HERE
                        return string + Integer.toString(integer);
                    })
            .map(joinedStringAndInt -> {
                ... use joinedStringAndInt
            });
}

其中 BiFunction 中的第三种类型是组合值的类型。这里我只是选择将String和Integer组合成另一个String

作为旁注，如果您认为可以有同一个 Observable 的多个实例访问 intCache 变量，doOnNext 解决方案是不安全的。

Answer 2

另一个解决方案就是使用嵌套。您可以在嵌套结构中同时拥有最后可观察到的发射和参数值：

public void main() {
    Observable.fromIterable(Arrays.asList(1, 2, 3, 4, 5))
        .flatMap(id -> getEvenOdd(id)
            .map(string -> joinStringAndInt(string, id)))
        .map(joinedStringAndInt -> {
          // do your stuff here
          return something;
        });
}

private String joinStringAndInt(String string, Integer number) {
    return number.toString() + " " + string;
}

private Observable<String> getEvenOdd(Integer id) {
    if (id % 2 == 0) {
      return Observable.just("even");
    } else {
      return Observable.just("odd");
    }
}

如何访问上游对象，然后在 RxJava2 中访问下游对象

How can I access an upstream object, later down stream in RxJava2

java

rx-java2