Множество наблюдаемых одним подписчиком

Question

Мне нужно использовать события из нескольких источников. Количество источников варьируется, источник может быть динамически добавлен или удален.

Поскольку я хочу возможность менять набор источников "на лету", без необходимости повторной подписки, merge это не так.

Попробовал что-то вроде

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

sourcesSubject.subscribe(...);

Observable<Integer> source1 = ...;
source1.subscribe(sourcesSubject); // add first source

Observable<Integer> source2 = ...;
source2.subscribe(sourcesSubject); // add second source

..... // How to remove a source?

но понятия не имею, как удалить источник.

Есть ли простое решение?

Answer 1

Я бы предложил посмотреть на объединение операторов, как merge (и вариант варианта mergeWith) или же concat (concatWith) который устранит необходимость использования Subject, При отписке все источники будут остановлены.

Answer 2

Нашли самый простой способ удалить источник:

Observable<Integer> source = ...;
Subscription subscription = source.subscribe(sourcesSubject); // add source
subscription.unsubscribe(); // remove source