Rx retry with delay operator

RxRetryExtensions.kt

private const val REPO_DEFAULT_RETRY_COUNT = 5 // 62 sec total ~1 min
private val zipper = BiFunction { t: Throwable, u: Int -> Pair(t, u) }

fun Completable.networkStateRetry(retryAttempts: Int = REPO_DEFAULT_RETRY_COUNT): Completable =
    retryWhen {
        // retryAttempts + 1 для того чтобы была выброшена ошибка
        it.zipWith(Flowable.range(1, retryAttempts + 1), zipper)
            .flatMap { result ->
                val delayTime = 1L * 2.0.pow(result.second.toDouble()).roundToLong()
                return@flatMap (
                    if (
                        result.first is NetworkErrorException &&
                        result.second <= retryAttempts
                    ) {
                        Timber.d("PendingStatusException")
                        Timber.d("retry: ${result.second}")
                        Timber.d("delayTime: $delayTime")
                        Flowable.just(result.second)
                    } else {
                        Timber.d(result.first)
                        Flowable.error(result.first) // сработает сразу без ожидания в delayTime
                    }
                    )
                    .delay(delayTime, TimeUnit.SECONDS)
            }
    }

fun <T: Any> Single<T>.networkStateRetry(retryAttempts: Int = REPO_DEFAULT_RETRY_COUNT): Single<T> =
    retryWhen {
        // One more in range to prevent zip from closing right away with request still in process.
        it.zipWith(Flowable.range(1, retryAttempts + 1), zipper)
            .flatMap { result ->
                val delayTime = 1L * 2.0.pow(result.second.toDouble()).roundToLong()
                return@flatMap (
                    if (
                        result.first is NetworkErrorException &&
                        result.second <= retryAttempts
                    ) {
                        Timber.d("PendingStatusException")
                        Timber.d("retry: ${result.second}")
                        Timber.d("delayTime: $delayTime")
                        Flowable.just(result.second)
                    } else {
                        Timber.d(result.first)
                        Flowable.error(result.first) // сработает сразу без ожидания в delayTime
                    }
                    )
                    .delay(delayTime, TimeUnit.SECONDS)
            }
    }

/**
 * Starts retrying with increased times (2, 4, 8, 16 etc.) but stops increasing time after certain amount of retries.
 * After that keeps retrying indefinitely with the same delay.
 * For example with param = 3 - retries would after 2, 4, 8, 16, 16, 16, 16, 16 etc seconds.
 */
fun <T> Single<T>.networkStateRetryIndefinitely(timeIncreaseCount: Int = REPO_DEFAULT_RETRY_COUNT): Single<T> =
    retryWhen {
        var counter = 1
        it.flatMap { result ->
            if (result is NetworkErrorException) {
                val delayTime = 2.0.pow(counter).roundToLong()
                if (counter <= timeIncreaseCount) counter++
                Timber.d("PendingStatusException delayTime: $delayTime")
                Flowable.just(Unit).delay(delayTime, TimeUnit.SECONDS)
            } else {
                Timber.d(result)
                Flowable.error(result)
            }
        }
    }