RxJava 2 chaîne complète de la force avec l'observable infini

Question

J'ai une chaîne observable qui a observable infini sur les observables supérieurs et non infinis après elle. Comme ceci:

repo.infinitGetItems() 
     .switchMap(items -> Observable 
          .just(items) 
          .flatMap(items -> repo.nonInfinitObs(items))); 

Ce que je veux est de compléter toute la chaîne quand repo.nonInfinitObs envoie événement onComplete. Maintenant, il ne se termine pas parce que repo.infinitGetItems() est jusqu'à la vie. Puis-je faire quelque chose comme force complète pour la chaîne entière dans rxJava2?

Merci d'avance!

Answer 1

Vous pouvez arrêter la séquence principale via takeUntil et des moyens d'écoulement externes:

PublishSubject<Integer> stop = PublishSubject.create(); 

repo.infinitGetItems() 
    .takeUntil(stop) 
    .switchMap(items -> repo.nonInfinitObs(items) 
         .doOnComplete(() -> stop.onComplete()) 
);