A weird version of RxStore that uses Kotlin and Moshi

RxStore.kt

import android.content.Context
import android.os.FileObserver
import com.squareup.moshi.JsonAdapter
import com.squareup.moshi.Moshi
import com.squareup.moshi.Types
import okio.Okio
import rx.Observable
import rx.Scheduler
import rx.lang.kotlin.observable
import rx.lang.kotlin.single
import rx.schedulers.Schedulers
import java.io.File
import java.util.concurrent.Executors

class RxStore(context: Context, directoryName: String, val moshi: Moshi) {
  val scheduler = Schedulers.from(Executors.newSingleThreadExecutor())
  val directory = context.getDir(directoryName, Context.MODE_PRIVATE)

  inline fun <reified T : Any> get(key: String): Store<T> {
    if (!directory.exists()) directory.mkdir()

    val adapter = moshi.adapter(T::class.java)
    val file = File(directory, key)
    file.createNewFile()
    return Store<T>(file, adapter, scheduler)
  }

  inline fun <reified T : Any> getList(key: String): ListStore<T> {
    if (!directory.exists()) directory.mkdir()

    val listType = Types.newParameterizedType(List::class.java, T::class.java)
    val adapter: JsonAdapter<List<T>> = moshi.adapter(listType)
    val file = File(directory, key)
    file.createNewFile()
    return ListStore(file, adapter, scheduler)
  }

  class Store<T>(val file: File, val adapter: JsonAdapter<T>, val scheduler: Scheduler) {
    private val updates = observable<T> { subscriber ->
      object : FileObserver(file.absolutePath, CLOSE_WRITE) {
        override fun onEvent(event: Int, path: String?) {
          try {
            val source = Okio.buffer(Okio.source(file))
            val value = adapter.nullSafe().fromJson(source)
            subscriber.onNext(value)
          } catch(e: Exception) {
            subscriber.onError(e)
          }
        }
      }.apply { startWatching() }
    }.share()

    fun put(value: T) {
      observePut(value).subscribe({}, { throw it })
    }

    fun observePut(value: T) = single<T> { subscriber ->
      try {
        val sink = Okio.buffer(Okio.sink(file))
        adapter.toJson(sink, value)
        sink.close()
        subscriber.onSuccess(value)
      } catch (e: Exception) {
        subscriber.onError(e)
      }
    }.subscribeOn(scheduler)

    fun get() = single<T?> { subscriber ->
      if (!file.exists()) {
        subscriber.onSuccess(null)
      }

      try {
        val source = Okio.buffer(Okio.source(file))

        if (source.exhausted()) {
          subscriber.onSuccess(null)
        } else {
          val value = adapter.nullSafe().fromJson(source)
          subscriber.onSuccess(value)
        }
      } catch(e: Exception) {
        subscriber.onError(e)
      }
    }.subscribeOn(scheduler)

    fun asObservable(): Observable<T?> = updates.startWith(get().toObservable())
        .subscribeOn(scheduler)

    fun clear() {
      file.delete()
    }
  }

  class ListStore<T>(val file: File, val adapter: JsonAdapter<List<T>>, val scheduler: Scheduler) {
    private val updates = observable<List<T>> { subscriber ->
      object : FileObserver(file.absolutePath, CLOSE_WRITE) {
        override fun onEvent(event: Int, path: String?) {
          try {
            val source = Okio.buffer(Okio.source(file))
            val value = adapter.nullSafe().fromJson(source)
            subscriber.onNext(value)
          } catch(e: Exception) {
            subscriber.onError(e)
          }
        }
      }.apply { startWatching() }
    }.share()

    fun put(value: List<T>) {
      observePut(value).subscribe({}, { throw it })
    }

    fun observePut(value: List<T>) = single<List<T>> { subscriber ->
      try {
        val sink = Okio.buffer(Okio.sink(file))
        adapter.toJson(sink, value)
        sink.close()
        subscriber.onSuccess(value)
      } catch (e: Exception) {
        subscriber.onError(e)
      }
    }.subscribeOn(scheduler)

    fun get() = single<List<T>> { subscriber ->
      if (!file.exists()) {
        subscriber.onSuccess(emptyList())
      }

      try {
        val source = Okio.buffer(Okio.source(file))

        if (source.exhausted()) {
          subscriber.onSuccess(emptyList())
        } else {
          val value = adapter.nullSafe().fromJson(source)
          subscriber.onSuccess(value)
        }
      } catch(e: Exception) {
        subscriber.onError(e)
      }
    }.subscribeOn(scheduler)

    fun asObservable(): Observable<List<T>> = updates.startWith(get().toObservable())
        .subscribeOn(scheduler)

    fun observeAddToList(value: T) = get()
        .map { it.plus(value) }
        .flatMap { observePut(it) }
        .subscribeOn(scheduler)

    fun addToList(value: T) {
      observeAddToList(value).subscribe({}, { throw it })
    }

    fun observeRemoveFromList(value: T) = get()
        .map { it.minus(value) }
        .flatMap { observePut(it) }
        .subscribeOn(scheduler)

    fun removeFromList(value: T) {
      observeRemoveFromList(value).toObservable().subscribe({}, { throw it })
    }

    fun observeRemoveFromList(position: Int) = get()
        .map { it.minus(it[position]) }
        .flatMap { observePut(it) }
        .subscribeOn(scheduler)

    fun removeFromList(position: Int) {
      observeRemoveFromList(position).toObservable().subscribe({}, { throw it })
    }

    fun observeUpdate(value: T) = get()
        .map {
          it.toMutableList().apply {
            val index = indexOf(value)
            removeAt(index)
            add(index, value)
          }
        }
        .flatMap { observePut(it) }
        .subscribeOn(scheduler)

    fun update(value: T) {
      observeUpdate(value).toObservable().subscribe({}, { throw it })
    }

    fun clear() {
      file.delete()
    }
  }
}