结合之前的 observable

Question

我正在尝试使用 RxJava、RxAndroid 和 Mosby3 将两种表单插入合并到一个中，但我找不到让它工作的方法。

我的结构：

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

我在哪里激发我的意图（MVI 模式）：

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

并将它们绑定在一起：

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

我的问题是，我的 uploadImage 返回一个以字符串结尾的内部结构，但是，如何获取返回的字符串，将其添加到我的 command 对象（在此对象中设置返回的 URL）并继续流（将我的命令发送到云端）？

谢谢！

Answer 1

直接在第一个 flatMap 内的 observable 上使用 flatMap。在这种情况下，您可以同时参考 checkinIntent 和命令

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

替代解决方案：将 Pair<CheckinIntent, Command> 从 bookingRepository.doCheckin(...) 传递给 Observable，如下所示：

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

---

只是一些其他说明：

您几乎希望在 MVI 中更喜欢 switchMap() 而不是 flatMap()。 switchMap 取消订阅以前的订阅，而 flatMap 没有。这意味着，如果您像在代码中所做的那样 flatMap 被剪断，并且如果由于某种原因在旧的 checkinIntent 尚未完成时触发了新的 checkinIntent（即 imageRepository.uploadImage() 仍在进行中），您最终会得到两个流这将调用CheckinView::render，因为第一个仍然继续工作并通过您建立的可观察流向下发出结果。 switchMap() 通过在开始“切换映射”新意图之前取消订阅第一个（未完成的）意图来防止这种情况发生，这样您当时只有一个流。

您构建CheckinIntent 的方式应移至演示者。对于“转储”视图来说，这有点“逻辑”。 Observable.just(BitmapFactory.decodeFile(...)) 也在主线程上运行。我建议使用Observable.fromCallable( () -> BitmapFactory.decodeFile(...))，因为后者会推迟他的“工作”（位图解码），直到实际订阅了这个 observable，然后您可以应用后台调度程序。 Observable.just() 基本相同：

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