Add initializer block to binder

mvicore-base/src/commonMain/kotlin/com/badoo/mvicore/common/binder/Binder.kt

@@ -8,13 +8,14 @@ import com.badoo.mvicore.common.lifecycle.Lifecycle
 import com.badoo.mvicore.common.lifecycle.Lifecycle.Event.BEGIN
 import com.badoo.mvicore.common.lifecycle.Lifecycle.Event.END
 import com.badoo.mvicore.common.source
+import com.badoo.mvicore.common.sources.DelayUntilSource
 
 abstract class Binder : Cancellable {
     internal abstract fun <Out, In> connect(connection: Connection<Out, In>)
 }
 
-fun binder(): Binder = SimpleBinder()
-fun binder(lifecycle: Lifecycle): Binder = LifecycleBinder(lifecycle)
+fun binder(init: Binder.() -> Unit = { }): Binder = SimpleBinder(init)
+fun binder(lifecycle: Lifecycle, init: Binder.() -> Unit = { }): Binder = LifecycleBinder(lifecycle, init)
 
 fun <In> Binder.bind(connection: Pair<Source<In>, Sink<In>>) {
     val (from, to) = connection
@@ -25,18 +26,44 @@ fun <Out, In> Binder.bind(connection: Connection<Out, In>) {
     connect(connection)
 }
 
-internal class SimpleBinder : Binder() {
+internal class SimpleBinder(init: Binder.() -> Unit) : Binder() {
     private val cancellables = mutableListOf<Cancellable>()
+
+    /**
+     * Stores internal end of every `emitter` connected through binder
+     * Allows for propagation of events in case emission disposes the binder
+     * E.g.:
+     * from -> internalSource -> to // Events from `from` are propagated to `to`
+     * #dispose()
+     * from xx internalSource -> to // Remaining events from `internalSource` are propagated to `to`
+     */
     private val fromToInternalSource = mutableMapOf<Source<*>, SimpleSource<*>>()
 
+    /**
+     * Delay events emitted on subscribe until `init` lambda is executed
+     */
+    private val initialized = source(initialValue = false)
+
+    init {
+        init()
+        initialized(true)
+    }
+
     override fun <Out, In> connect(connection: Connection<Out, In>) {
         val internalSource = getInternalSourceFor(connection.from!!)
-        val outSource = if (connection.connector != null) {
+        val transformedSource = if (connection.connector != null) {
             connection.connector.invoke(internalSource)
         } else {
             internalSource as Source<In>
         }
-        outSource.connect(connection.to)
+
+        val delayInitialize = if (initialized.value!!) {
+            transformedSource
+        } else {
+            DelayUntilSource(initialized, transformedSource)
+        }
+
+        delayInitialize.connect(connection.to)
     }
 
     private fun <T> getInternalSourceFor(from: Source<T>): SimpleSource<T> =
@@ -51,10 +78,10 @@ internal class SimpleBinder : Binder() {
     }
 }
 
-internal class LifecycleBinder(lifecycle: Lifecycle) : Binder() {
+internal class LifecycleBinder(lifecycle: Lifecycle, init: Binder.() -> Unit) : Binder() {
     private var lifecycleActive = false
     private val cancellables = mutableListOf<Cancellable>()
-    private val innerBinder = SimpleBinder()
+    private val innerBinder = SimpleBinder(init)
     private val connections = mutableListOf<Connection<*, *>>()
 
     init {

mvicore-base/src/commonMain/kotlin/com/badoo/mvicore/common/binder/NotNullConnector.kt

@@ -1,8 +1,7 @@
 package com.badoo.mvicore.common.binder
 
-import com.badoo.mvicore.common.Cancellable
-import com.badoo.mvicore.common.Sink
 import com.badoo.mvicore.common.Source
+import com.badoo.mvicore.common.sources.MapNotNullSource
 
 internal class NotNullConnector<Out, In>(private val mapper: (Out) -> In?): Connector<Out, In> {
     override fun invoke(element: Source<out Out>): Source<In> {
@@ -12,14 +11,3 @@ internal class NotNullConnector<Out, In>(private val mapper: (Out) -> In?): Conn
     override fun toString(): String =
         mapper.toString()
 }
-
-private class MapNotNullSource<Out, In>(private val delegate: Source<Out>, private val mapper: (Out) -> In?): Source<In> {
-    override fun connect(sink: Sink<In>): Cancellable =
-        delegate.connect {
-            mapper(it)?.let { sink(it) }
-        }
-
-    override fun cancel() {
-        delegate.cancel()
-    }
-}

mvicore-base/src/commonMain/kotlin/com/badoo/mvicore/common/source.kt

@@ -1,5 +1,9 @@
 package com.badoo.mvicore.common
 
+/**
+ * NOTE: in conversions from other frameworks you need to override equals and hashCode
+ * to support binder "emit after dispose" functionality
+ */
 interface Source<T> : Cancellable {
     fun connect(sink: Sink<T>): Cancellable
 }

mvicore-base/src/commonMain/kotlin/com/badoo/mvicore/common/sources/DelayUntilSource.kt

@@ -0,0 +1,44 @@
+package com.badoo.mvicore.common.sources
+
+import com.badoo.mvicore.common.Cancellable
+import com.badoo.mvicore.common.Sink
+import com.badoo.mvicore.common.Source
+import com.badoo.mvicore.common.cancellableOf
+
+internal class DelayUntilSource<T>(
+    private val signal: Source<Boolean>,
+    private val delegate: Source<T>
+): Source<T> {
+    override fun connect(sink: Sink<T>): Cancellable {
+        val delayedSink = DelayedSink(sink)
+        val cancelDelay = signal.connect { if (it) { delayedSink.send() } }
+        val cancelSubscription = delegate.connect(delayedSink)
+        return cancellableOf {
+            cancelDelay.cancel()
+            cancelSubscription.cancel()
+        }
+    }
+
+    override fun cancel() {
+        delegate.cancel()
+    }
+
+    private class DelayedSink<T>(private val delegate: Sink<T>) : Sink<T> {
+        private var passThrough = false
+        private val events = mutableListOf<T>()
+
+        override fun invoke(value: T) {
+            if (passThrough) {
+                delegate(value)
+            } else {
+                events += value
+            }
+        }
+
+        fun send() {
+            passThrough = true
+            events.forEach { delegate(it) }
+            events.clear()
+        }
+    }
+}

mvicore-base/src/commonMain/kotlin/com/badoo/mvicore/common/sources/MapNotNullSource.kt

@@ -0,0 +1,16 @@
+package com.badoo.mvicore.common.sources
+
+import com.badoo.mvicore.common.Cancellable
+import com.badoo.mvicore.common.Sink
+import com.badoo.mvicore.common.Source
+
+internal class MapNotNullSource<Out, In>(private val delegate: Source<Out>, private val mapper: (Out) -> In?): Source<In> {
+    override fun connect(sink: Sink<In>): Cancellable =
+        delegate.connect {
+            mapper(it)?.let { sink(it) }
+        }
+
+    override fun cancel() {
+        delegate.cancel()
+    }
+}

mvicore-base/src/commonTest/kotlin/com/badoo/mvicore/common/binder.kt

@@ -197,4 +197,16 @@ class BinderTest {
 
         sink.assertValues(0)
     }
+
+    @Test
+    fun binder_messages_sent_on_initialize_are_not_lost() {
+        val passThroughSource = source<Int>()
+        val binder = binder {
+            bind(source to passThroughSource)
+            source.invoke(0)
+            bind(passThroughSource to sink)
+        }
+
+        sink.assertValues(0)
+    }
 }