Support Rx2 (#531)

* Add rx2 module

* Support Rx2

Signed-off-by: Matt Ramotar <mramotar@dropbox.com>

* Add unit tests

Signed-off-by: Matt Ramotar <mramotar@dropbox.com>

* Format

Signed-off-by: Matt Ramotar <mramotar@dropbox.com>

---------

Signed-off-by: Matt Ramotar <mramotar@dropbox.com>

buildSrc/src/main/kotlin/Deps.kt

@@ -39,6 +39,7 @@ object Deps {
         const val serializationJson = "org.jetbrains.kotlinx:kotlinx-serialization-json:${Version.kotlinxSerialization}"
         const val coroutinesAndroid = "org.jetbrains.kotlinx:kotlinx-coroutines-android:${Version.kotlinxCoroutines}"
         const val coroutinesCore = "org.jetbrains.kotlinx:kotlinx-coroutines-core:${Version.kotlinxCoroutines}"
+        const val coroutinesRx2 = "org.jetbrains.kotlinx:kotlinx-coroutines-rx2:${Version.kotlinxCoroutines}"
         const val dateTime = "org.jetbrains.kotlinx:kotlinx-datetime:0.4.0"
     }
 
@@ -47,6 +48,10 @@ object Deps {
         const val clientCio = "io.ktor:ktor-client-cio:${Version.ktor}"
     }
 
+    object Rx {
+        const val rx2 = "io.reactivex.rxjava2:rxjava:2.2.21"
+    }
+
     object SqlDelight {
         const val gradlePlugin = "com.squareup.sqldelight:gradle-plugin:${Version.sqlDelight}"
         const val driverAndroid = "com.squareup.sqldelight:android-driver:${Version.sqlDelight}"
@@ -61,6 +66,7 @@ object Deps {
         const val core = "androidx.test:core:${Version.testCore}"
         const val coroutinesTest = "org.jetbrains.kotlinx:kotlinx-coroutines-test:${Version.kotlinxCoroutines}"
         const val junit = "junit:junit:${Version.junit}"
+        const val truth = "com.google.truth:truth:${Version.truth}"
     }
 
     object Touchlab {

buildSrc/src/main/kotlin/Version.kt

@@ -33,4 +33,5 @@ object Version {
     const val sqlDelight = "1.5.4"
     const val sqlDelightGradlePlugin = sqlDelight
     const val store = "5.0.0-alpha05"
+    const val truth = "1.1.3"
 }

rx2/build.gradle.kts

@@ -0,0 +1,64 @@
+@file:Suppress("UnstableApiUsage")
+
+import com.vanniktech.maven.publish.SonatypeHost.S01
+import org.jetbrains.dokka.gradle.DokkaTask
+
+plugins {
+    kotlin("android")
+    id("com.android.library")
+    id("com.vanniktech.maven.publish")
+    id("org.jetbrains.dokka")
+    id("org.jetbrains.kotlinx.kover")
+    `maven-publish`
+    kotlin("native.cocoapods")
+}
+
+dependencies {
+    implementation(Deps.Kotlinx.coroutinesRx2)
+    implementation(Deps.Kotlinx.coroutinesCore)
+    implementation(Deps.Kotlinx.coroutinesAndroid)
+    implementation(Deps.Rx.rx2)
+    implementation(project(":store"))
+
+    testImplementation(kotlin("test"))
+    with(Deps.Test) {
+        testImplementation(junit)
+        testImplementation(core)
+        testImplementation(coroutinesTest)
+        testImplementation(truth)
+    }
+}
+
+android {
+    compileSdk = 33
+
+    defaultConfig {
+        minSdk = 24
+        targetSdk = 33
+    }
+
+    lint {
+        disable += "ComposableModifierFactory"
+        disable += "ModifierFactoryExtensionFunction"
+        disable += "ModifierFactoryReturnType"
+        disable += "ModifierFactoryUnreferencedReceiver"
+    }
+
+    compileOptions {
+        sourceCompatibility = JavaVersion.VERSION_11
+        targetCompatibility = JavaVersion.VERSION_11
+    }
+}
+
+tasks.withType<DokkaTask>().configureEach {
+    dokkaSourceSets.configureEach {
+        reportUndocumented.set(false)
+        skipDeprecated.set(true)
+        jdkVersion.set(8)
+    }
+}
+
+mavenPublishing {
+    publishToMavenCentral(S01)
+    signAllPublications()
+}

rx2/gradle.properties

@@ -0,0 +1,3 @@
+POM_NAME=org.mobilenativefoundation.store
+POM_ARTIFACT_ID=rx2
+POM_PACKAGING=jar

rx2/src/main/AndroidManifest.xml

@@ -0,0 +1,2 @@
+<?xml version="1.0" encoding="utf-8"?>
+<manifest package="org.mobilenativefoundation.store.rx2" />

rx2/src/main/kotlin/org/mobilenativefoundation/store/rx2/RxFetcher.kt

@@ -0,0 +1,70 @@
+package org.mobilenativefoundation.store.rx2
+
+import io.reactivex.Flowable
+import io.reactivex.Single
+import kotlinx.coroutines.reactive.asFlow
+import org.mobilenativefoundation.store.store5.Fetcher
+import org.mobilenativefoundation.store.store5.FetcherResult
+import org.mobilenativefoundation.store.store5.Store
+
+/**
+ * Creates a new [Fetcher] from a [flowableFactory].
+ *
+ * [Store] does not catch exception thrown in [flowableFactory] or in the returned [Flowable]. These
+ * exception will be propagated to the caller.
+ *
+ * Use when creating a [Store] that fetches objects in a multiple responses per request
+ * network protocol (e.g Web Sockets).
+ *
+ * @param flowableFactory a factory for a [Flowable] source of network records.
+ */
+fun <Key : Any, Output : Any> Fetcher.Companion.ofResultFlowable(
+        flowableFactory: (key: Key) -> Flowable<FetcherResult<Output>>
+): Fetcher<Key, Output> = ofResultFlow { key: Key -> flowableFactory(key).asFlow() }
+
+/**
+ * "Creates" a [Fetcher] from a [singleFactory].
+ *
+ * [Store] does not catch exception thrown in [singleFactory] or in the returned [Single]. These
+ * exception will be propagated to the caller.
+ *
+ * Use when creating a [Store] that fetches objects in a single response per request network
+ * protocol (e.g Http).
+ *
+ * @param singleFactory a factory for a [Single] source of network records.
+ */
+fun <Key : Any, Output : Any> Fetcher.Companion.ofResultSingle(
+        singleFactory: (key: Key) -> Single<FetcherResult<Output>>
+): Fetcher<Key, Output> = ofResultFlowable { key: Key -> singleFactory(key).toFlowable() }
+
+/**
+ * "Creates" a [Fetcher] from a [flowableFactory] and translate the results to a [FetcherResult].
+ *
+ * Emitted values will be wrapped in [FetcherResult.Data]. if an exception disrupts the stream then
+ * it will be wrapped in [FetcherResult.Error]. Exceptions thrown in [flowableFactory] itself are
+ * not caught and will be returned to the caller.
+ *
+ * Use when creating a [Store] that fetches objects in a multiple responses per request
+ * network protocol (e.g Web Sockets).
+ *
+ * @param flowFactory a factory for a [Flowable] source of network records.
+ */
+fun <Key : Any, Output : Any> Fetcher.Companion.ofFlowable(
+        flowableFactory: (key: Key) -> Flowable<Output>
+): Fetcher<Key, Output> = ofFlow { key: Key -> flowableFactory(key).asFlow() }
+
+/**
+ * Creates a new [Fetcher] from a [singleFactory] and translate the results to a [FetcherResult].
+ *
+ * The emitted value will be wrapped in [FetcherResult.Data]. if an exception is returned then
+ * it will be wrapped in [FetcherResult.Error]. Exceptions thrown in [singleFactory] itself are
+ * not caught and will be returned to the caller.
+ *
+ * Use when creating a [Store] that fetches objects in a single response per request network
+ * protocol (e.g Http).
+ *
+ * @param singleFactory a factory for a [Single] source of network records.
+ */
+fun <Key : Any, Output : Any> Fetcher.Companion.ofSingle(
+        singleFactory: (key: Key) -> Single<Output>
+): Fetcher<Key, Output> = ofFlowable { key: Key -> singleFactory(key).toFlowable() }

rx2/src/main/kotlin/org/mobilenativefoundation/store/rx2/RxSourceOfTruth.kt

@@ -0,0 +1,64 @@
+package org.mobilenativefoundation.store.rx2
+
+
+import io.reactivex.Completable
+import io.reactivex.Flowable
+import io.reactivex.Maybe
+import kotlinx.coroutines.reactive.asFlow
+import kotlinx.coroutines.rx2.await
+import kotlinx.coroutines.rx2.awaitSingleOrNull
+import org.mobilenativefoundation.store.store5.SourceOfTruth
+
+/**
+ * Creates a [Maybe] source of truth that is accessible via [reader], [writer], [delete] and
+ * [deleteAll].
+ *
+ * @param reader function for reading records from the source of truth
+ * @param writer function for writing updates to the backing source of truth
+ * @param delete function for deleting records in the source of truth for the given key
+ * @param deleteAll function for deleting all records in the source of truth
+ *
+ */
+fun <Key : Any, Local : Any> SourceOfTruth.Companion.ofMaybe(
+        reader: (Key) -> Maybe<Local>,
+        writer: (Key, Local) -> Completable,
+        delete: ((Key) -> Completable)? = null,
+        deleteAll: (() -> Completable)? = null
+): SourceOfTruth<Key, Local> {
+    val deleteFun: (suspend (Key) -> Unit)? =
+            if (delete != null) { key -> delete(key).await() } else null
+    val deleteAllFun: (suspend () -> Unit)? = deleteAll?.let { { deleteAll().await() } }
+    return of(
+            nonFlowReader = { key -> reader.invoke(key).awaitSingleOrNull() },
+            writer = { key, output -> writer.invoke(key, output).await() },
+            delete = deleteFun,
+            deleteAll = deleteAllFun
+    )
+}
+
+/**
+ * Creates a ([Flowable]) source of truth that is accessed via [reader], [writer], [delete] and
+ * [deleteAll].
+ *
+ * @param reader function for reading records from the source of truth
+ * @param writer function for writing updates to the backing source of truth
+ * @param delete function for deleting records in the source of truth for the given key
+ * @param deleteAll function for deleting all records in the source of truth
+ *
+ */
+fun <Key : Any, Local : Any> SourceOfTruth.Companion.ofFlowable(
+        reader: (Key) -> Flowable<Local>,
+        writer: (Key, Local) -> Completable,
+        delete: ((Key) -> Completable)? = null,
+        deleteAll: (() -> Completable)? = null
+): SourceOfTruth<Key, Local> {
+    val deleteFun: (suspend (Key) -> Unit)? =
+            if (delete != null) { key -> delete(key).await() } else null
+    val deleteAllFun: (suspend () -> Unit)? = deleteAll?.let { { deleteAll().await() } }
+    return of(
+            reader = { key -> reader.invoke(key).asFlow() },
+            writer = { key, output -> writer.invoke(key, output).await() },
+            delete = deleteFun,
+            deleteAll = deleteAllFun
+    )
+}

rx2/src/main/kotlin/org/mobilenativefoundation/store/rx2/RxStore.kt

@@ -0,0 +1,49 @@
+package org.mobilenativefoundation.store.rx2
+
+
+import io.reactivex.Completable
+import io.reactivex.Flowable
+import kotlinx.coroutines.rx2.asFlowable
+import kotlinx.coroutines.rx2.rxCompletable
+import kotlinx.coroutines.rx2.rxSingle
+import org.mobilenativefoundation.store.store5.ExperimentalStoreApi
+import org.mobilenativefoundation.store.store5.Store
+import org.mobilenativefoundation.store.store5.StoreBuilder
+import org.mobilenativefoundation.store.store5.StoreReadRequest
+import org.mobilenativefoundation.store.store5.StoreReadResponse
+import org.mobilenativefoundation.store.store5.impl.extensions.fresh
+import org.mobilenativefoundation.store.store5.impl.extensions.get
+
+/**
+ * Return a [Flowable] for the given key
+ * @param request - see [StoreReadRequest] for configurations
+ */
+fun <Key : Any, Output : Any> Store<Key, Output>.observe(request: StoreReadRequest<Key>): Flowable<StoreReadResponse<Output>> =
+        stream(request).asFlowable()
+
+/**
+ * Purge a particular entry from memory and disk cache.
+ * Persistent storage will only be cleared if a delete function was passed to
+ * [StoreBuilder.persister] or [StoreBuilder.nonFlowingPersister] when creating the [Store].
+ */
+fun <Key : Any, Output : Any> Store<Key, Output>.observeClear(key: Key): Completable =
+        rxCompletable { clear(key) }
+
+/**
+ * Purge all entries from memory and disk cache.
+ * Persistent storage will only be cleared if a deleteAll function was passed to
+ * [StoreBuilder.persister] or [StoreBuilder.nonFlowingPersister] when creating the [Store].
+ */
+@ExperimentalStoreApi
+fun <Key : Any, Output : Any> Store<Key, Output>.observeClearAll(): Completable =
+        rxCompletable { clear() }
+
+/**
+ * Helper factory that will return data as a [Single] for [key] if it is cached otherwise will return fresh/network data (updating your caches)
+ */
+fun <Key : Any, Output : Any> Store<Key, Output>.getSingle(key: Key) = rxSingle { this@getSingle.get(key) }
+
+/**
+ * Helper factory that will return fresh data as a [Single] for [key] while updating your caches
+ */
+fun <Key : Any, Output : Any> Store<Key, Output>.freshSingle(key: Key) = rxSingle { this@freshSingle.fresh(key) }

rx2/src/main/kotlin/org/mobilenativefoundation/store/rx2/RxStoreBuilder.kt

@@ -0,0 +1,26 @@
+package org.mobilenativefoundation.store.rx2
+
+import io.reactivex.Scheduler
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.rx2.asCoroutineDispatcher
+import org.mobilenativefoundation.store.store5.Store
+import org.mobilenativefoundation.store.store5.StoreBuilder
+
+/**
+ *  A store multicasts same [Output] value to many consumers (Similar to RxJava.share()), by default
+ *  [Store] will open a global scope for management of shared responses, if instead you'd like to control
+ *  the scheduler that sharing/multicasting happens in you can pass a @param [scheduler]
+ *
+ *  Note this does not control what scheduler a response is emitted on but rather what thread/scheduler
+ *  to use when managing in flight responses. This is usually used for things like testing where you
+ *  may want to confine to a scheduler backed by a single thread executor
+ *
+ *   @param scheduler - scheduler to use for sharing
+ *  if a scheduler is not set Store will use [GlobalScope]
+ */
+fun <Key : Any, Network : Any, Output : Any, Local : Any> StoreBuilder<Key, Network, Output, Local>.withScheduler(
+        scheduler: Scheduler
+): StoreBuilder<Key, Network, Output, Local> {
+    return scope(CoroutineScope(scheduler.asCoroutineDispatcher()))
+}