Add source parameter to the connection

mvicore-android/src/main/java/com/badoo/mvicore/android/AndroidBindings.kt

@@ -2,8 +2,8 @@ package com.badoo.mvicore.android
 
 import android.arch.lifecycle.LifecycleOwner
 import com.badoo.mvicore.binder.Binder
-import com.badoo.mvicore.binder.lifecycle.Lifecycle as BinderLifecycle
 import android.arch.lifecycle.Lifecycle as AndroidLifecycle
+import com.badoo.mvicore.binder.lifecycle.Lifecycle as BinderLifecycle
 
 abstract class AndroidBindings<T : Any>(
     lifecycleOwner: LifecycleOwner

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

@@ -3,8 +3,9 @@ package com.badoo.mvicore.binder
 import com.badoo.mvicore.binder.lifecycle.Lifecycle
 import com.badoo.mvicore.binder.lifecycle.Lifecycle.Event.BEGIN
 import com.badoo.mvicore.binder.lifecycle.Lifecycle.Event.END
-import com.badoo.mvicore.consumer.middleware.ConsumerMiddleware
-import com.badoo.mvicore.consumer.wrap
+import com.badoo.mvicore.consumer.middleware.base.Middleware
+import com.badoo.mvicore.consumer.wrapWithMiddleware
+import com.badoo.mvicore.extension.mapNotNull
 import io.reactivex.Observable
 import io.reactivex.ObservableSource
 import io.reactivex.disposables.CompositeDisposable
@@ -16,38 +17,36 @@ class Binder(
     private val lifecycle: Lifecycle? = null
 ) : Disposable {
     private val disposables = CompositeDisposable()
-    private val connections = mutableListOf<Pair<Connection<out Any>, ConsumerMiddleware<out Any>?>>()
+    private val connections = mutableListOf<Pair<Connection<*, *>, Middleware<*, *>?>>()
     private val connectionDisposables = CompositeDisposable()
     private var isActive = false
 
     init {
-        lifecycle?.apply {
-            disposables += Observable.wrap(this)
-                .distinctUntilChanged()
-                .subscribe {
-                    when (it) {
-                       BEGIN -> bindConnections()
-                       END -> unbindConnections()
-                    }
-                }
+        lifecycle?.let {
+            disposables += it.setupConnections()
         }
 
         disposables += connectionDisposables
     }
 
-    fun <T : Any> bind(connection: Pair<ObservableSource<out T>, Consumer<T>>) {
-        bind(Connection(
-            from = connection.first,
-            to = connection.second
-        ))
+    // region bind
+
+    fun <T: Any> bind(connection: Pair<ObservableSource<T>, Consumer<T>>) {
+        bind(
+            Connection(
+                from = connection.first,
+                to = connection.second,
+                transformer = null
+            )
+        )
     }
 
-    fun <T : Any> bind(connection: Connection<T>) {
+    fun <Out: Any, In: Any> bind(connection: Connection<Out, In>) {
         val consumer = connection.to
-        val middleware = consumer.wrap(
+        val middleware = consumer.wrapWithMiddleware(
             standalone = false,
             name = connection.name
-        ) as? ConsumerMiddleware<T>
+        ) as? Middleware<Out, In>
 
         when {
             lifecycle != null -> {
@@ -60,40 +59,69 @@ class Binder(
         }
     }
 
-    private fun <T : Any> subscribeWithLifecycle(
-        connection: Connection<T>,
-        middleware: ConsumerMiddleware<T>?
+    private fun <Out: Any, In: Any> subscribeWithLifecycle(
+        connection: Connection<Out, In>,
+        middleware: Middleware<Out, In>?
     ) {
         connectionDisposables += Observable
             .wrap(connection.from)
             .subscribeWithMiddleware(connection, middleware)
     }
 
-    private fun <T: Any> subscribe(
-        connection: Connection<T>,
-        middleware: ConsumerMiddleware<T>?
+    private fun <Out: Any, In: Any> subscribe(
+        connection: Connection<Out, In>,
+        middleware: Middleware<Out, In>?
     ) {
         disposables += Observable.wrap(connection.from)
             .subscribeWithMiddleware(connection, middleware)
     }
 
-    private fun <T: Any> Observable<out T>.subscribeWithMiddleware(
-        connection: Connection<T>,
-        middleware: ConsumerMiddleware<T>?
-    ): Disposable = run {
-        middleware?.onBind(connection)
-        middleware?.let {
-            this.doOnNext { middleware.onElement(connection, it) }
-                .doFinally { middleware.onComplete(connection) }
-        } ?: this
-    }.subscribe(middleware ?: connection.to)
+    private fun <Out: Any, In: Any> Observable<Out>.subscribeWithMiddleware(
+        connection: Connection<Out, In>,
+        middleware: Middleware<Out, In>?
+    ): Disposable =
+        applyTransformer(connection)
+            .run {
+                if (middleware != null) {
+                    middleware.onBind(connection)
+
+                    this
+                        .doOnNext { middleware.onElement(connection, it) }
+                        .doFinally { middleware.onComplete(connection) }
+                        .subscribe(middleware)
+
+                } else {
+                    subscribe(connection.to)
+                }
+            }
+
+    private fun <Out: Any, In: Any> Observable<Out>.applyTransformer(
+        connection: Connection<Out, In>
+    ): Observable<In> =
+        connection.transformer?.let {
+            mapNotNull(it)
+        } ?: this as Observable<In>
+
+    // endregion
+
+    // region lifecycle
+
+    private fun Lifecycle.setupConnections() =
+        Observable.wrap(this)
+            .distinctUntilChanged()
+            .subscribe {
+                when (it) {
+                    BEGIN -> bindConnections()
+                    END -> unbindConnections()
+                }
+            }
 
     private fun bindConnections() {
         isActive = true
         connections.forEach { (connection, middleware) ->
             subscribeWithLifecycle(
-                connection as Connection<Any>,
-                middleware as? ConsumerMiddleware<Any>
+                connection as Connection<Any, Any>,
+                middleware as? Middleware<Any, Any>
             )
         }
     }
@@ -103,6 +131,8 @@ class Binder(
         connectionDisposables.clear()
     }
 
+    // endregion
+
     override fun isDisposed(): Boolean =
         disposables.isDisposed
 
@@ -114,6 +144,3 @@ class Binder(
         disposables.clear()
     }
 }
-
-
-

mvicore/src/main/java/com/badoo/mvicore/binder/Connection.kt

@@ -1,13 +1,12 @@
 package com.badoo.mvicore.binder
 
-import com.badoo.mvicore.extension.mapNotNull
-import io.reactivex.Observable
 import io.reactivex.ObservableSource
 import io.reactivex.functions.Consumer
 
-data class Connection<T>(
-    val from: ObservableSource<out T>? = null,
-    val to: Consumer<T>,
+data class Connection<Out, In>(
+    val from: ObservableSource<Out>? = null,
+    val to: Consumer<In>,
+    val transformer: ((Out) -> In?)? = null,
     val name: String? = null
 ) {
     companion object {
@@ -18,25 +17,24 @@ data class Connection<T>(
         name == null
 
     override fun toString(): String =
-        "<${name ?: ANONYMOUS}> (${from ?: "?"} --> $to)"
+        "<${name ?: ANONYMOUS}> (${from ?: "?"} --> $to${transformer?.let { " using $it" } ?: ""})"
 }
 
-infix fun <Out, In> Pair<ObservableSource<out Out>, Consumer<In>>.using(transformer: (Out) -> In?): Connection<In> =
+infix fun <Out, In> Pair<ObservableSource<Out>, Consumer<In>>.using(transformer: (Out) -> In?): Connection<Out, In> =
     Connection(
-        from = Observable
-            .wrap(first)
-            .mapNotNull(transformer),
-        to = second
+        from = first,
+        to = second,
+        transformer = transformer
     )
 
-infix fun <T> Pair<ObservableSource<out T>, Consumer<T>>.named(name: String): Connection<T> =
+infix fun <T> Pair<ObservableSource<T>, Consumer<T>>.named(name: String): Connection<T, T> =
     Connection(
         from = first,
         to = second,
         name = name
     )
 
-infix fun <T> Connection<T>.named(name: String) =
+infix fun <Out, In> Connection<Out, In>.named(name: String) =
     copy(
         name = name
     )

mvicore/src/main/java/com/badoo/mvicore/binder/lifecycle/Lifecycle.kt

@@ -1,7 +1,6 @@
 package com.badoo.mvicore.binder.lifecycle
 
 import com.badoo.mvicore.binder.lifecycle.internal.FromObservableSource
-import io.reactivex.Observable
 import io.reactivex.ObservableSource
 
 interface Lifecycle : ObservableSource<Lifecycle.Event> {

mvicore/src/main/java/com/badoo/mvicore/consumer/Consumer.kt

@@ -1,6 +1,7 @@
 package com.badoo.mvicore.consumer
 
-import com.badoo.mvicore.consumer.middleware.ConsumerMiddleware
+import com.badoo.mvicore.consumer.middleware.base.Middleware
+import com.badoo.mvicore.consumer.middleware.base.StandaloneMiddleware
 import com.badoo.mvicore.consumer.middlewareconfig.Middlewares
 import com.badoo.mvicore.consumer.middlewareconfig.NonWrappable
 import io.reactivex.functions.Consumer
@@ -18,12 +19,12 @@ import io.reactivex.functions.Consumer
  * @param postfix       Passed on to [ConsumerMiddleware], in most cases you shouldn't need to override this.
  * @param wrapperOf     Passed on to [ConsumerMiddleware], in most cases you shouldn't need to override this.
  */
-fun <T : Any> Consumer<T>.wrap(
+fun <In : Any> Consumer<In>.wrapWithMiddleware(
     standalone: Boolean = true,
     name: String? = null,
     postfix: String? = null,
     wrapperOf: Any? = null
-): Consumer<T> {
+): Consumer<In> {
     val target = wrapperOf ?: this
     if (target is NonWrappable) return this
 
@@ -33,13 +34,32 @@ fun <T : Any> Consumer<T>.wrap(
         current = it.applyOn(current, target, name, standalone)
     }
 
-    if (current is ConsumerMiddleware<T> && standalone) {
-        (current as ConsumerMiddleware<T>).initAsStandalone(
-            name = name,
-            wrapperOf = target,
+    if (current is Middleware<*, *> && standalone) {
+        return StandaloneMiddleware(
+            wrappedMiddleware = current as Middleware<In, In>,
+            name = name ?: wrapperOf?.javaClass?.canonicalName,
             postfix = postfix
         )
     }
 
     return current
 }
+
+@Deprecated(
+    "Use wrapWithMiddleware directly",
+    ReplaceWith(
+        "wrapWithMiddleware(standalone, name, postfix, wrapperOf)",
+        "com.badoo.mvicore.consumer.wrapWithMiddleware"
+    )
+)
+fun <In: Any> Consumer<In>.wrap(
+    standalone: Boolean = true,
+    name: String? = null,
+    postfix: String? = null,
+    wrapperOf: Any? = null
+) = wrapWithMiddleware(
+    standalone,
+    name,
+    postfix,
+    wrapperOf
+)

mvicore/src/main/java/com/badoo/mvicore/consumer/middleware/ConsumerMiddleware.kt

@@ -1,74 +1,9 @@
 package com.badoo.mvicore.consumer.middleware
 
-import com.badoo.mvicore.binder.Connection
-import io.reactivex.disposables.Disposable
-import io.reactivex.functions.Consumer
+import com.badoo.mvicore.consumer.middleware.base.Middleware
 
-abstract class ConsumerMiddleware<T : Any>(
-    protected val wrapped: Consumer<T>
-) : Consumer<T>, Disposable {
-
-    protected val innerMost = innerMost()
-    private var standaloneConnection: Connection<T>? = null
-    private var standaloneConnectionDisposed: Boolean = false
-
-    private fun innerMost(): Consumer<T> {
-        var current = wrapped
-        while (current is ConsumerMiddleware<T>) {
-            current = current.wrapped
-        }
-
-        return current
-    }
-
-    fun initAsStandalone(name: String? = null, wrapperOf: Any? = null, postfix: String?) {
-        standaloneConnection = Connection(
-            to = innerMost,
-            name = name ?: "${(wrapperOf ?: innerMost).javaClass.canonicalName}.${postfix ?: "input"}"
-        ).also {
-            onBind(it)
-        }
-    }
-
-    private fun check(connection: Connection<T>) {
-        if (standaloneConnection != null && connection != standaloneConnection) {
-            throw IllegalStateException("Middleware was initialised in standalone mode, can't accept other connections")
-        }
-    }
-
-    open fun onBind(connection: Connection<T>) {
-        check(connection)
-        if (wrapped is ConsumerMiddleware) {
-            wrapped.onBind(connection)
-        }
-    }
-
-    override fun accept(t: T) {
-        standaloneConnection?.let { onElement(it, t) }
-        innerMost.accept(t)
-    }
-
-    open fun onElement(connection: Connection<T>, element: T) {
-        check(connection)
-        if (wrapped is ConsumerMiddleware) {
-            wrapped.onElement(connection, element)
-        }
-    }
-
-    open fun onComplete(connection: Connection<T>) {
-        check(connection)
-        if (wrapped is ConsumerMiddleware) {
-            wrapped.onComplete(connection)
-        }
-    }
-
-    override fun dispose() {
-        standaloneConnection?.let {
-            onComplete(it)
-            standaloneConnectionDisposed = true
-        }
-    }
-
-    override fun isDisposed(): Boolean =
-        standaloneConnection == null || standaloneConnectionDisposed
-}
+@Deprecated(
+    "Left for compatibility reasons",
+    ReplaceWith("Middleware<Any, T>", "com.badoo.mvicore.consumer.middleware.base")
+)
+typealias ConsumerMiddleware<T> = Middleware<Any, T>