FlatMap и темы - как избежать дублирования подписчиков?

Question

Скажем, у меня есть следующий код:

BehaviorSubject<Integer> subject = BehaviorSubject.create(); 
BehaviorSubject<Integer> subject2 = BehaviorSubject.create(); 
subject. 
     doOnNext(number -> subject2.onNext(number)). 
     flatMap(number -> subject2). 
     subscribe(number -> System.out.println("Number " + number)); 

for(int i = 0; i < 10; i++) { 
    subject.onNext(i); 
} 

Выход он производит:

Number 0 
Number 1 
Number 1 
Number 2 
Number 2 
Number 2 
Number 3 
Number 3 
Number 3 
Number 3 
Number 4 
Number 4 
Number 4 
Number 4 
Number 4 

И дальше и дальше.

Я понимаю, почему это происходит - это потому, что каждый раз, когда вызывается flatMap, новый абонент добавляется к subject2.

Вопрос в том, как избежать этого?

Answer 1

Вы можете ограничить subject2 к одному элементу внутри flatMap как это:

flatMap(number -> subject2.first())

Код

BehaviorSubject<Integer> subject = BehaviorSubject.create(); 
    BehaviorSubject<Integer> subject2 = BehaviorSubject.create(); 
    subject. 
      doOnNext(number -> subject2.onNext(number)). 
      flatMap(number -> subject2.first()). 
      subscribe(number -> System.out.println("Number " + number)); 

    for(int i = 0; i < 10; i++) { 
     subject.onNext(i); 
    } 

производит вывод, который не содержит дубликатов

Answer 2

попробовать switchMap вместо flatMap.

switchMap только подписка на последние Observable получает. Он отписывается от всех ранее подписанных Observable с.

http://reactivex.io/RxJava/javadoc/rx/Observable.html#switchMap