android - 用 RxJava 替换监听器

Question

目前，我正在研究向 RxJava 的迁移，并认为我的经理(accountManager)将是一个有趣的起点。目前，管理器有一个监听器列表，并在帐户更新和出现问题时相应地发送更新。

private List<WeakReference<ProfileChangeListener>> mListeners = new ArrayList<>();

public interface ProfileChangeListener {
    void onProfileUpdated(Account account);
    void onProfileFailed(Exception e);
}

我的 Rx 解决方案涉及一个主题

private SerializedSubject<Account, Account> mManagerSubject = new SerializedSubject<>(BehaviorSubject.<Account>create());

public Observable<Account> observe() {
    return mManagerSubject;
}

然后当更新发生时，我调用以下方法之一:

private void onProfileUpdated(Account account) {
    mManagerSubject.onNext(account);
}

private void onProfileFailed(final Exception e) {
    mManagerSubject.onError(e);
}

问题

问题是，一旦 onError 被调用，任何通过 observe 监听的人将永远不会从 onNext 获得另一个更新。

我仍然希望订阅者接收 onError，以便他们可以处理错误状态，但稍后仍可以使用更新的帐户调用 onNext 并且我仍然希望订阅者处理更新的帐户。

我尝试过使用 onErrorResumeNext、onErrorReturn onExceptionResumeNext 的解决方案，但它们都没有传播 onError。

TLDR:在调用 onError 后如何保持订阅者订阅，同时仍然传播 onError？

Answer 1

Rx 中的“错误”一开始可能有点难以理解，因为它们的含义与大多数人的预期略有不同。

来自Error Handling documentation (强调我的):

An Observable typically does not throw exceptions. Instead it notifies any observers that an unrecoverable error has occurred by terminating the Observable sequence with an onError notification.

onError() 应该在 Observable 遇到不可恢复错误时使用 - 即当您的 Observable 时code> 无法继续发射元素。当您订阅时，您可能会使用 onErrorResumeNext 之类的方法来尝试一些恢复操作，但这应该是源 Observable 的结尾。

相反，您可能需要调整 Observable 发出的内容以支持发出错误项，或者包含一个指示遇到错误的标志。

如果您的错误确实无法恢复，那么您可能需要重新审视您的恢复策略并尝试稍微不同的方法。