如何连续执行许多 RxJava2 通量

Question

我正在介绍自己关于RxJava2，但我觉得我做错了什么。就我而言，我想执行以下一些异步操作。

在这个例子中，第一个动作是检查设备是否已连接(wifi或数据，让我们承认这需要时间(，然后我想连接到一个api，然后我想做一个http调用来获取一个列表(可观察(，然后使用它。如果其中一个操作失败，则应在订阅中引发并处理 onError 或异常。

我有这个代码谁工作：

Single.create((SingleEmitter<Boolean> e) -> e.onSuccess(Connectivity.isDeviceConnected(MainActivity.this)) )
.subscribeOn(Schedulers.io())
.flatMap(isDeviceConnected -> {
Log.i("LOG", "isDeviceConnected : "+ isDeviceConnected);
if(!isDeviceConnected)
throw new Exception("whatever"); // TODO : Chercher vrai erreur
return awRepository.getFluxAuthenticate(host, port, user, password); // Single<DisfeApiAirWatch>
})
.toObservable()
.flatMap(awRepository::getFluxManagedApps)  // List of apps : Observable<AirwatchApp>
.observeOn(AndroidSchedulers.mainThread())
.doFinally(this::hideProgressDialog)
.subscribe(
app -> Log.i("LOG", "OnNext : "+ app),
error -> Log.i("LOG", "Error : " + error),
() -> Log.i("LOG", "Complete : ")
);

但是做一个为一个简单的"如果"而用布尔值的单曲听起来是错误的。一个可比拟表似乎更合乎逻辑(工作或不工作，继续或停止(。我尝试使用以下代码，但它不起作用。

Completable.create((CompletableEmitter e) -> {
if(Connectivity.isDeviceConnected(MainActivity.this))
e.onComplete(); // Guess not good, should call the complete of subscribe ?
else
e.onError(new Exception("whatever"));
} ).toObservable()
.subscribeOn(Schedulers.io())
.flatMap(awRepository.getFluxAuthenticate(host, port, user, password)) //Single<DisfeApiAirWatch>
.toObservable()
.flatMap(awRepository::getFluxManagedApps) // List of apps : Observable<AirwatchApp>
.observeOn(AndroidSchedulers.mainThread())
.doFinally(this::hideProgressDialog)
.subscribe(
app -> Log.i("LOG", "OnNext : "+ app),
error -> Log.i("LOG", "Error : " + error),
() -> Log.i("LOG", "Complete : ")
);

如何使这段代码工作？

我知道我可以在可兼容对象上进行第一次订阅，并在这个的"onSuccess"中编写另一个通量/其余代码。但我不认为彼此内部的堆栈流是一个很好的解决方案。

此致敬意

Answer 1

Completable

没有值，因此永远不会调用flatMap。您必须使用andThen并使身份验证成功值成为后续flatMap的输入：

Completable.create((CompletableEmitter e) -> {
if(Connectivity.isDeviceConnected(MainActivity.this))
e.onComplete();
else
e.onError(new Exception("whatever"));
})
.subscribeOn(Schedulers.io())
.andThen(awRepository.getFluxAuthenticate(host, port, user, password)) // <-----------
.flatMapObservable(awRepository::getFluxManagedApps)
.observeOn(AndroidSchedulers.mainThread())
.doFinally(this::hideProgressDialog)
.subscribe(
app -> Log.i("LOG", "OnNext : "+ app),
error -> Log.i("LOG", "Error : " + error),
() -> Log.i("LOG", "Complete : ")

(;