Retry Flowable method on Error (RxJava 2) - Android (Kotlin)

RetryWithDelay.kt

//Class 
import io.reactivex.Flowable
import io.reactivex.functions.Function
import java.util.concurrent.TimeUnit

class RetryWithDelay(private val maxRetries: Int, private val retryDelayMillis: Int, private val retryIf: Function<Throwable, Boolean>, private val unit: TimeUnit = TimeUnit.MILLISECONDS) : Function<Flowable<out Throwable>, Flowable<*>> {
    private var retryCount: Int = 0

    override fun apply(attempts: Flowable<out Throwable>): Flowable<*> {
        return attempts.doOnNext { println("RetryWithDelay $it") }.flatMap { throwable ->
            if (retryIf.apply(throwable) && (maxRetries < 0 || ++retryCount < maxRetries)) {
                // When this Flowable calls onNext, the original Flowable will be retried (i.e. re-subscribed).
                Flowable.timer(retryDelayMillis.toLong(), unit)
            } else Flowable.error<Any>(throwable) // Max retries hit. Just pass the error along.
        }
    }
}

//Extensions 
import io.reactivex.Flowable
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.exceptions.CompositeException
import io.reactivex.functions.Function
import io.reactivex.schedulers.Schedulers

fun <T> Flowable<T>.retryOnError(): Flowable<T> = this
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .onErrorResumeNext { throwable: Throwable -> if (throwable is CompositeException) Flowable.error(throwable.exceptions[0]) else Flowable.error(throwable) }
    .retryWhen(RetryWithDelay(3, 1000, Function { t -> t is ConnectException }))


//Usages
repository.getSomething()
            .retryOnError()
            .subscribe({
                //
            }, { error ->
                //
            }).add()