Returning combined Observable only if a condition matches

Question

Consider the following set and object:

Observable.from(users); // Where users = List<User> and each user has a userId
Observable.just(location); // Where location has id, userId, coordinates

What I would like to do is iterate over the list of users, and upon the first encounter where location.userId.equals(user.userId); is queried in a database, return a combined object. If userIds do not match move to the next user. And terminate the loop once 1 match is found.

How can I achieve this with RxJava?

I initially thought to use:

Observable.zip(Observable.from(users), Observable.just(location), new Func2<User, Location, UserLocation>() { ... });`

Does anyone have a better alternative?

Edit:

I thought maybe I could solve this with a simple solution, but alright I'll explain everything more clearly.

So once I have location.userId, and user.userId I need to also query a database which will return an Observable<Boolean> indicating whether it is also true in our database. If that condition matches then I return a combined object.

So the entire flow looks like this:

for each user in Users {
    checkIfAlreadyExistsInDatabase(user.userId, location.userId) // Returns Observable<Boolean>

    // If exists in db AND user.userId == location.userId return combined object and terminate the loop
}

This was previously being done synchronously without RxJava I converted the method checkIfAlreadyExistsInDatabase to Rx and use Schedulers.io to ping the database on a background thread to make the app more response. The issue came up when I had to iterate over an array of Users and match the id with Location AND also ping my database.

In order for me to call the method checkIfAlreadyExistsInDatabase I need to grab a user.userId and to do that I need to iterate over users and filter with location.userId.

So:

1. Iterate over users
2. If user.userId matches location.userId check if it exists in database
3. If exists in database return a combined object
4. Terminate loop once 1 match is found

Answer 1

The problem with zip function is that it emits one item from left Observable and one item from right Observable. So the function you provided will execute only for the first user. But this is a good direction. Just repeat the second Observable appropriate amount of times - use repeat. If you really want to do this using RxJava, here is the suggested approach:

Observable.zip(Observable.from(userList),
        Observable.just(location).repeat(userList.size()),
        new Func2<User, Location, User>() {
            @Override
            public User call(User user, Location location) {
                return user.id.equals(location.id) ? user : null;
            }
        })
        .filter(new Func1<User, Boolean>() {
            @Override
            public Boolean call(User user) {
                return user != null;
            }
        });

However with this approach null is passed through Observable stream and this is not recommended.

I wouldn't do it with RxJava, just with traditional Java's Iterator.

Answer 2

It's not just Observable.just(location), right? If it's already known, then it's trivial. Assuming that it's also procured asynchronously, then the 2-parameter form of flatMap is your friend:

Observable
.just(location)
.flatMap(loc -> Observable
    .from(users)
    .filter(user -> user.userId == location.userId)
    .flatMap(user -> checkIfAlreadyExistsInDatabase(user.userId, loc.userId)
                     .filter(value->value),
             (user, value) -> user)
    .take(1)
    .map(user -> combine(user, location)
)
.subscribe(...);