rx-java - 如何使用 rxJava 实现一系列连续操作？

Question

我有包含 3 个连续操作的下载过程:preProcess , downloading , postProcess .每个操作都具有异步性质(preProcess 调用 API，downloading 等待文件下载等)。 UI 必须显示正在执行的操作(例如，“正在准备……”、“正在下载……”、“正在解包……”)。
我看到整个过程为 Observable发出整个操作的当前状态。每个操作也是一个 observable，它在执行开始时发出他的状态并在执行后完成。

    Observable.OnSubscribe<DownloadStatus>() {
        @Override
        public void call(Subscriber<? super DownloadStatus> subscriber) {
            subscriber.onNext(DownloadStatus.PRE_PROCESS);
            doPreProcess()
                    .subscribe(new Action1<File>() {
                        @Override
                        public void call(File file) {
                            subscriber.onCompleted();
                        }
                    });
        }
    });

    Observable<DownloadStatus> mDonwloadingOperation = Observable.create(new Observable.OnSubscribe<DownloadStatus>() {
        @Override
        public void call(final Subscriber<? super DownloadStatus> subscriber) {
            subscriber.onNext(DownloadStatus.DOWNLOADING);
            doDownloading()
                    .subscribe(new Action1<File>() {
                        @Override
                        public void call(File file) {
                            subscriber.onCompleted();
                        }
                    });
        }
    });

    Observable<DownloadStatus> mPosProcessOperation = Observable.create(new Observable.OnSubscribe<DownloadStatus>() {
        @Override
        public void call(Subscriber<? super DownloadStatus> subscriber) {
            subscriber.onNext(DownloadStatus.POST_PROCESS);
            doPostProcess()
                    .subscribe(new Action1<File>() {
                        @Override
                        public void call(File file) {
                            subscriber.onCompleted();
                        }
                    });
        }
    });

一方面，每个操作都应该等到之前的操作完成。另一方面，订阅者需要接收每个发出的状态(例如 PRE_PROCESS -> DOWNLOADING -> POST_PROCESS -> onComplete)

我无法使用 merge因为每个操作都应该依赖于前一个操作的完成。
我无法使用 flatMap因为我不知道如何传播发出的状态。我认为 Subject可能是解决方案，但我也不知道如何传播发出的状态。

如何使用 rxJava 解决此类问题？感谢您提供任何想法/线索。

Answer 1

concat 是你所需要的。一旦前一个完成，它就会订阅连接的 observable。
concatMap也可以像 flatMap 一样工作但连接扁平的投影。有一个很好的图表here关于这两者的区别。