Fix for missing pre-binding events

binder/src/main/java/com/badoo/binder/AccumulatorSubject.kt

@@ -0,0 +1,54 @@
+package com.badoo.binder
+
+import io.reactivex.ObservableSource
+import io.reactivex.Observer
+import io.reactivex.functions.Consumer
+import io.reactivex.subjects.PublishSubject
+import java.util.concurrent.atomic.AtomicBoolean
+
+interface Drainable {
+    fun drain()
+}
+
+class AccumulatorSubject<T>(
+    private val initialState: T? = null
+) : ObservableSource<T>, Consumer<T>, Drainable {
+
+    private val items: MutableList<T> = mutableListOf()
+    private val events: PublishSubject<T> = PublishSubject.create()
+
+    private var drained = AtomicBoolean(false)
+
+    init {
+        initialState?.also { items.add(it) }
+    }
+
+    override fun subscribe(observer: Observer<in T>) {
+        events.subscribe(observer)
+        if (!drained.get()) {
+            items.forEach { observer.onNext(it) }
+        } else {
+            initialState?.also { observer.onNext(it) }
+        }
+    }
+
+    override fun accept(value: T?) {
+        value?.also {
+            if (!drained.get()) {
+                items.add(value)
+            }
+            events.onNext(value)
+        }
+    }
+
+    override fun drain() {
+        if (!drained.get()) {
+            drained.set(true)
+            items.clear()
+        }
+    }
+
+    companion object {
+        fun <T> create() = AccumulatorSubject<T>()
+    }
+}

binder/src/main/java/com/badoo/binder/Binder.kt

@@ -17,8 +17,9 @@ import io.reactivex.rxkotlin.plusAssign
 class Binder(
     private val lifecycle: Lifecycle? = null,
 ) : Disposable {
+    private var drained: Boolean = false
+    private val bindings = mutableListOf<Binding>()
     private val disposables = CompositeDisposable()
-    private val connections = mutableListOf<Pair<Connection<*, *>, Middleware<*, *>?>>()
     private val connectionDisposables = CompositeDisposable()
     private var isActive = false
 
@@ -51,21 +52,27 @@ class Binder(
 
         when {
             lifecycle != null -> {
-                connections += (connection to middleware)
-                if (isActive) {
-                    subscribeWithLifecycle(connection, middleware)
+                with(Binding(connection, middleware)) {
+                    bindings.add(this)
+                    if (drained) {
+                        this.drain()
+                    }
+                    accumulate()
+                    if (isActive) {
+                        subscribeWithLifecycle<In>(this)
+                    }
                 }
             }
             else -> subscribe(connection, middleware)
         }
     }
 
-    private fun <Out, In> subscribeWithLifecycle(
-        connection: Connection<Out, In>,
-        middleware: Middleware<Out, In>?
-    ) {
-        connectionDisposables += wrap(connection.from)
-            .subscribeWithMiddleware(connection, middleware)
+    private fun <In> subscribeWithLifecycle(binding: Binding) {
+        connectionDisposables += wrap(binding.source)
+            .subscribeWithMiddleware(
+                binding.connection as Connection<Any?, In>,
+                binding.middleware as? Middleware<Any?, In>,
+            )
     }
 
     private fun <Out, In> subscribe(
@@ -120,12 +127,8 @@ class Binder(
 
     private fun bindConnections() {
         isActive = true
-        connections.forEach { (connection, middleware) ->
-            subscribeWithLifecycle(
-                connection as Connection<Any, Any>,
-                middleware as? Middleware<Any, Any>
-            )
-        }
+        bindings.forEach { it.accumulate() }
+        bindings.forEach { subscribeWithLifecycle<Any>(it) }
     }
 
     private fun unbindConnections() {
@@ -160,6 +163,13 @@ class Binder(
             this
         }
 
+    fun drain() {
+        if (!drained) {
+            bindings.forEach { it.drain() }
+            drained = true
+        }
+    }
+
     class BinderObserveOnScope(
         private val binder: Binder,
         private val observeScheduler: Scheduler

binder/src/main/java/com/badoo/binder/Binding.kt

@@ -0,0 +1,25 @@
+package com.badoo.binder
+
+import com.badoo.binder.middleware.base.Middleware
+import io.reactivex.ObservableSource
+import io.reactivex.subjects.UnicastSubject
+
+internal class Binding(
+    val connection: Connection<*, *>,
+    val middleware: Middleware<*, *>?
+) {
+
+    var source: ObservableSource<*>? = null
+        private set
+
+    fun accumulate() {
+        source = connection.from?.let { source ->
+            UnicastSubject.create<Any>()
+                .also { observer -> source.subscribe(observer) }
+        }
+    }
+
+    fun drain() {
+        (connection.from as? Drainable)?.drain()
+    }
+}

binder/src/test/java/com/badoo/binder/AccumulatorSubjectTest.kt

@@ -0,0 +1,116 @@
+package com.badoo.binder
+
+import com.badoo.binder.lifecycle.Lifecycle
+import com.badoo.binder.lifecycle.ManualLifecycle
+import io.reactivex.functions.Consumer
+import io.reactivex.subjects.PublishSubject
+import org.junit.jupiter.api.Test
+
+class AccumulatorSubjectTest {
+
+    @Test
+    fun `GIVEN the producer is an accumulator AND the consumer subscribes after the lifecycle started AND before the drain THEN consumes all the events produced`() {
+        var score = 0
+        val lifecycle: ManualLifecycle = Lifecycle.manual()
+        val producer = AccumulatorSubject.create<Int>()
+        val consumer = PublishSubject.create<Int>()
+        val testObserver = consumer.test()
+        val binder = Binder(lifecycle)
+
+        producer.accept(score++)
+        producer.accept(score++)
+        lifecycle.begin()
+        binder.bind(producer to Consumer { consumer.onNext(it) })
+        binder.drain()
+
+        testObserver.onComplete()
+        testObserver.assertValues(0, 1)
+    }
+
+    @Test
+    fun `GIVEN the producer is an accumulator AND the consumer subscribes after lifecycle started AND before the drain AND producer produces an event after the drain THEN consumes all the produced events`() {
+        var score = 0
+        val lifecycle: ManualLifecycle = Lifecycle.manual()
+        val producer = AccumulatorSubject.create<Int>()
+        val consumer = PublishSubject.create<Int>()
+        val testObserver = consumer.test()
+        val binder = Binder(lifecycle)
+
+        producer.accept(score++)
+        producer.accept(score++)
+        lifecycle.begin()
+        binder.bind(producer to Consumer { consumer.onNext(it) })
+        binder.drain()
+        producer.accept(score++)
+
+        testObserver.onComplete()
+        testObserver.assertValues(0, 1, 2)
+    }
+
+    @Test
+    fun `GIVEN the producer is an accumulator AND the consumer subscribes after the lifecycle started AND after the drain THEN the consumer consumes only the events produced after the drain`() {
+        var score = 0
+        val lifecycle: ManualLifecycle = Lifecycle.manual()
+        val producer = AccumulatorSubject.create<Int>()
+        val consumer = PublishSubject.create<Int>()
+        val testObserver = consumer.test()
+        val binder = Binder(lifecycle)
+
+        producer.accept(score++)
+        producer.accept(score++)
+        lifecycle.begin()
+        binder.drain()
+        binder.bind(producer to Consumer { consumer.onNext(it) })
+        producer.accept(score++)
+
+        testObserver.onComplete()
+        testObserver.assertValues(2)
+    }
+
+    @Test
+    fun `GIVEN the producer is an accumulator WHEN the consumer subscribes after the lifecycle started AND before the drain THEN should receive events produced before the drain and published when the lifecycle is active`() {
+        var score = 0
+        val lifecycle: ManualLifecycle = Lifecycle.manual()
+        val producer = AccumulatorSubject.create<Int>()
+        val consumer = PublishSubject.create<Int>()
+        val testObserver = consumer.test()
+        val binder = Binder(lifecycle)
+
+        producer.accept(score++)
+        producer.accept(score++)
+        lifecycle.begin()
+        binder.bind(producer to Consumer { consumer.onNext(it) })
+        binder.drain()
+        lifecycle.end()
+        producer.accept(score++)
+        lifecycle.begin()
+        producer.accept(score++)
+
+        testObserver.onComplete()
+        testObserver.assertValues(0, 1, 3)
+    }
+
+    @Test
+    fun `GIVEN the producer is an accumulator WHEN the consumer subscribes after the lifecycle restarted THEN should receive only the events after the restart when the lifecycle is active`() {
+        var score = 0
+        val lifecycle: ManualLifecycle = Lifecycle.manual()
+        val producer = AccumulatorSubject.create<Int>()
+        val consumer = PublishSubject.create<Int>()
+        val testObserver = consumer.test()
+        val binder = Binder(lifecycle)
+
+        producer.accept(score++)
+        producer.accept(score++)
+        lifecycle.begin()
+        binder.drain()
+        lifecycle.end()
+        binder.bind(producer to Consumer { consumer.onNext(it) })
+        producer.accept(score++)
+        lifecycle.begin()
+        producer.accept(score++)
+
+        testObserver.onComplete()
+        testObserver.assertValues(3)
+    }
+
+}