Rx Java and If-Then-Else like behavior:

RxIfThenElseTransformer.kt

/*
 * Copyright (c) 2018 Intrepid Pursuits, Inc. All rights reserved.
 */
package io.intrepid.bleidiom.util

import io.reactivex.Observable
import io.reactivex.ObservableSource
import io.reactivex.ObservableTransformer
import io.reactivex.functions.Function

/**
 * Creates a [RxIfThenElseTransformer] that allows emitted values to be processed over
 * various 'pipelines', much like letting them flow through an 'if then else-if then else-if .. else' statement.
 *
 * Example
 *
 *      val observables = Observable.fromIterable(List<Int>(100){ it })
 *      val ifObs = observables.compose(
 *          composeIf<Int,Pair<Int,Int>> {
 *              it % 2 == 0
 *          } then {
 *              map { it to EVEN }
 *          } elseif {
 *              it % 3 == 0
 *          } then {
 *              map { it to DIV_BY_THREE }
 *          } elseif {
 *              it % 5 == 0
 *          } then {
 *              map { it to DIV_BY_FIVE }
 *          } default { map { it to INTEGER } }
 *      )
 *
 *  The code above will produce 100 Pairs whose first values are numbers from 0 through 99
 *  and whose second value are showing the type of the number (EVEN, DIV_BY_THREE,
 *  DIV_BY_FIVE or INTEGER). The result, the number of produced Pairs, will be the same when
 *  you run this as a simple `if then else` or `when` statement inside a loop.
 */
fun <T, R> composeIf(predicate: (T) -> Boolean): RxIfThenElseTransformer.IfBuilder<T, R>.Then {
    return RxIfThenElseTransformer.IfBuilder<T, R>().Then(predicate)
}

private typealias Block<T, R> = Pair<Function<in T, Boolean>, Function<Observable<T>, Observable<R>>>

class RxIfThenElseTransformer<T, R> private constructor(private val blocks: List<Block<T, R>>)
    : ObservableTransformer<T, R> {
    override fun apply(upstream: Observable<T>): ObservableSource<R> = upstream
            .groupBy { indexOf(it) }
            .flatMap { group ->
                val index = group.key!!
                if (index >= 0) blocks[index].second.apply(group)
                else throw Exception("The 'default' clause was not specified for this RxIfThenElseTransformer")
            }

    private fun indexOf(value: T) = blocks.indexOfFirst { it.first.apply(value) }

    class IfBuilder<T, R> internal constructor() {
        var blocks = mutableListOf<Block<T, R>>()

        infix fun elseif(predicate: (T) -> Boolean) = Then(predicate)

        infix fun default(block: Observable<T>.() -> (Observable<R>)): RxIfThenElseTransformer<T, R> {
            blocks.add(Function<T, Boolean> { true } to Function { it.block() })
            return RxIfThenElseTransformer(blocks)
        }

        inner class Then(private val predicate: (T) -> Boolean) {
            infix fun then(block: Observable<T>.() -> (Observable<R>)): IfBuilder<T, R> {
                blocks.add(Function<T, Boolean> { predicate(it) } to Function { it.block() })
                return this@IfBuilder
            }
        }
    }
}

RxIfThenElseTransformerTest.kt

/*
 * Copyright (c) 2018 Intrepid Pursuits, Inc. All rights reserved.
 */
package io.intrepid.bleidiom

import io.intrepid.bleidiom.util.composeIf
import io.reactivex.Observable
import io.reactivex.observers.TestObserver
import org.junit.Test
import kotlin.test.assertEquals

class RxIfThenElseTransformerTest {
    @Test
    fun test_if_then_else() {
        val values = List(100) { it }
        val observables = Observable.fromIterable(values)

        val ifObs = observables.compose(
                composeIf<Int,Pair<Int,Int>> {
                    it % 2 == 0
                } then {
                    map { it to 1 }
                } elseif {
                    it % 3 == 0
                } then {
                    map { it to 2 }
                } elseif {
                    it % 5 == 0
                } then {
                    map { it to 3 }
                } default { map { it to 4 } }
        )

        val testObserver = TestObserver<Pair<Int,Int>>()
        ifObs.subscribe(testObserver)

        // Test if the boolean predicates in the 'composeIf'
        // yield the same result as when they are used in a
        // simple for-loop with when-clauses in the same order.
        testObserver.values().forEach { (value, cat) ->
            when {
                value % 2 == 0 -> assertEquals(1, cat)
                value % 3 == 0 -> assertEquals(2, cat)
                value % 5 == 0 -> assertEquals(3, cat)
                else -> assertEquals(4, cat)
            }
        }
    }
}