RxJava 무한 관측 가능 2 무한 완료 체인

Question

무한 관측 대상이 무한 관측 가능하고 무한 관측 대상이 관측 가능 체인이 있습니다. 이처럼 :

는

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

내가 원하는 repo.nonInfinitObs는 onComplete를 이벤트를 보낼 때 모든 체인을 완성하는 것입니다. repo.infinitGetItems()가 살아있을 때까지 완료되지 않았습니다.

rxJava2의 전체 체인에 대해 강제 완료와 같은 작업을 수행 할 수 있습니까?

미리 감사드립니다.

Answer 1

당신은 takeUntil 어떤 흐름 외부 수단을 통해 주 계열을 중지 할 수 있습니다

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

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