我如何创建订阅可观察对象的 BehaviorSubject？

Question

我有一个函数应该 return 一个 BehaviorSubject。该主题旨在 return Profile

的最新版本

(user)Profile 只是一个包含对三个成员的引用的 POJO：
- User,
- 该用户的 MeasurementList、
- 还有一个 Deadline。

其中两个属性是通过改造调用获得的，其中一个已经保存在 class 变量中。

每当 observable 发出新的 measurement list 或 deadline 时，BehaviorSubject 应该发出新的更新配置文件。

这是应该发生的事情的图表（希望有用）

这是我目前所拥有的

 public BehaviorSubject<Profile> observeProfile() {

        if (profileBS == null) {

            profileBS = BehaviorSubject.create();

            Observable o = Observable.combineLatest(
                    Observable.just(userAccount),
                    observeMeasurements(),
                    observeDeadline(),
                    Profile::new
            );


            profileBS.subscribeTo(o); //subscribeTo does not exist, but this is what I am trying to figure out how to do.

        }
        return profileBS;
    }

谁能帮我正确创建这个 BehaviorSubject？

谢谢。

Answer 1

Subject 实现了 Observer 接口，因此您可以执行以下操作

public BehaviorSubject<Profile> observeProfile() {

        if (profileBS == null) {

            profileBS = BehaviorSubject.create();

            Observable o = Observable.combineLatest(
                    Observable.just(userAccount),
                    observeMeasurements(),
                    observeDeadline(),
                    Profile::new
            );

            // Note the following returns a subscription. If you want to prevent leaks you need some way of unsubscribing.
            o.subscribe(profileBS); 

        }
        return profileBS;
}

请注意，您应该想出一种方法来处理生成的订阅。

Answer 2

我想我和你面临着同样的问题。在我给出答案之前，我想先讲一些历史。所以我想很多人都看到了 JakeWharton 在 Devoxx 上的演讲：The State of Managing State with RxJava

所以他想出了一个基于一些变压器的架构。但问题是，即使是那些 Transformer 实例也会在您的 ViewObservables 之外生存。每次您的 ViewObservable 使用它们时，它们仍然会创建新的 Observable。是因为operator的概念。

因此，一个常见的解决方案是像您在问题中所做的那样使用主题作为网关。但新的问题是你太早订阅了你的来源。

订阅操作应在 subscribeActual() 中完成，这将由下游的 subscribe() 方法触发。但是你在中间订阅了你的上游。你失去了你的事件。我为解决这个问题而苦苦挣扎，但从未找到解决方案。

但最近感谢 Davik 的回答：我想到了这样的事情：

public BehaviorSubject<Profile> observeProfile() {
    public Observable resultObservable;
    if (profileBS == null) {

        profileBS = BehaviorSubject.create();
        //this source observable should save as a field as well 
        o = Observable.combineLatest(
                Observable.just(userAccount),
                observeMeasurements(),
                observeDeadline(),
                Profile::new
        );
        //return this observable instead of your subject
        resultObservable = profileBS.mergeWith(
            Observable.empty()
                      .doOnComplete(() -> {
                          o.subscribe(profileBS);
                      }))
    } return resultObservable;
}

诀窍就在于此。您使用此方法来制作像 doAfterSubscribe 这样的运算符。所以只有下游已经订阅了你的主题。您的主题将订阅您的原始上游资源。

希望这对您有所帮助。抱歉我的英语不好。

我如何创建订阅可观察对象的 BehaviorSubject？

How can I create a BehaviorSubject that subscribes to an observable?

java

android

rx-java

retrofit

reactivex