class ObservableBus<T>(cacheCount: Int = 0) {

    private val subject: SerializedSubject<Observable<T>, Observable<T>> = PublishSubject<Observable<T>>().toSerialized()
    private val live: MutableSet<Observable<T>> = ConcurrentHashMap.newKeySet<Observable<T>>()

    private val observable = subject.flatMap { obs: Observable<T> -> obs.takeWhile { live.contains(obs) } }
            .observeOnFx().let { if (cacheCount > 0) it.cacheWithInitialCapacity(cacheCount) else it }

    fun toObservable(): Observable<T> = observable

    operator fun plusAssign(observable: Observable<T>) = add(observable)
    operator fun minusAssign(observable: Observable<T>) = remove(observable)

    fun add(observable: Observable<T>) {
        live.add(observable)
        subject.onNext(observable)
    }
    fun remove(observable: Observable<T>) {
        live.remove(observable)
    }
}