class MyApp: App(MyView::class)

class MyView : View("My View") {


    override val root = vbox {

        val source = Observable.interval(1, TimeUnit.SECONDS).publish()

        val isSubscribed = AtomicBoolean(false)

        val cacheList = ArrayList<Long>()

        //cache the emissions
        source.takeWhile { !isSubscribed.get() }
                .subscribe { cacheList.add(it) }

        //merge the cache with emissions going forward
        val merged = Observable.merge(
                cacheList.toObservable().doOnCompleted { cacheList.clear() },
                source
        )

        //connect the ConnectableObservable
        source.connect()

        Thread.sleep(7000) //wait seven seconds

        //bring in Subscriber
        merged.doOnSubscribe { isSubscribed.set(true) }.subscribe { println(it) }
    }
}