我目前正在尝试围绕 RxJava 进行研究，但是在以优雅的方式处理服务调用异常时遇到了一些麻烦。
基本上，我有一个（改造）服务，它返回一个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;
}
}
现在我想要的是将该通用响应映射到List<Account>包含在数据 JsonElement 字段中的 a （我假设您不关心Account对象的外观，所以我不会用它污染帖子）。以下代码非常适合成功案例，但我找不到处理 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);
有一个更好的方法吗？这确实有效，onError 会向我的观察者触发，我会收到我抛出的错误，但看起来我做的绝对不对。
提前致谢！
编辑：
让我澄清一下我想要实现的目标：
我想要一个可以从 UI 调用的类（例如，Activity、Fragment 或其他）。该类将采用 aObserver<List<Account>>作为参数，如下所示：
public Subscription loadAccounts(Observer<List<Account>> observer, boolean forceRefresh) {
...
}
该方法将返回一个订阅，当 UI 被分离/销毁/等时可以取消订阅。
参数化的观察者将处理 onNext 以处理传入帐户列表的成功响应。OnError 将处理任何异常，但也会传递任何 API 异常（例如，如果响应状态！= 200，我们将创建一个 Throwable 并将其传递给 onError）。理想情况下，我不想只是“抛出”异常，我想将它直接传递给观察者。这就是我看到的所有例子。
复杂之处在于我的改造服务返回一个ServiceResponse对象，所以我的观察者无法订阅它。我想出的最好的方法是在我的观察者周围创建一个观察者包装器，如下所示：
@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 的人在他们的演示文稿中真的向我推销了它，而且我非常渴望学习它。