rx-java - 如何在运行时观察到 "merge"？

Question

我的用例是 View 组中的点击监听器:

• 每个 View发布感冒Observable<Click> API。

• View 组管理 View 集合，并提供 Observable<ViewClick>从 subview 发出点击的 API。

我希望每个 subview 在添加到复合可观察对象时都被订阅，但前提是复合对象有订阅。如果不是，则应仅在订阅组合时订阅子项。

到目前为止，我的解决方案是:

PublishSubject<ViewClick> composite;
Map<View, Subscription> subscriptions;

void addSource(View view, Observable<Click> clicks) {
    Subscription s = clicks.map(click -> new ViewClick(view, click)
     .subscribe(composite::onNext);
    subscriptions.put(view, s);
}

void removeSource(View v) {
    Subscription s = subscriptions.get(v);  
    s.unsubscribe;
    subscriptions.remove(v);  
}

Observable<ViewClick> compositeClicks() {
    return composite;
}

但是，这意味着 clicks observables 不再是冷的，因为无论是否订阅了 compositeClicks，它们都会被订阅.

有没有更好的办法？

Answer 1

您需要一个特殊的“热”Subject，例如 BufferUntilSubscriber，它保持源 Observables 直到单个消费者订阅它。不幸的是，该类不是官方 API 的一部分，但是，2.0 将有一个官方的 UnicastSubject 用于同样的目的:

UnicastSubject<Observable<Integer>> inputs = UnicastSubject.create();

Observable<String> p = inputs
        .flatMap(v -> v.map(u -> u.toString()))
        .publish()
        .autoConnect()
        ;

inputs.onNext(Observable.just(1)
    .doOnSubscribe(s -> System.out.println("Subscribed to 1")));
inputs.onNext(Observable.just(2)
    .doOnSubscribe(s -> System.out.println("Subscribed to 2")));
inputs.onNext(Observable.just(3)
    .doOnSubscribe(s -> System.out.println("Subscribed to 3")));

System.out.println("Subscribing to p");

p.subscribe(System.out::println);