RxExt

RxExt.kt

fun <T : Any> Single<T>.withProgress(progress: MutableState<Boolean>): Single<T> = this
    .doOnSubscribe { progress.value = true }
    .doFinally { progress.value = false }

fun Completable.withProgress(progress: MutableState<Boolean>): Completable = this
    .doOnSubscribe { progress.value = true }
    .doFinally { progress.value = false }

fun <T : Any> Observable<T>.withProgress(progress: RxValueMutable<Boolean>): Observable<T> = this
    .doOnSubscribe { progress.set(true) }
    .doFinally { progress.set(false) }

fun <T : Any> Single<T>.withProgress(progress: RxValueMutable<Boolean>): Single<T> = this
    .doOnSubscribe { progress.set(true) }
    .doFinally { progress.set(false) }

fun Completable.withProgress(progress: RxValueMutable<Boolean>): Completable = this
    .doOnSubscribe { progress.set(true) }
    .doFinally { progress.set(false) }

fun Completable.withMultiSourceProgress(source: MultiSourceProgress): Completable = this
    .doOnSubscribe { source.addProgressProcess() }
    .doFinally { source.removeProgressProcess() }

fun <T : Any> Observable<T>.toMainSwitchIfNeeded(): Observable<T> = this
    .switchMap {
        if (isMainThread()) {
            Observable.just(it)
        } else {
            Observable.just(it).toMain()
        }
    }

fun <T : Any> Observable<Optional<T>>.filterNotNull(): Observable<T> = this
    .filter { it.isPresent }
    .map { it.value!! }

fun <T : Any> Single<T>.toMainSwitchIfNeeded(): Single<T> = this
    .flatMap {
        if (isMainThread()) {
            Single.just(it)
        } else {
            Single.just(it).toMain()
        }
    }

fun <T : Any> Single<T>.assert(statement: (T) -> Boolean): Single<T> = this
    .flatMap {
        if (statement(it)) {
            Single.just(it)
        } else {
            Single.error(RuntimeException("Statement is wrong"))
        }
    }

    
    
    
class MultiSourceProgress {

    private val _isProgress = RxValueMutable(false)
    val isProgress = _isProgress.asImmutable()

    private var processes = 0

    fun addProgressProcess() {
        processes += 1
        _isProgress.setDistinct(processes > 0)
    }

    fun removeProgressProcess() {
        processes -= 1
        _isProgress.setDistinct(processes > 0)
    }
}






    @Synchronized
    fun setDistinctOnMain(newValue: T) {
        setRunnable?.let {
            mainHandler.removeCallbacks(it)
            setRunnable = null
        }
        if (isMainThread()) {
            setDistinct(newValue)
        } else {
            setRunnable = Runnable {
                setDistinct(newValue)
                setRunnable = null
            }
            mainHandler.post(setRunnable!!)
        }
    }