Create Observable from list based on conditions

Question

I'm realtive new to RxJava so I'm struggling on doing this kind of stuff, the thing is

I have one Observable that emits a List mapping from another class called Gatekeeper like this:

        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);
        }
    });

This is working fine, so now I want to create an Observable that emits only the first item that is valid from the list given some conditions, what I've created here is something like this:

  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;
                          }
                      }

My problem is here, when validating the ParkappGate I have to call one method that I've created that returns an Observable<ParkappParking> and another that returns a Observable and evaluate them with the ParkappGate passed as a parameter, then return if all conditions are true or not from the method isValidGate(ParkappGate gate)

The one simple approach I saw was converting the Observable<ParkappParking> and the Observable<Boolean> to values using toBlocking.first() but this doesn't seem the correct approach to Reactive X so I would like to know how to do it right.

Answer 1

So your problem is that your validation function is basically an Observable ? Too bad filter can't accept observable as predicate. You could flatmap your validation function but doing so you lose the reference to the object on which validation function was evaluated. What you could do however is using a temporary object to store both the object and the result of validation function, then filter on the result property.

This is an idea of what it could look. Severals notes;

• First, I rewrote your code with basics operators map, flatmap and filter to simplify it.

• Second, I used Observable for everything but you could achieve more expressive function signatures by using Single for isValidGate(...) and Maybe for nearestValidGate(...).

• Last, I used Rx 2 but with Rx 1 it should be mainly the same.

The code in 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();
    }
}

And the version with 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();
    }
}

You could also write your own filterWithObservable or filterWithSingle operator.

Answer 2

I want to create an Observable that emits only the first item that is valid

To emit the first item that matches with a given predicate, according to your requirements, you can use either the operator first with a predicate first(Func1<? super T, Boolean> predicate) or firstOrDefault with a default value and a predicate firstOrDefault(T defaultValue, Func1<? super T, Boolean> predicate).

In your case it could be something like this:

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

Answer 3

First of all, your first method could be written much more succintly as:

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

Note that you probably don't need a toList() if you consider all your Observables to be iterator-equivalents.

To filter out the gates that aren't valid, just use the filter operator:

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

See? Everything is easier if you don't mix collections and Observables.