java - RxJava中响应式(Reactive)和命令式代码之间的错误处理

给定一个接口(interface):

public interface FastlyRxApi {

    @GET("/service/{service_id}/version/{version}/backend")
    Observable<List<Backend>> listBackends(@Path("service_id") String serviceId, @Path("version") String versionId);

    @PUT("/service/{service_id}/version/{version}/backend/{old_name}")
    Observable<Backend> updateBackend(@Path("service_id") String serviceId, @Path("version") String version, @Path("old_name") String oldName, @Body Backend updatedBacked);

}

和一些客户端代码:

Integer expectedFirstByteTimeout = 10000;

// Use a final array to capture any problem found within our composed Observables
final FastlyEnvException[] t = new FastlyEnvException[1];

fastlyRxApi.listBackends(serviceId, newVersion)
    .flatMap(Observable::fromIterable)
    .filter(backend -> !expectedFirstByteTimeout.equals(backend.getFirstByteTimeout()))
    .flatMap(backend -> {
        backend.setFirstByteTimeout(expectedFirstByteTimeout);
        return fastlyRxApi.updateBackend(serviceId, newVersion, backend.getName(), backend);
    }).subscribe(ignore -> {
}, e -> {
    t[0] = new FastlyEnvException("failed to configure backends", e);
});

if (t[0] != null) {
    throw t[0];
}

使用FastlyEnvException的最终数组来捕获上下文以进行错误处理，感觉就像我做错了事，并且缺少某些方面。

我在这里使用锤子而不是 Screwdriver 吗？即我应该为此使用RxJava吗？除了错误处理之外，这似乎给了我很好的可读性。首选的惯用法是什么？

最佳答案

使用onErrorResumeNext:

.onErrorResumeNext(err ->
     Observable.error(new FastlyEnvException("failed to configure backends", e)))
.toBlocking();
.subscribe();

值得注意的是.toBlocking()，这将使Observable链等待其完成。

鉴于subscription()没有错误处理程序，它将重新引发异常。