Reactive Repository

Repo.kt

class Repo<in Key : Any, Value : Any>(
    private val dataSource: DataSource<Key, Value>,
    private val persister: Persister<Key, Value>
) {

  private val cleared = PublishSubject.create<Key>()
  private val cache = HashMap<Key, Flowable<Value>>()

  fun stream(key: Key): Flowable<Value> =
      cache.getOrPut(key) { newStream(key) }

  fun clear(key: Key): Completable =
      persister.delete(key)
          .doOnComplete { cleared.onNext(key) }

  private fun newStream(key: Key): Flowable<Value> =
      persister.get(key)
          .toFlowable()
          .flatMap {
            when (it) {
              is Result.Fresh -> Flowable.just(it.value)
              is Result.Stale -> Single.just(it.value).concatWith(getAndStore(key))
              is Result.Missing -> getAndStore(key).toFlowable()
            }
          }
          .repeatWhen { it.switchMap { cleared.toFlowable(BackpressureStrategy.LATEST) } }
          .share()

  private fun getAndStore(key: Key): Single<Value> =
      dataSource.get(key)
          .flatMap { value ->
            persister.set(key, value)
                .toSingleDefault(value)
          }
}

interface DataSource<in Key : Any, Value : Any> {
  fun get(key: Key): Single<Value>
}

interface Persister<in Key : Any, Value : Any> {
  fun get(key: Key): Single<Result<Value>>
  fun set(key: Key, value: Value): Completable
  fun delete(key: Key): Completable
}

@Suppress("unused")
sealed class Result<Value : Any> {
  data class Fresh<Value : Any>(val value: Value) : Result<Value>()
  data class Stale<Value : Any>(val value: Value) : Result<Value>()
  class Missing<Value : Any> : Result<Value>()
}