RxJava的使用问题。。报错

"

更新叙述一下需求，其实就是简单的

取数据的时候先判断本地有无缓存，有缓存则从数据库中取得，否则从网络请求并且保存到本地。

因为才接触rxjava，这样写目前倒是完成了功能（运转正常），但是这样写真的是合理的吗？
本地数据库查询 queryDP
如果没查询到数据
deepSpaceBean=null
onNext（deepSpaceBean）

    @Override
    public Observable getDP(final String date) {
        return Observable
                .create(new Observable.OnSubscribe<DeepSpaceBean>() {
                    @Override
                    public void call(Subscriber<? super DeepSpaceBean> subscriber) {
                        DeepSpaceBean deepSpaceBean = null;
                        try {
                            deepSpaceBean = queryDP(date);
                        } catch (Exception e) {
                            e.printStackTrace();
                            XIAOHUException xiaohuException = new XIAOHUException(e, XIAOHUException.DB_QUERY);
                            subscriber.onError(xiaohuException);
                        }
                        subscriber.onNext(deepSpaceBean);
                        subscriber.onCompleted();
                    }
                });
    }```
    然后接下来的操作符

          .flatMap(new Func1<DeepSpaceBean, Observable<DeepSpaceBean>>() {
                @Override
                public Observable<DeepSpaceBean> call(DeepSpaceBean deepSpaceBean) {
                   **（关键1）** if (deepSpaceBean == null) {
                        return mDPsRemoteDataSource.getDP(date)
                                .doOnNext(new Action1<DeepSpaceBean>() {
                                    @Override
                                    public void call(DeepSpaceBean deepSpaceBean) {
                                        try {
                                            mDPsLoaclDataSource.saveDP(deepSpaceBean);
                                        } catch (Exception e) {
                                            e.printStackTrace();
                                        }
                                    }
                                });
                    } else {
                       **（关键2）** return Observable.just(deepSpaceBean);
                    }
                }
            });

关键1处，通过判断 发射源 的queryDP的返回值是否为空来区分执行不同的操作，
1为空 执行另一个mDPsRemoteDataSource.getDP(date)操作，retrofit的rx式返回

@Override
public Observable getDP(String date) {
    return PDORetrofit.getApiService().getAPOD(date);

}

    @GET("/planetary/apod?api_key=" + API_KEY)
Observable<DeepSpaceBean> getAPOD(@Query("date") String date);

 然后通过操作符doOnNext执行保存到本地数据库的操作（就是执行sql语句，返回为void，不关心结果）
关键处2，如果本地数据库查询（发射源）到了数据 deepSpaceBean != null，于是将它“包装”成observable重新发射出去。