我目前正试图绕过RxJava，但我在处理服务调用异常方面遇到了一些麻烦 .
基本上，我有一个（Retrofit）服务返回 Observable<ServiceResponse> . ServiceResponse 的定义如下：
public class ServiceResponse {
private int status;
private String message;
private JsonElement data;
public JsonElement getData() {
return data;
}
public int getStatus() {
return status;
}
public String getMessage() {
return message;
}
}
现在我想要的是将该泛型响应映射到数据JsonElement字段中包含的 List<Account> （我假设你不关心 Account 对象是什么样的，所以我赢了't pollute the post with it). The following code works really well for the success case, but I can' t找到一个很好的方法来处理我的API异常：
service.getAccounts()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.map(new Func1<ServiceResponse, AccountData>() {
@Override
public AccountData call(ServiceResponse serviceResponse) {
// TODO: ick. fix this. there must be a better way...
ResponseTypes responseType = ResponseTypes.from(serviceResponse.getStatus());
switch (responseType) {
case SUCCESS:
Gson gson = new GsonBuilder().create();
return gson.fromJson(serviceResponse.getData(), AccountData.class);
case HOST_UNAVAILABLE:
throw new HostUnavailableException(serviceResponse.getMessage());
case SUSPENDED_USER:
throw new SuspendedUserException(serviceResponse.getMessage());
case SYSTEM_ERROR:
case UNKNOWN:
default:
throw new SystemErrorException(serviceResponse.getMessage());
}
}
})
.map(new Func1<AccountData, List<Account>>() {
@Override
public List<Account> call(AccountData accountData) {
Gson gson = new GsonBuilder().create();
List<Account> res = new ArrayList<Account>();
for (JsonElement account : accountData.getAccounts()) {
res.add(gson.fromJson(account, Account.class));
}
return res;
}
})
.subscribe(accountsRequest);
有一个更好的方法吗？这个 does 工作，onError会激发给我的观察者，我会收到我扔的错误，但看起来我似乎并没有这样做 .
提前致谢！
Edit:
让我澄清一下我想要实现的目标：
我希望有一个可以从UI调用的类（例如，Activity或Fragment，或其他） . 该类将 Observer<List<Account>> 作为参数，如下所示：
public Subscription loadAccounts(Observer<List<Account>> observer, boolean forceRefresh) {
...
}
当UI被分离/销毁/等时，该方法将返回可以取消订阅的订阅 .
参数化观察者将处理onNext，以便在Accounts列表中传递成功的响应 . OnError会处理任何异常，但也会传递任何API异常（例如，如果响应状态！= 200，我们将创建一个Throwable并将其传递给onError） . 理想情况下，我不想只是“抛出”异常，我想将它直接传递给Observer . 这就是我看到的所有例子 .
复杂的是我的Retrofit服务返回一个 ServiceResponse 对象，所以我的观察者无法订阅它 . 我提出的最好的方法是在Observer周围创建一个Observer包装器，如下所示：
@Singleton
public class AccountsDatabase {
private AccountsService service;
private List<Account> accountsCache = null;
private PublishSubject<ServiceResponse> accountsRequest = null;
@Inject
public AccountsDatabase(AccountsService service) {
this.service = service;
}
public Subscription loadAccounts(Observer<List<Account>> observer, boolean forceRefresh) {
ObserverWrapper observerWrapper = new ObserverWrapper(observer);
if (accountsCache != null) {
// We have a cached value. Emit it immediately.
observer.onNext(accountsCache);
}
if (accountsRequest != null) {
// There's an in-flight network request for this section already. Join it.
return accountsRequest.subscribe(observerWrapper);
}
if (accountsCache != null && !forceRefresh) {
// We had a cached value and don't want to force a refresh on the data. Just
// return an empty subscription
observer.onCompleted();
return Subscriptions.empty();
}
accountsRequest = PublishSubject.create();
accountsRequest.subscribe(new ObserverWrapper(new EndObserver<List<Account>>() {
@Override
public void onNext(List<Account> accounts) {
accountsCache = accounts;
}
@Override
public void onEnd() {
accountsRequest = null;
}
}));
Subscription subscription = accountsRequest.subscribe(observerWrapper);
service.getAccounts()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(accountsRequest);
return subscription;
}
static class ObserverWrapper implements Observer<ServiceResponse> {
private Observer<List<Account>> observer;
public ObserverWrapper(Observer<List<Account>> observer) {
this.observer = observer;
}
@Override
public void onCompleted() {
observer.onCompleted();
}
@Override
public void onError(Throwable e) {
observer.onError(e);
}
@Override
public void onNext(ServiceResponse serviceResponse) {
ResponseTypes responseType = ResponseTypes.from(serviceResponse.getStatus());
switch (responseType) {
case SUCCESS:
Gson gson = new GsonBuilder().create();
AccountData accountData = gson.fromJson(serviceResponse.getData(), AccountData.class);
List<Account> res = new ArrayList<>();
for (JsonElement account : accountData.getAccounts()) {
res.add(gson.fromJson(account, Account.class));
}
observer.onNext(res);
observer.onCompleted();
break;
default:
observer.onError(new ApiException(serviceResponse.getMessage(), responseType));
break;
}
}
}
}
我仍然觉得我没有正确使用这个 . 我以前没有见过其他人使用ObserverWrapper . 也许我不应该使用RxJava，虽然SoundCloud和Netflix的人员在他们的演示文稿中真的把它卖给了我，我非常渴望学习它 .
2 回答
请阅读下面我添加了一个编辑 .
使用RxJava抛出Action / Func / Observer是完全正确的 . 该异常将由框架传播到您的Observer . 如果你只限于调用onError，那么你就会扭曲自己来实现这一点 .
据说这是一个建议，就是简单地删除这个包装器，并在Service.getAccount ...的Observables链中添加一个简单的验证Action .
我将使用与 Map 链接的doOnNext（new ValidateServiceResponseOrThrow）（新的MapValidResponseToAccountList） . 这些是简单的类，它们实现了必要的代码，以使Observable链更具可读性 .
这是使用我建议简化的loadAccount方法 .
Edit: 除非有人另有说法，否则我认为过去曾引起过例行动，但我不推荐这样做.1717241_
如果使用flatMap，您将拥有更清晰的Exception堆栈 . 如果从Action内部抛出，Exception堆栈实际上将包含
rx.exceptions.OnErrorThrowable$OnNextValue异常，这是不理想的 .让我用flatMap演示上面的例子 .
正如你所看到的，差异是微妙的 .
ValidateServiceResponse现在实现Func1而不是Action1，我们不再使用throw关键字 . 我们改用Observable.error(new Throwable). 我相信这与预期的Rx合约更合适 .你可以阅读这篇关于错误处理的好文章http://blog.danlew.net/2015/12/08/error-handling-in-rxjava/