android - 如何在 RxJava 中刷新 observable？

Question

我有一个 observable，它包装了一个 HTTP 请求

mObservable =  retryObservable(mService.getAddressList(getUserId(), true, 1, Integer.MAX_VALUE, "id", true)
            .map(r -> {
                return r.getItems();
            })
            .observeOn(AndroidSchedulers.mainThread()));

然后订阅

mSubscription = mObservable.subscribe(items -> {
                mAddressAdapter.swapItems(items);
            }, getActivityBase()::showError);

当订阅初始化到来时，cold observable 被激活并触发 HTTP 请求。现在，我知道基础数据已更改，我需要提出相同但新的请求。我试过了

mSubscription.unsubscribe();

然后调用

mObservable.subscribe(items -> {doSomething();})

同样，根据我的理解，订阅应该会触发 observable，但它不起作用。有什么建议吗？

Answer 1

一旦 Observable 完成，它就不会发布任何新项目。这是 Rx 契约(Contract)。

将您的代码包装到一个方法中，每次都创建一个新的可观察对象。

Observable<?> getObservable() {
    return retryObservable(mService.getAddressList(getUserId(), true, 1, Integer.MAX_VALUE, "id", true)
            .map(r -> {
                return r.getItems();
            })
            .observeOn(AndroidSchedulers.mainThread()));
}

正如@DaveSexton 在评论中提到的，在 RxJava 中使用 defer 函数有更好的方法

Do not create the Observable until a Subscriber subscribes; create a fresh Observable on each subscription

Pass defer( ) an Observable factory function (a function that generates Observables), and defer( ) will return an Observable that will call this function to generate its Observable sequence afresh each time a new Subscriber subscribes.

更多信息:https://github.com/ReactiveX/RxJava/wiki/Creating-Observables#defer