RxKotlin retryWhen & takeUntil example

RetryApiCall.kt

companion object {
        private const val COUNTER_START = 1
        private const val MAX_ATTEMPTS = 5
        private const val ORIGINAL_DELAY_IN_SECONDS = 2L
}

//Not important
val transactionJson = JsonObject().apply { addProperty(TRANSACTION_ID, transactionId) }
var disposable: Disposable? = null
        
disposable = serverClient.queryTransactionStatus(transactionJson)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            // Add a repeat condition, which will stop either onComplete() or when the max attempts have been reached
            .repeatWhen { observable ->
                observable.zipWith(
                    Observable.range(
                        COUNTER_START,
                        MAX_ATTEMPTS
                    )
                ) { _, attempt: Int -> attempt }
                    .flatMap { repeatAttempt ->
                        Log.v(TAG, "flatMap, call, repeatAttempt $repeatAttempt")
                        // Increase the waiting time
                        Observable.timer(repeatAttempt * ORIGINAL_DELAY_IN_SECONDS, TimeUnit.SECONDS)
                    }
            }
            // Add a predicate to stop the retrying earlier than MAX_ATTEMPTS
            .takeUntil { resp -> isTransactionFinalized(resp) }
            // Pass only the desired responses to onNext() in the subscribe block
            .filter { resp -> isTransactionFinalized(resp) }
            .subscribe(
                { response ->
                    if (response.isSuccessful) {
                        val result = response.body()
                        Log.v(TAG, "subscribe, call, isSuccessful, status = ${result?.transactionStatus}")
                        result?.let {
                            handleTransactionStatus(skuId, it.transactionStatus)
                        }
                    }
                    disposable?.dispose()
                },
                {
                    disposable?.dispose()
                }
            )

private fun isTransactionFinalized(transaction: Response<QueryTransactionResult>): Boolean {
        return if (transaction.isSuccessful) {
            val result = transaction.body()
            if (DEBUG) {
                Log.d(TAG, "isTransactionFinalized, call, isSuccessful, status = ${result?.transactionStatus}")
            }

            result?.transactionStatus != QueryTransactionResult.TransactionStatus.IN_PROGRESS
        } else {
            if (DEBUG) {
                Log.d(TAG, "isTransactionFinalized, call, !isSuccessful, result = false")
            }
            false
        }
    }