inline val defaultExceptions: Set<Class<out Exception>>
    get() = setOf(
        ClosedReceiveChannelException::class.java,
        ConnectException::class.java,
        UnresolvedAddressException::class.java
    )
/**
 *  Retries a `Flow<T>` indefinitely with exponential backoff
 *  
 *  @param delay -- delay in seconds between retries
 *  @param delayAtMost -- maximum delay in seconds after which this value is used to delay
 */
public fun <T> Flow<T>.retryWithBackoff(
    delay: Int = 1,
    delayAtMost: Int = 10,
    exceptions: Set<Class<out Exception>> = defaultExceptions
): Flow<T> {

    var _delay = delay

    return retryWhen { cause, attempt ->

        val shouldRetry = exceptions.any { (cause::class.java == it) }

        if (shouldRetry) {
            delay(_delay.seconds.inWholeMilliseconds)
            val next = _delay * 2
            _delay = if (next < delayAtMost) {
                next
            } else delayAtMost
            true
        } else false
    }
}