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

我有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());

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

最佳答案

您能否将错误回调添加到.subscribe()？我的猜测是，第三次ReplaySubject溢出CombineLatest的缓冲区。代替创建主题，应该使用.replay().autoConnect(0)

CompositeSubscription CompositeSubscription =新的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

关于java - 如何使用ReplaySubject重新运行CompositeSubscription？，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/33611142/