Kotlin Rx Extensions

_AppModule.kt

@Module
class AppModule() {

    //...

    @Provides
    @Singleton
    fun getSchedulers(looperThread: LooperThread): Schedulers {
        return RxSchedulers(looperThread)
    }
    
    @Provides
    @Singleton
    fun getLooperThread(): LooperThread {
        val looperThread = LooperThread()
        looperThread.start()
        return looperThread
    }
    
    //...
}

KotlinRxExt.kt

import rx.Observable
import rx.Observer
import rx.Subscriber
import rx.Subscription
import rx.functions.Action0
import rx.functions.Action1

fun <T> Observable<T>.uiSubscribe(schedulers: Schedulers, subscriber: Subscriber<in T>): Subscription {
    return subscribeOn(schedulers.io)
            .observeOn(schedulers.mainThread)
            .subscribe(subscriber)
}

fun <T> Observable<T>.uiSubscribe(schedulers: Schedulers, observer: Observer<in T>): Subscription {
    return subscribeOn(schedulers.io)
            .observeOn(schedulers.mainThread)
            .subscribe(observer)
}

fun <T> Observable<T>.uiSubscribe(schedulers: Schedulers,
                                  onNext: Action1<in T>,
                                  onError: Action1<Throwable>? = null,
                                  onComplete: Action0? = null): Subscription {
    return subscribeOn(schedulers.io)
            .observeOn(schedulers.mainThread)
            .subscribe(onNext, onError, onComplete)
}

fun <T> Observable<T>.uiSubscribe(schedulers: Schedulers, onNext: (T) -> Unit): Subscription {
    return uiSubscribe(schedulers, onNext, null, null)
}

fun <T> Observable<T>.uiSubscribe(schedulers: Schedulers,
                                  onNext: (T) -> Unit,
                                  onError: ((Throwable) -> Unit)? = null,
                                  onComplete: (() -> Unit)? = null): Subscription {

    return subscribeOn(schedulers.io)
            .observeOn(schedulers.mainThread)
            .subscribe({ next ->
                           onNext(next)
                       }, { throwable ->
                           onError?.apply {
                               this(throwable)
                           }
                       }, {
                           onComplete?.apply {
                               this()
                           }
                       })
}

LooperThread.kt

import android.os.Looper

class LooperThread : Thread() {

    private lateinit var looper: Looper
    fun getLooper(): Looper {
        return looper
    }

    override fun run() {
        Looper.prepare()
        looper = Looper.myLooper()
        Looper.loop()
    }

}

MockRxSchedulers.kt

import rx.Scheduler

class MockSchedulers : Schedulers {
    override val io: Scheduler
        get() = rx.schedulers.Schedulers.immediate()
    override val loopedIo: Scheduler
        get() = rx.schedulers.Schedulers.immediate()
    override val mainThread: Scheduler
        get() = rx.schedulers.Schedulers.immediate()
    override val computation: Scheduler
        get() = rx.schedulers.Schedulers.immediate()
}

RxSchedulers.kt

import android.os.Handler
import right.apps.photo.map.data.common.LooperThread
import rx.Scheduler
import rx.android.schedulers.AndroidSchedulers
import rx.android.schedulers.HandlerScheduler
import javax.inject.Inject
import javax.inject.Singleton

@Singleton
class RxSchedulers @Inject constructor(val looperThread: LooperThread) : Schedulers {

    override val io: Scheduler
        get() = rx.schedulers.Schedulers.io()
    override val loopedIo: Scheduler
        get() = HandlerScheduler.from(Handler(looperThread.getLooper()))
    override val mainThread: Scheduler
        get() = AndroidSchedulers.mainThread()
    override val computation: Scheduler
        get() = rx.schedulers.Schedulers.computation()
}

Schedulers.kt

import rx.Scheduler

interface Schedulers {
    val io: Scheduler
    val loopedIo: Scheduler
    val mainThread: Scheduler
    val computation: Scheduler
}