我目前正在尝试全神贯注于 RxJava，但我在以优雅的方式处理服务调用异常方面遇到了一些麻烦。

基本上，我有一个返回 Observable<ServiceResponse> 的 (Retrofit) 服务. 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 字段中(我假设您不关心 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 或其他)。该类(class)需要 Observer<List<Account>>作为这样的参数:

public Subscription loadAccounts(Observer<List<Account>> observer, boolean forceRefresh) {
    ...
}

该方法将返回一个订阅，当 UI 分离/销毁/等时可以取消订阅。

参数化的观察者将处理 onNext 以传递一个帐户列表的成功响应。 OnError 会处理任何异常，但也会传递任何 API 异常(例如，如果响应状态 != 200，我们将创建一个 Throwable 并将其传递给 onError)。理想情况下，我不想只是“抛出”异常，我想将它直接传递给观察者。这就是我看到的所有示例所做的。

复杂的是我的 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 的人在他们的演示中确实向我推销了它，我非常渴望学习它。

最佳答案

请阅读下面我添加的编辑。

用 RxJava 将 Action/Func/Observer 扔进去是完全正确的。异常将由框架传播到您的观察者。 如果您限制自己只调用 onError，那么您将扭曲自己来实现这一点。

话虽如此，一个建议是简单地删除这个包装器并添加一个简单的验证 service.getAccount 中的操作... Observables 链。

我会使用与 map (new MapValidResponseToAccountList) 链接的 doOnNext(new ValidateServiceResponseOrThrow)。这些是简单的类，它们实现了必要的代码以保持 Observable 链更具可读性。

这是使用我的建议简化的 loadAccount 方法。

public Subscription loadAccounts(Observer<List<Account>> observer, boolean forceRefresh) {
    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(observer);
    }

    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 EndObserver<List<Account>>() {

        @Override
        public void onNext(List<Account> accounts) {
            accountsCache = accounts;
        }

        @Override
        public void onEnd() {
            accountsRequest = null;
        }
    });

    Subscription subscription = accountsRequest.subscribe(observer);

    service.getAccounts()
            .doOnNext(new ValidateServiceResponseOrThrow())
            .map(new MapValidResponseToAccountList())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(accountsRequest);

    return subscription;
}

private static class ValidateResponseOrThrow implements Action1<ServiceResponse> {
        @Override
        public void call(ServiceResponse response) {
            ResponseTypes responseType = ResponseTypes.from(serviceResponse.getStatus());
            if (responseType != SUCCESS)
                throw new ApiException(serviceResponse.getMessage(), responseType));
        }
    }

private static class MapValidResponseToAccountList implements Func1<ServiceResponse, List<Account>> {
    @Override
    public Message call(ServiceResponse response) {
        // add code here to map the ServiceResponse into the List<Accounts> as you've provided already
    }
}

编辑: 除非有人另有说明，否则我认为最好使用 flatMap 返回错误。 我过去曾抛出 Exceptions from Action，但我认为这不是推荐的方式。

如果您使用 flatMap，您将拥有一个更干净的异常堆栈。如果您从 Action 内部抛出异常堆栈 实际上将包含不理想的 rx.exceptions.OnErrorThrowable$OnNextValue 异常。

让我用 flatMap 来演示上面的例子。

private static class ValidateServiceResponse implements rx.functions.Func1<ServiceResponse, Observable<ServiceResponse>> {
    @Override
    public Observable<ServiceResponse> call(ServiceResponse response) {
        ResponseTypes responseType = ResponseTypes.from(serviceResponse.getStatus());
        if (responseType != SUCCESS)
            return Observable.error(new ApiException(serviceResponse.getMessage(), responseType));
        return Observable.just(response);
    }
}

service.getAccounts()
    .flatMap(new ValidateServiceResponse())
    .map(new MapValidResponseToAccountList())
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(accountsRequest);

如您所见，两者之间的区别很细微。 ValidateServiceResponse 现在实现了 Func1 而不是 Action1 并且我们不再使用 throw 关键字。我们使用 Observable.error(new Throwable) 代替。我相信这更符合预期的 Rx 合约。

关于android - 在 RxJava 中处理 API 异常，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/24243410/