Kotlin and Rx sample

ExerciseEditInteractor.kt

package com.mig35.training.business.exercises.details

import com.mig35.training.business.common.BaseProfileLoadingInteractor
import com.mig35.training.data.model.exercise.Exercise
import com.mig35.training.data.model.rights.Rights
import com.mig35.training.data.model.user.PublicProfile
import com.mig35.training.data.repository.IAuthRepository
import com.mig35.training.data.repository.IExerciseRepository
import com.mig35.training.data.repository.IRightsRepository
import io.reactivex.Completable
import io.reactivex.Observable
import io.reactivex.Single

class ExerciseEditInteractor(private val public: Boolean,
                             publicProfile: PublicProfile?,
                             authRepository: IAuthRepository,
                             private val exercisesRepository: IExerciseRepository,
                             private val rightsRepository: IRightsRepository) :
        BaseProfileLoadingInteractor(publicProfile, authRepository), IExerciseEditInteractor {

    override fun canUserEditExercise(): Observable<Boolean> {
        return profileUid
                .flatMapObservable { rightsRepository.getUserRights(it) }
                .map { t ->
                    if (public) {
                        t.contains(Rights.ADMIN) || t.contains(Rights.MANAGER)
                    } else {
                        true
                    }
                }
                .onErrorReturnItem(false)
    }

    override fun saveExercise(exercise: Exercise): Single<Exercise> =
            if (public)
                exercisesRepository
                        .savePublicExercise(exercise)
            else
                profileUid
                        .flatMap { exercisesRepository.savePrivateExercise(exercise, it) }

    override fun removeExercise(exercise: Exercise): Completable =
            if (public) {
                exercisesRepository
                        .removePublicExercise(exercise)
            } else {
                profileUid
                        .flatMapCompletable { exercisesRepository.removePrivateExercise(exercise, it) }
            }
}

LaunchInteractor.kt

package com.mig35.training.business.launch

import android.os.SystemClock
import com.mig35.training.data.repository.IAuthRepository
import com.mig35.training.data.repository.IConfigRepository
import com.mig35.training.ui.launch.model.LaunchNavigation
import com.mig35.training.utils.rx.general.RxSchedulersAbs
import io.reactivex.Completable
import io.reactivex.CompletableSource
import io.reactivex.Single
import io.reactivex.subjects.BehaviorSubject
import java.util.concurrent.TimeUnit

class LaunchInteractor(private val authRepository: IAuthRepository,
                       private val configRepository: IConfigRepository,
                       private val rxSchedulers: RxSchedulersAbs) : ILaunchInteractor {

    override fun getLaunchNavigation(skipWaiting: Boolean): Single<LaunchNavigation> {
        val startTime = SystemClock.elapsedRealtime()
        return Completable
                .ambArray(Completable.timer(MAX_SPLASH_WAIT_TIME, TimeUnit.MILLISECONDS), fetchRemoteConfig())
                .observeOn(rxSchedulers.computationScheduler)
                .andThen(authRepository.getCurrentPrivateUser())
                .map { LaunchNavigation.HOME }
                .onErrorResumeNext {
                    authRepository.hasEverLogin()
                            .map { aBoolean -> if (aBoolean) LaunchNavigation.LOGIN else LaunchNavigation.REGISTRATION }
                }
                .flatMap { launchNavigation ->
                    if (skipWaiting) {
                        Single.just(launchNavigation)
                    } else {
                        Single.just(launchNavigation)
                                .delay(MIN_SPLASH_WAIT_TIME - (SystemClock.elapsedRealtime() - startTime), TimeUnit.MILLISECONDS)
                    }
                }
    }

    private fun fetchRemoteConfig(): CompletableSource? {
        val publishSource = BehaviorSubject.create<Boolean>()
        configRepository
                .fetchRemoteConfig()
                .subscribeOn(rxSchedulers.ioScheduler)
                .onErrorComplete()
                .subscribe({ publishSource.onComplete() })
        return publishSource.ignoreElements()
    }

    companion object {

        private val MIN_SPLASH_WAIT_TIME = 1000L
        private val MAX_SPLASH_WAIT_TIME = 3000L
    }
}