A port of RxJS's expand to RxKotlin.

expand.kt

import io.reactivex.Flowable
import io.reactivex.Maybe
import io.reactivex.Observable
import io.reactivex.Single
import io.reactivex.rxkotlin.Flowables
import io.reactivex.rxkotlin.Maybes
import io.reactivex.rxkotlin.Observables
import io.reactivex.rxkotlin.Singles

fun <T> Observables.expand(source: Observable<T>, expander: (T) -> Observable<T>): Observable<T> {
    return Observable.defer {
        source.mergeWith(source.flatMap {
            expand(expander(it), expander)
        })
    }
}

fun <T> Flowables.expand(source: Flowable<T>, expander: (T) -> Flowable<T>): Flowable<T> {
    return Flowable.defer {
        source.mergeWith(source.flatMap {
            expand(expander(it), expander)
        })
    }
}

fun <T> Singles.expand(source: Single<T>, expander: (T) -> Single<T>): Flowable<T> {
    return Flowables.expand(source.toFlowable()) {
        expander(it).toFlowable()
    }
}

fun <T> Maybes.expand(source: Maybe<T>, expander: (T) -> Maybe<T>): Flowable<T> {
    return Flowables.expand(source.toFlowable()) {
        expander(it).toFlowable()
    }
}