CacheObservableFactory.kt

package io.adev.rxwrapper

import io.reactivex.Observable
import io.reactivex.Scheduler
import io.reactivex.schedulers.Schedulers
import java.lang.System.currentTimeMillis

abstract class CacheObservableFactory<T, in C>(
        private val expireTime: Long = 2000,
        workScheduler: Scheduler = Schedulers.io()) : ObservableFactory<T, C>(workScheduler) {

    private inner class CacheEntity(
            val observable: Observable<T>,
            val createTime: Long,
            @Volatile var validCache: Boolean = true) {

        fun isValid() = createTime + expireTime > currentTimeMillis() && validCache
    }

    private val cache: MutableMap<C?, CacheEntity> = mutableMapOf()

    @Synchronized override fun create(criteria: C?): Observable<T> {
        clear()
        var entity = cache[criteria]
        if (entity == null || !entity.isValid()) {
            val observable = super.create(criteria)
                    .doOnNext { entity?.validCache = true }
                    .doOnComplete { entity?.validCache = true }
                    .doOnError { entity?.validCache = false }
                    .cache()
            entity = CacheEntity(
                    observable,
                    currentTimeMillis())
            cache[criteria] = entity
        }
        return cache[criteria]!!.observable
    }

    @Synchronized fun cache(observable: Observable<T>, criteria: C? = null) {
        cache[criteria] = CacheEntity(
                observable,
                currentTimeMillis())
    }

    @Synchronized fun clear() {
        val toDelete = cache.filter { (_, entity) -> !entity.isValid() }
        toDelete.onEach { (criteria, _) ->
            cache.remove(criteria)
        }
    }

    @Synchronized override fun invalidate(criteria: C?) {
        cache[criteria]?.validCache = false
        clear()
    }

    @Synchronized override fun invalidateAll() {
        cache.onEach { (_, entity) -> entity.validCache = false }
        clear()
    }

}