java - 如何使用 ReplaySubject 重新运行 CompositeSubscription？

Question

我有 CompositeSubscription ，并在那里添加带有 ReplaySubject 的 Subscription

 CompositeSubscription compositeSubscription = new CompositeSubscription();
    ReplaySubject subject = ReplaySubject.create();

 compositeSubscription.add(
                manager.getAllContacts()
                .toList()
                .doOnNext(new Action1<List<Person>>() {
                    @Override
                    public void call(List<Person> persons) {
                        allPersons = persons;
                        Log.e(TAG, "BookContacts: " + "allPersons = " + allPersons.size());
                        setupViewPager();
                    }
                })
                .subscribe(subject));

然后我用这个 ReplaySubject 添加第二个订阅

compositeSubscription.add(Observable.combineLatest(subject,
                                                           (PublishSubject<List<CustomUser>>) execute(
                                                                   manager.getDigitsContacts()),
                                                           new Func2<List<Person>, List<CustomUser>, Object>() {
                                                               @Override
                                                               public Object call(List<Person> persons, List<CustomUser> customUsers) {
                                                                  //... my code with persons and customUsers...
                                                                   return null;
                                                               }
                                                           })
                                            .subscribe());

代码正常运行，在完成 ReplaySubject hasCompleted = true 之后。

但是当我尝试添加第三个 Subscription 时，它不会调用“call()”方法

compositeSubscription.add(Observable.combineLatest(subject,
                                                           (PublishSubject<List<CustomUser>>) execute(
                                                                   manager.getFacebookContacts()),  //<-----manager.getFacebookContacts() is run, but doesn't call call() method
                                                           new Func2<List<Person>, List<CustomUser>, Object>() {
                                                               @Override
                                                               public Object call(List<Person> persons, List<CustomUser> customUsers) {
                                                                  //...this method is not called after  "manager.getFacebookContacts()"
                                                                   return null;
                                                               }
                                                           })
                                            .subscribeOn(Schedulers.newThread())
                                            .subscribe());

如何解决？...因为如果我同时添加订阅，它就可以正常工作。

Answer 1

能否请您将错误回调添加到.subscribe()？我的猜测是第三次 ReplaySubject 溢出了 combineLatest 的缓冲区。您应该使用 .replay().autoConnect(0)

而不是创建主题

CompositeSubscription compositeSubscription = new CompositeSubscription();

Observable<List<Person>> persons = manager.getAllContacts()
    .toList()
    .doOnNext(new Action1<List<Person>>() {
        @Override
        public void call(List<Person> persons) {
            allPersons = persons;
            Log.e(TAG, "BookContacts: " + "allPersons = " + allPersons.size());
            setupViewPager();
        }
    }).replay().autoConnect(0, s -> compositeSubscription.add(s));

然后使用persons代替subject