RxJava retryWhen メソッドの使用例

Main.kt

// RetryWithDelay 使用例
fun main(args: Array<String>) {
    Observable.create<String> { subscriber ->
        subscriber.onNext("test")
        subscriber.onError(Throwable("error occurred"))
    }
    .retryWhen(RetryWithDelay(3, 1000L))
    // toBlocking() や Thread.sleep() 的なものがないとメインスレッドが一瞬で終了してリトライ中の処理ごと死んでしまう
    // 実処理では blocking するわけに行かないので処理中に死んだ場合を考慮する必要があるかも
    //   -> Android なら onStop() での unsubscribe() ? 要調査
    .toBlocking()
    .subscribe({ println(it) }, { println(it) })
}

RetryWithDelay.kt

// 下記 stackoverflow の実装例を kotlin で書きなおしたもの
// @see http://stackoverflow.com/questions/22066481/rxjava-can-i-use-retry-but-with-delay
class RetryWithDelay(val maxRetries: Int, val retryDelayMillis: Long)
: Func1<Observable<out Throwable>, Observable<*>> {
    private var retryCount = 0
    override fun call(attempts: Observable<out Throwable>): Observable<*> {
        return attempts.cast(Throwable::class.java)
        .flatMap { throwable ->
            if (++retryCount < maxRetries) {
                Observable.timer(retryDelayMillis, TimeUnit.MILLISECONDS)
            } else {
                Observable.error(throwable)
            }
        }
    }
}