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

        val isSubscribed = AtomicBoolean(false)

        //cache the emissions
        val cached = source.takeWhile { !isSubscribed.get() }
                .toList()
                .cacheWithInitialCapacity(1)


        //merge the cache with emissions going forward
        val merged = Observable.merge(
                cached.flatMap { it.toObservable() },
                source
        )

        //kick off cache immediately by subscribing to it
        cached.subscribe()

        //connect the ConnectableObservable
        source.connect()

        Thread.sleep(7000) //wait seven seconds

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