hi, guys. I cannot get the logic behind `retryWhen...

# rx

[vaskir]

hi, guys. I cannot get the logic behind retryWhen function. The following code is intended to pause for 500 ms after any upstream error, then resubscribe:

suspend fun main() {
    coroutineScope {
        val source = rxFlowable {
            for (i in 1..5) {
                if (i == 3)
                    throw Exception("It's $i and we are failing")
                else {
                    send(i)
                    delay(100)
                }
            }
        }

        source
            .retryWhen {
                rxFlowable<Int> {
                    println("pause...")
                    delay(500)
                }
            }
            .forEach { println(it) }
    }
}

but the output is:

pause...
1
2

Process finished with exit code 0

What I'm doing wrong?