(#4) Fix flowable not desubscribing correctly from the store when disposed. Add optional backpressure strategy to Store.flowable() (#5)

* (#4) Fix flowable not desubscribing correctly from the store when disposed. Add test cases to verify this behaviour

* (#4) Change flowable() default backpressure strategy to Buffer

Author: Adrián García

README.md

@@ -178,7 +178,7 @@ Add the following dependencies to your app's `build.gradle`:
 
 ```groovy
 dependencies {
-    def mini_version = "1.0.8"
+    def mini_version = "1.0.9"
     // Minimum working dependencies
     implementation "com.github.bq.mini-kotlin:mini-android:$mini_version"
     kapt "com.github.bq.mini-kotlin:mini-processor:$mini_version"

mini-common/src/main/java/mini/Store.kt

@@ -21,8 +21,8 @@ abstract class Store<S> : Closeable {
         }
     }
 
+    internal val listeners = Vector<(S) -> Unit>()
     private var _state: Any? = NO_STATE
-    private val listeners = Vector<(S) -> Unit>()
 
     /** Set new state, equivalent to [asNewState]*/
     protected fun setState(state: S) {

mini-common/src/test/kotlin/mini/SampleStore.kt

@@ -6,4 +6,6 @@ class SampleStore : Store<String>() {
     fun updateState(s: String) {
         newState = s
     }
+
+    val storeSubscriptions = listeners
 }

mini-rx2/build.gradle

@@ -9,6 +9,7 @@ dependencies {
 
     implementation "io.reactivex.rxjava2:rxjava:$rx_version"
 
+    testApi files(project(":mini-common").sourceSets.test.runtimeClasspath) // Add mini-common test classes as dependency
     testCompileOnly 'junit:junit:4.12'
     testImplementation "org.amshove.kluent:kluent:1.44"
 }

mini-rx2/src/main/java/mini/rx/RxUtils.kt

@@ -1,10 +1,11 @@
 package mini.rx
 
+import io.reactivex.BackpressureStrategy
+import io.reactivex.BackpressureStrategy.BUFFER
 import io.reactivex.Flowable
 import io.reactivex.Observable
 import io.reactivex.disposables.CompositeDisposable
 import io.reactivex.disposables.Disposable
-import io.reactivex.processors.PublishProcessor
 import io.reactivex.subjects.PublishSubject
 import mini.Store
 
@@ -57,20 +58,17 @@ class DefaultSubscriptionTracker : SubscriptionTracker {
     }
 }
 
-fun <S> Store<S>.flowable(hotStart: Boolean = true): Flowable<S> {
-    val processor = PublishProcessor.create<S>()
-    val subscription = subscribe(hotStart = false) {
-        processor.offer(it)
-    }
-    return processor.doOnTerminate { subscription.close() }
-        .let { if (hotStart) it.startWith(state) else it }
-}
-
 fun <S> Store<S>.observable(hotStart: Boolean = true): Observable<S> {
     val subject = PublishSubject.create<S>()
     val subscription = subscribe(hotStart = false) {
         subject.onNext(it)
     }
-    return subject.doOnTerminate { subscription.close() }
+    return subject
+        .doOnDispose { subscription.close() }
+        .doOnTerminate { subscription.close() }
         .let { if (hotStart) it.startWith(state) else it }
-}
+}
+
+fun <S> Store<S>.flowable(hotStart: Boolean = true,
+                          backpressureStrategy: BackpressureStrategy = BUFFER): Flowable<S> =
+    observable(hotStart).toFlowable(backpressureStrategy)

mini-rx2/src/test/kotlin/mini/rx/RxUtilsKtTest.kt

@@ -1,6 +1,7 @@
 package mini.rx
 
 import mini.SampleStore
+import org.amshove.kluent.`should be empty`
 import org.amshove.kluent.`should be equal to`
 import org.junit.Test
 
@@ -27,6 +28,27 @@ class RxUtilsKtTest {
         sentState `should be equal to` "abc"
     }
 
+    @Test
+    fun `flowable completes`() {
+        val store = SampleStore()
+        var sentState = ""
+        val disposable = store.flowable(hotStart = false).subscribe {
+            sentState = it
+        }
+        disposable.dispose() //Clear it
+        store.updateState("abc")
+        sentState `should be equal to` "" //No change should be made
+    }
+
+    @Test
+    fun `flowable disposes correctly`() {
+        val store = SampleStore()
+        val disposable = store.flowable(hotStart = false).subscribe()
+        disposable.dispose() //Clear it
+
+        store.storeSubscriptions.`should be empty`()
+    }
+
     @Test
     fun `observable sends initial state`() {
         val store = SampleStore()
@@ -48,4 +70,25 @@ class RxUtilsKtTest {
         store.updateState("abc") //Set before subscribe
         sentState `should be equal to` "abc"
     }
+
+    @Test
+    fun `observable completes`() {
+        val store = SampleStore()
+        var sentState = ""
+        val disposable = store.observable(hotStart = false).subscribe {
+            sentState = it
+        }
+        disposable.dispose() //Clear it
+        store.updateState("abc")
+        sentState `should be equal to` "" //No change should be made
+    }
+
+    @Test
+    fun `observable disposes correctly`() {
+        val store = SampleStore()
+        val disposable = store.observable(hotStart = false).subscribe()
+        disposable.dispose() //Clear it
+
+        store.storeSubscriptions.`should be empty`()
+    }
 }