java如何使用RxJava/RxAndroid在OnNext中引发异常

Question

我在安卓应用程序中对网络通话进行改造。现在，如果响应出现错误（可能是错误的数据），我不希望执行onComplete。请看代码片段

restClient.getService().getProjectDetail(projectId)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(new DisposableObserver<List<Project>>() {
                    @Override
                    public void onNext(List<Project> value) {
                        /*Something wrong in the data and I would like to execute onError*/
                    }

                    @Override
                    public void onError(Throwable e) {
                        handleError(e, 0, "");
                        hideProgressDialog();
                    }

                    @Override
                    public void onComplete() {
                        hideProgressDialog();
                    }
                });

提前谢谢

Answer 1

你可以用平面地图。例：

restClient.getService().getProjectDetail(projectId)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .flatMap(new Function<Integer, Observable<List<Project>>>() {
                    @Override
                    public Observable<List<Project>> apply(List<Project> x) {
                        if(validate(x)){
                            return Observable.error(new Exception("Response is invalid"));
                        }else {
                          return   Observable.just(x);
                        }
                    }

                    public boolean validate(List<Project> x){
                        return x.size()==0;
                    }
                })
                .subscribeWith(new DisposableObserver<List<Project>>() {
                    @Override
                    public void onNext(List<Project> value) {
                        /*Something wrong in the data and I would like to execute onError*/
                    }

                    @Override
                    public void onError(Throwable e) {
                        handleError(e, 0, "");
                        hideProgressDialog();
                    }

                    @Override
                    public void onComplete() {
                        hideProgressDialog();
                    }
                });

其他示例：

Observable.just(1)
                .flatMap(new Function<Integer, Observable<Integer>>() {
                    @Override
                    public Observable<Integer> apply(Integer x) {
                        if(validate(x)){
                            return Observable.error(new Exception("Response is invalid"));
                        }else {
                          return   Observable.just(x);
                        }

                    }

                    public boolean validate(Integer x){
                        return x==1;
                    }
                })
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                    }
                    @Override
                    public void onComplete() {
                    }
                    @Override
                    public void onError(Throwable e) {
                        Log.e("ERROR", e.getMessage());
                    }
                    @Override
                    public void onNext(Integer integer) {
                        Log.d("flatMap", "onNext: " + integer.toString());
                    }
            });