Flowable Clock: a Clock that switches its behavior whenever a new item is emitted by a given Flowable

FlowableClock.kt

import io.reactivex.Flowable
import io.reactivex.disposables.Disposable
import org.threeten.bp.Clock
import org.threeten.bp.Instant
import org.threeten.bp.ZoneId
import kotlin.properties.Delegates

/**
 * A [Clock] that acts as the latest clock emitted so far by the given Flowable.
 *
 * It is assumed that the given Flowable will have emitted at least one item before any Clock
 * method is called on this instance; if these methods are called before the first emission,
 * an [IllegalStateException] will be thrown.
 */
class FlowableClock(clocks: Flowable<Clock>) : Clock(), Disposable {
    private var currentClock: Clock by Delegates.notNull()

    private val clocks = clocks.replay(1).autoConnect()

    private val subscription: Disposable = clocks.subscribe { clock ->
        currentClock = clock
    }

    // Clock

    override fun withZone(newZone: ZoneId): Clock =
        FlowableClock(clocks.map { it.withZone(newZone) })

    override fun getZone(): ZoneId = currentClock.zone

    override fun instant(): Instant = currentClock.instant()

    // Disposable

    override fun isDisposed(): Boolean = subscription.isDisposed

    override fun dispose() = subscription.dispose()
}