Return combined observation only subject to compliance

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 location.userId.equals(user.userId); over the list of users, and after the first meeting with the request location.userId.equals(user.userId); return the merged object in the database. If userId does not match the transition to the next user. And end the loop as soon as 1 match is found.

How can I achieve this with RxJava?

Originally intended to use:

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

Does anyone have a better alternative?

Edit:

I thought that maybe I can solve it with a simple solution, but well, I will explain it more clearly.

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

So, the whole stream 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 previously ran synchronously without RxJava . I converted the checkIfAlreadyExistsInDatabase method to Rx and used Schedulers.io to ping the database in the background thread to make the application more responsive. The problem arose when I had to iterate over an array of users and map the identifier to the "Location" field, as well as "ping my database".

To call the checkIfAlreadyExistsInDatabase method, I need to capture user.userId , and for this I need to iterate over users and filter using location.userId .

So:

• Iterate over users
• If user.userId matches location.userId, check if the database exists
• If a merged object exists in the database
• End the loop if 1 match is found

Answer 1

The problem with the zip function is that it emits one element on the left of the Observable and one element on the right of the Observable . This way, the function you provided will only be performed for the first user. But this is a good direction. Just repeat the second Observable appropriate number of times - use repeat . If you really want to do this with 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 the Observable stream, and this is not recommended.

I would not do this with RxJava, just with the traditional Java Iterator .

Answer 2

This is not just Observable.just(location) , right? If this is already known, then this is trivial. Assuming it is also acquired 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(...); 

Answer 3

Ok, this seems to be a good way to do this (not tested, think so be careful):

 location = getLocation(); Observable.from(users) .flatMap(user -> return mPresenterRx.userIdAlreadyInDatabase(location, user.userId) // This returns a Observable<Boolean> .flatMap(alreadyExists -> alreadyExists ? Observable.just(user) : null)) // Flatmap back to a user .filter(user -> return user != null) .take(1) .map(user -> return combine(location, user)) // Returns a UserLocation .subscribe(...); 

It is possible if you want to continue to perform operations on this, you can then pin or merge Latest on this, and the other can be observed:

 Observable<UserLocation> getUserLocationObservable() { location = getLocation(); return Observable.from(users) .flatMap(user -> return mPresenterRx.userIdAlreadyInDatabase(location, user.userId) .flatMap(alreadyExists -> alreadyExists ? Observable.just(user) : null)) .filter(user -> return user != null) .take(1) .map(user -> return combine(location, user)); // Returns a UserLocation } // Combine Observable<UserLocation> userLocationObs = getUserLocationObservable(); Observable<OtherModel> otherModelObs = getOtherModelObs(); Observable.zip(userLocationObs, otherModelObs, ...) .subscribe(...); 

If anyone sees problems here, please comment below.