多播粘性可观察到具有on subscription / on处置行为

我正在尝试创建Observable具有以下特征的：

1. 允许多个并发和/或连续的订户
2. 向每个新订户发出最后发出的项目
3. 当第一个订阅者订阅时，以及最后一个订阅被撤消时，它会执行某些操作

BehaviorSubject带有doOnSubscribe/的AdoOnDispose满足＃1和＃2，但为每个订户（而不是第一个和最后一个）运行订阅/处置。添加share满足＃1和＃3，但仅将最后发出的项目发射给第一个并发订户。

我想出了一个似乎可行的解决方案，但感觉就像一个丑陋的hack：

AtomicInteger subs = new AtomicInteger();
Observable<String> test = BehaviorSubject.createDefault("foo")
        .doOnSubscribe(x -> {
            if(subs.getAndIncrement() == 0) {
                // do something
            }
        })
        .doOnDispose(() -> {
            if(subs.decrementAndGet() == 0) {
                // do something
            }
        });

是否存在可以达到相同效果的现有运营商或运营商组合？