ktor websocket flow retry mechanism

MessageService.kt

@Serialized
data class MessageDto(
    @SerialName(value = "id")
    val id: String,
    @SerialName(value = "message")
    val message: String
)

class MessageService constructor(
    private val httpClient: HttpClient
) {

    private val sendMessageFlow: Flow<MessageDto> = MutableSharedFlow()
    
    fun sendMessage(messageDto) {
        sendMessageFlow.emit(messageDto)
    }

    fun messages(): Flow<MessageDto> = flow {
        httpClient.wss(
            method = HttpMethod.Get,
            host = getHostname(),
            port = getPort(),
            path = "/v1/messages"
        ) {
            coroutineScope {
                sendMessageFlow
                    .conflate()
                    .map(::sendSerialized)
                    .launchIn(this)
                while (true) {
                    emit(receiveDeserialized<MessageDto>())
                }
            }
        }
    }.retryWithBackoff()
        .flowOn(coroutineDispatcher)
}

retry.kt

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
    }
}