RxDelay.kt

object Rx {
    fun <T> delayFlowable(delay: Long, timeUnit: TimeUnit, scheduler: Scheduler = Schedulers.computation(),
            block: () -> Flowable<T>): Flowable<T> {
        return Completable.timer(delay, timeUnit, scheduler).andThen(Flowable.defer { block() })
    }

    fun <T> delayObservable(delay: Long, timeUnit: TimeUnit, scheduler: Scheduler = Schedulers.computation(),
            block: () -> Observable<T>): Observable<T> {
        return Completable.timer(delay, timeUnit, scheduler).andThen(Observable.defer { block() })
    }

    fun <T> delaySingle(delay: Long, timeUnit: TimeUnit,
            scheduler: Scheduler = Schedulers.computation(), block: () -> Single<T>): Single<T> {
        return Completable.timer(delay, timeUnit, scheduler).andThen(Single.defer { block() })
    }

    fun <T> delayMaybe(delay: Long, timeUnit: TimeUnit, scheduler: Scheduler = Schedulers.computation(),
            block: () -> Maybe<T>): Maybe<T> {
        return Completable.timer(delay, timeUnit, scheduler).andThen(Maybe.defer { block() })
    }

    fun delayCompletable(delay: Long, timeUnit: TimeUnit, scheduler: Scheduler = Schedulers.computation(),
            block: () -> Completable): Completable {
        return Completable.timer(delay, timeUnit, scheduler).andThen(Completable.defer { block() })
    }
}