使用 RxJava/RxAndroid 依次执行不同类型的 Observable

Question

我有一个远程调用（改造） - 我将其转换为 Observable。我们称它为 Observable Y。 现在，我还有一段代码可以通过 GPS 和 NETWORK 提供商查找地理位置。我在那里有一个计时器，它基本上限制了可以执行地理搜索的时间。我们称它为 Task X。我想将其转换为 Observable X。

然后，我想要一个订阅，它将执行 Observable X（即查找位置），一旦它返回一个位置，我将以某种方式“分析”，然后我将传递该位置进入 Observable Y（改造调用），或者干脆退出（如果那个“原始”位置在我的情况下就足够了）

在任何时候，我都希望能够中断所有“过程”。根据我收集到的信息，我可以通过简单地取消订阅来实现这一点，对吧？ 然后下次再订阅这个订阅。

问题：

1. Can all of that be implemented via RxJava/RxAndroid ?
2. Does it even make sense implementing it with Rx ? or is there a more efficient way?
3. How is it done with Rx? 
(More specifically : (a) How do I convert task Y into an Observable Y?
                     (b) How do I perform them in sequence with only one subscription?)

Answer 1

1- 可以通过 RxJava 实现

2- 这是你目前最好的选择

3-

3-a Observable.fromCallable() 可以解决问题

3-b flatmap 运算符用于链接可观察调用 你可以这样继续：

private Location searchForLocation() {
    // of course you will return not null location
    return null;
}

// your task X
//mock your location fetching
private Observable<Location> getLocationObservableX() {
    return Observable.fromCallable(() -> searchForLocation());
}

//your task Y
//replace CustomData with simple String
//just to mock your asynchronous retrofit call
private Observable<List<String>> getRetrofitCallObservableY(String param){
    return Observable.just(new ArrayList<String>());
}


//subscribe
private void initialize() {
    getLocationObservableX()
            .filter(location -> {
                //place your if else here
                //condition
                //don't continue tu retrofit
                boolean condition = false;

                if (condition) {
                    //process
                    //quit and pass that Location in Broadcas
                    //you shall return false if you don't want to continue
                    return false;
                }
                return true;
            })
            //filter operation does not continue here if you return false
            .flatMap(location -> getRetrofitCallObservableY("param"))
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(response -> {
                //do what you want with response
            });
}