根据条件从列表中创建 Observable

Question

我是 RxJava 的新手，所以我在做这种事情时很努力，事情是

我有一个 Observable，它从另一个名为 Gatekeeper 的类发出 List 映射，如下所示：

        List<ParkappGate> actualgates = Observable.just(gates).concatMap(new Func1<List<Gatekeeper>, Observable<? extends List<ParkappGate>>>() {
        @Override
        public Observable<? extends List<ParkappGate>> call(List<Gatekeeper> gatekeepers) {
            final List<ParkappGate> gates = new ArrayList<ParkappGate>();
            for (Gatekeeper gate :
                    gatekeepers) {
                mDataManager.getGateById(gate.getCode()).subscribe(new Subscriber<ParkappGate>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(ParkappGate gate) {
                        gates.add(gate);
                    }
                });
            }
            return Observable.just(gates);
        }
    });

这工作正常，所以现在我想创建一个 Observable，它只发出在某些条件下列表中有效的第一项，我在这里创建的是这样的：

  Observable<ParkappGate>   nearestValidGate = actualgates.
            concatMap(new Func1<List<ParkappGate>, Observable<? extends ParkappGate>>() {
                          @Override
                          public Observable<? extends ParkappGate> call(List<ParkappGate> parkappGates) {
                              for (ParkappGate gate :
                                      parkappGates) {
                                  if (isValidGate(gate))
                                      return Observable.just(gate);
                              }
                              return null;
                          }
                      }

我的问题在这里，在验证 ParkappGate 时，我必须调用我创建的一个返回 Observable<ParkappParking> 的方法和另一个返回 Observable 并使用作为参数传递的 ParkappGate 评估它们，然后返回如果全部条件为真或不来自方法isValidGate(ParkappGate gate)

我看到的一种简单方法是使用 toBlocking.first() 将 Observable<ParkappParking> 和 Observable<Boolean> 转换为值，但这似乎不是 Reactive X 的正确方法，所以我想知道如何正确地做到这一点。

Answer 1

所以你的问题是你的验证函数基本上是一个 Observable ？太糟糕的过滤器不能接受可观察的作为谓词。您可以对验证函数进行平面映射，但这样做会丢失对评估验证函数的对象的引用。然而你可以做的是使用一个临时对象来存储对象和验证函数的结果，然后过滤结果属性。

这是一个关于它的外观的想法。几个笔记;

• 首先，我用基本运算符 map、flatmap 和 filter 重写了您的代码以简化它。

• 其次，我对所有事情都使用了 Observable，但是您可以通过使用 Single 用于 isValidGate(...) 和 Maybe 用于最近的ValidGate(...) 来获得更具表现力的函数签名。

• 最后，我使用了 Rx 2，但使用 Rx 1 应该基本相同。

Java pre-8 中的代码：

package io.nemo.commons;

import java.util.ArrayList;
import java.util.List;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;

import static java.lang.Boolean.TRUE;

public class ParkSO {

    class Gatekeeper {
        public int getCode() {
            return 1; // Dump implementation
        }
    }

    class ParkappGate {

    }

    class ParkappParking {

    }

    class DataManager {
        public Observable<ParkappGate> getGateById(int code) {
            return Observable.just(new ParkappGate()); // Dumb implementation
        }
    }

    class GateKeeperValidation {
        private ParkappGate gate;
        private Boolean validation;

        public ParkappGate getGate() {
            return gate;
        }

        public Boolean getValidation() {
            return validation;
        }

        public GateKeeperValidation(ParkappGate gate, Boolean validation) {
            this.gate = gate;
            this.validation = validation;
        }
    }

    Observable<ParkappParking> getParking(ParkappGate gate) {
        return Observable.just(new ParkappParking()); // Dumb implementation
    }

    Observable<Boolean> isValidParking(ParkappParking parking) {
        return Observable.just(TRUE); // Dumb implementation
    }

    Observable<Boolean> isValidGate(ParkappGate gate) {
        return getParking(gate)
                .flatMap(new Function<ParkappParking, ObservableSource<Boolean>>() {
                    @Override
                    public ObservableSource<Boolean> apply(ParkappParking parkappParking) throws Exception {
                        return isValidParking(parkappParking);
                    }
                });
    }

    void main() {
        final DataManager mDataManager = new DataManager();
        final List<Gatekeeper> gateKeepers = new ArrayList<>();

        Observable.fromIterable(gateKeepers)
                .flatMap(new Function<Gatekeeper, ObservableSource<ParkappGate>>() {
                    @Override
                    public ObservableSource<ParkappGate> apply(Gatekeeper gatekeeper) throws Exception {
                        return mDataManager.getGateById(gatekeeper.getCode());
                    }
                })
                .flatMap(new Function<ParkappGate, ObservableSource<GateKeeperValidation>>() {
                    @Override
                    public ObservableSource<GateKeeperValidation> apply(final ParkappGate parkappGate) throws Exception {
                        return isValidGate(parkappGate)
                                .map(new Function<Boolean, GateKeeperValidation>() {
                            @Override
                            public GateKeeperValidation apply(Boolean validation) throws Exception {
                                return new GateKeeperValidation(parkappGate, validation);
                            }
                        });
                    }
                })
                .filter(new Predicate<GateKeeperValidation>() {
                    @Override
                    public boolean test(GateKeeperValidation gateKeeperValidation) throws Exception {
                        return gateKeeperValidation.getValidation();
                    }
                })
                .firstElement();
    }
}

还有 Java 8 的版本：

package io.nemo.commons;

import java.util.ArrayList;
import java.util.List;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;

import static java.lang.Boolean.TRUE;

public class ParkSOJava8 {

    class Gatekeeper {
        public int getCode() {
            return 1; // Dump implementation
        }
    }

    class ParkappGate {

    }

    class ParkappParking {

    }

    class DataManager {
        public Observable<ParkappGate> getGateById(int code) {
            return Observable.just(new ParkappGate()); // Dumb implementation
        }
    }

    class GateKeeperValidation {
        private ParkappGate gate;
        private Boolean validation;

        public ParkappGate getGate() {
            return gate;
        }

        public Boolean getValidation() {
            return validation;
        }

        public GateKeeperValidation(ParkappGate gate, Boolean validation) {
            this.gate = gate;
            this.validation = validation;
        }
    }

    Observable<ParkappParking> getParking(ParkappGate gate) {
        return Observable.just(new ParkappParking()); // Dumb implementation
    }

    Observable<Boolean> isValidParking(ParkappParking parking) {
        return Observable.just(TRUE); // Dumb implementation
    }

    Observable<GateKeeperValidation> isValidGate(ParkappGate gate) {
        return getParking(gate)
                .flatMap(this::isValidParking)
                .map(validation -> new GateKeeperValidation(gate, validation));
    }

    void main() {
        final DataManager mDataManager = new DataManager();
        final List<Gatekeeper> gateKeepers = new ArrayList<>();

        Observable.fromIterable(gateKeepers)
                .map(Gatekeeper::getCode)
                .flatMap(mDataManager::getGateById)
                .flatMap(this::isValidGate)
                .filter(GateKeeperValidation::getValidation)
                .firstElement();
    }
}

您也可以编写自己的 filterWithObservable 或 filterWithSingle 运算符。

Answer 2

首先，你的第一个方法可以写得更简洁：

  Observable
  .from(gates)
  .flatMap(gate -> mDataManager.getGateById(gate.getCode()))

请注意，如果您认为所有 Observable 都是迭代器等价物，您可能不需要 toList()。

要过滤掉无效的门，只需使用filter 运算符：

  .filter(gate -> isValidGate(gate))
  .first()

看到了吗？如果你不混合集合和 Observable，一切都会变得更容易。

Answer 3

我想创建一个 Observable，它只发出第一个项目 有效

要发出与给定谓词匹配的第一个项目，根据您的要求，您可以使用运算符first 和谓词first(Func1<? super T, Boolean> predicate) 或firstOrDefault，默认值和谓词firstOrDefault(T defaultValue, Func1<? super T, Boolean> predicate)。

在你的情况下，它可能是这样的：

...
.first(this::isValidGate);