RxConnectable playground

main.kt

fun main() {
    val shared = Observable.interval(1, TimeUnit.SECONDS)
        .doOnSubscribe { println("Subscribed.") }
        .doOnDispose { println("Disposed.") }
        .replay() // or .publish()
        .autoConnect() // or .refcount()

    Thread {
        shared
            .take(3)
            .subscribe { println("A: $it") }
    }.start()
    Thread {
        shared
            .take(6)
            .subscribe { println("B: $it") }
    }.start()
    Thread {
        Thread.sleep(3_000)
        shared
            .take(6)
            .subscribe { println("C: $it") }
    }.start()
    Thread {
        Thread.sleep(7_000) // Starts after A, B and C finish - in case of refCount, resubscribes original observable, in case of autoconnect - not
        shared
            .take(6)
            .subscribe { println("D: $it") }
    }.start()

    Thread.sleep(20_000) // Wait till everything finish.
}