Combine previous observable

Question

I'm trying to combine two forms insertion in one using RxJava, RxAndroid and Mosby3, but I can't find a way to make it work.

My structure:

public final class CheckinIntent {

    private final CheckinCommand checkinCommand;
    private final Bitmap signature;

    public CheckinIntent(CheckinCommand checkinCommand, Bitmap signature) {
        this.checkinCommand = checkinCommand;
        this.signature = signature;
    }

    public CheckinCommand getCheckinCommand() {
        return checkinCommand;
    }

    public Bitmap getSignature() {
        return signature;
    }
}

Where I fire my intent (MVI pattern):

final Observable<Bitmap> signatureObservable = Observable.just(BitmapFactory.decodeFile(storage.getFile("signs", booking.getBookingId()).getAbsolutePath()));
        final Observable<CheckinCommand> checkinCommandObservable = Observable.just(new CheckinCommand(booking.getBookingId(), booking.getUserId(), booking.getPartnerId(), userDetailsTextView.getText().toString(), "google.com"));

        final Observable<CheckinIntent> intentObservable = Observable.zip(signatureObservable, checkinCommandObservable, (image, command) -> new CheckinIntent(command, image));

        return saveButtonClickObservable
                .flatMap(bla -> intentObservable);

And binding it all together:

 @Override
    protected void bindIntents() {
        Observable<CheckinViewState> checkinViewStateObservable =
                intent(CheckinView::sendCheckin)
                        .flatMap(checkinIntent -> imageRepository.uploadImage(checkinIntent.getSignature())
                        .flatMap(command ->  bookingRepository.doCheckin(command) <------ PROBLEM HERE, HOW CAN I ACCESS THE COMMAND FROM ABOVE ??
                                .subscribeOn(Schedulers.from(threadExecutor))
                                .map(CheckinViewState.Success::new)
                                .cast(CheckinViewState.class)
                                .startWith(new CheckinViewState.LoadingState())
                                .onErrorReturn(CheckinViewState.ErrorState::new))
                        .observeOn(postExecutionThread.getScheduler());

        subscribeViewState(checkinViewStateObservable, CheckinView::render);
}

Observable<CnhImageResponse> uploadImage(Bitmap bitmap);

My problem is, my uploadImage returns an internal structure that ends of on a String, but, how can I get the returned string, add it to my command object (setting the returned URL in this object) and continue the flow (sending my command to the cloud) ?

Thanks!

Answer 1

Just flatMap on the observable directly within the first flatMap. In that case you have reference to both, the checkinIntent and command

 @Override
 protected void bindIntents() {
        Observable<CheckinViewState> checkinViewStateObservable =
                intent(CheckinView::sendCheckin)
                        .flatMap(checkinIntent -> { 
                          return imageRepository.uploadImage(checkinIntent.getSignature()
                                                .flatMap(imageResponse ->  bookingRepository.doCheckin(command) <-- Now you have access to both, command and CnhImageResponse 
                         }) 
                         .subscribeOn(Schedulers.from(threadExecutor))
                         .map(CheckinViewState.Success::new)
                         .cast(CheckinViewState.class)
                         .startWith(new CheckinViewState.LoadingState())
                         .onErrorReturn(CheckinViewState.ErrorState::new))
                         .observeOn(postExecutionThread.getScheduler());

        subscribeViewState(checkinViewStateObservable, CheckinView::render);
}

Alternative solution: Pass a Pair<CheckinIntent, Command> to the Observable from bookingRepository.doCheckin(...) like this:

@Override
protected void bindIntents() {
        Observable<CheckinViewState> checkinViewStateObservable =
                intent(CheckinView::sendCheckin)
                        .flatMap(checkinIntent -> imageRepository.uploadImage(checkinIntent.getSignature()
                                                                 .map(imageResponse -> Pair.create(checkinIntent, imageResponse))) // Returns a Pair<CheckinIntent, CnhImageResponse>
                        .flatMap(pair ->  bookingRepository.doCheckin(pair.first) <-- Now you can access the pair holding both information
                                .subscribeOn(Schedulers.from(threadExecutor))
                                .map(CheckinViewState.Success::new)
                                .cast(CheckinViewState.class)
                                .startWith(new CheckinViewState.LoadingState())
                                .onErrorReturn(CheckinViewState.ErrorState::new))
                        .observeOn(postExecutionThread.getScheduler());

        subscribeViewState(checkinViewStateObservable, CheckinView::render);
}

---

Just a few other notes:

You almost ever want to prefer switchMap() over flatMap() in MVI. switchMap unsubscribes previous subscription while flatMap doesnt. That means that if you flatMap as you did in your code snipped and if for whatever reason a new checkinIntent is fired while the old one hasn't completed yet (i.e. imageRepository.uploadImage() is still in progress) you end up having two streams that will call CheckinView::render because the first one still continue to work and emit results down through your established observable stream. switchMap() prevents this by unsubscribing the first (uncompleted) intent before starting "switchMaping" the new intent so that you only have 1 stream at the time.

The way you build your CheckinIntent should be moved to the Presenter. This is kind of too much "logic" for a "dump" View. Also Observable.just(BitmapFactory.decodeFile(...)) is running on the main thread. I recommend to use Observable.fromCallable( () -> BitmapFactory.decodeFile(...)) as the later deferres his "work" (bitmap decoding) until this observable is actually subscribed and then you can apply background Schedulers. Observable.just() is basically the same as:

Bitmap bitmap = BitmapFactory.decodeFile(...); // Here is the "hard work" already done, even if observable below is not subscribed at all.
Observable.just(bitmap);