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

Question

我下载的过程,它由3个连续的操作:preProcess,downloading,postProcess.每个操作都具有异步性质(preProcess调用API,downloading等待下载文件等).UI必须显示正在执行哪些操作(例如,"准备......","下载...","解包......").我看到整个过程,因为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平坦的投影一样连接起来.这里有一个很好的图表,说明这两者之间的区别.