Kotlin LazyTimedCache Delegate

LazyTimedCache.kt

import io.reactivex.disposables.CompositeDisposable
import io.reactivex.disposables.Disposable
import io.reactivex.schedulers.Schedulers
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicReference
import kotlin.reflect.KProperty

/**
 * Property delegate that retrieves and caches a value for a duration
 * But will expire the cache after that duration, and re-retrieve the value on the next get()
 */
fun <T> lazyTimedCache(duration: Long,
                       timeUnit: TimeUnit,
                       compositeDisposable: CompositeDisposable? = null,
                       resetOnGet: Boolean = false,
                       getter: () -> T) = LazyTimedCache(duration,timeUnit,compositeDisposable, resetOnGet, getter)

class LazyTimedCache<T>(val duration: Long,
                        val timeUnit: TimeUnit,
                        val compositeDisposable: CompositeDisposable? = null,
                        val resetOnGet: Boolean,
                        val getter: () -> T) {

    private object UNINITIALIZED_VALUE

    private var _value: Any? = UNINITIALIZED_VALUE
    private val doRefresh = AtomicBoolean(true)
    private val currentTask = AtomicReference<Disposable?>()

    private fun generateTask() = Schedulers.computation().scheduleDirect({
        doRefresh.set(true)
        _value = null
        if (resetOnGet) currentTask.set(null)
    }, duration, timeUnit)


    operator fun getValue(thisRef: Any?, property: KProperty<*>): T {
        if (resetOnGet) currentTask.getAndSet(null)?.dispose()

        if (doRefresh.compareAndSet(true,false)) {
            _value = getter()
            doRefresh.set(false)
            val disposable = generateTask()
            compositeDisposable?.add(disposable)
            if (resetOnGet) {
                currentTask.set(disposable)
            }
        }
        @Suppress("UNCHECKED_CAST")
        return _value as T
    }
}