RxJava比较两个可观察到的不同项目(java 7)

Question

我有两个可观察到的ob1 & ob2，它们发出项。如果ob1.tem包含在ob2中，我想忽略它。

例子：

ob1 contains A,B,C,D

ob2 contains E,C,B,G,J,O

产出：

ob contains A,D

表达这一点的最好方法是什么？

Answer 1

有一个包含运算符，它返回是否可观测到的包含给定元素的Observable<Boolean>。

Observable<Boolean> contains = observable2.contains(string);

您需要的是将所有元素从observable1映射到那些布尔观测值。

observable1.map(new Func1<String, Observable<Boolean>>() {
    @Override
    public Observable<Boolean> call(final String string) {
        return observable2.contains(string);
    }
})

这将给您提供Observable<Observable<Boolean>>，但是很难使用它。因此，您将所有这些布尔可观测值串联为一个。幸运的是，您所需要做的就是使用concatMap而不是map

observable1.concatMap(new Func1<String, Observable<Boolean>>() {
    @Override
    public Observable<Boolean> call(final String string) {
        return observable2.contains(string);
    }
})

现在有了包含元素的observable1和包含observable1中元素是否包含在observable2中的布尔值的Observable<Boolean>。

observable1:            A ---- B ---- C ---- D
concatMapObservable:    false  true   true   false

您可以轻松地将这两个可观察到的zip转换为可观察的，这将传递不包含在observable2中的元素，并用空字符串替换其他元素。

[concatMapObservable].zipWith(observable1, new Func2<Boolean, String, String>() {
    @Override
    public String call(final Boolean contains, final String string) {
        return contains ? "" : string;
    }
}

然后过滤空字符串。

[zippedObservable].filter(new Func1<String, Boolean>() {
    @Override
    public Boolean call(final String string) {
        return !TextUtils.isEmpty(string);
    }
})

整个代码组合在一起：

Observable<String> observable1 = Observable.from(new String[]{"A", "B", "C", "D"});
Observable<String> observable2 = Observable.from(new String[]{"E", "C", "B", "G", "J", "O"});

observable1.concatMap(new Func1<String, Observable<Boolean>>() {
    @Override
    public Observable<Boolean> call(final String string) {
        return observable2.contains(string);
    }
}).zipWith(observable1, new Func2<Boolean, String, String>() {
    @Override
    public String call(final Boolean contains, final String string) {
        return contains ? "" : string;
    }
}).filter(new Func1<String, Boolean>() {
    @Override
    public Boolean call(final String string) {
        return !TextUtils.isEmpty(string);
    }
}).subscribe(new Action1<String>() {
    @Override
    public void call(final String string) {
        Log.d("observable:", string);
    }
});

Answer 2

在接收到ob2的所有项之前，无法计算操作的输出，因为在知道属于ob2的所有项之前，无法判断ob1中的项是否包含在ob2中。因此，您必须等待ob2完成，然后才能从ob1筛选项。

下面是使用RxJava2的可能解决方案。

    Observable<String> ob1 = Observable.fromArray("A", "B", "C", "D");
    Observable<String> ob2 = Observable.fromArray("E", "C", "B", "G", "J", "O");

    Observable.combineLatest(ob2.toList().toObservable(), ob1, (list, value) -> list.contains(value) ? "" : value)
            .filter(value -> !TextUtils.isEmpty(value))
            .subscribeOn(Schedulers.computation())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(value -> Log.d("foo", "Value: " + value));