Suspend coroutine until condition is true

user_8691695 0
Viewed 3187

I have a use case where I need to connect and disconnect from a class that acts as a service. Actions can be performed on the service only when the service is connected. Clients are notified when the service connects or disconnects by a callback:

class Service {

    constructor(callback: ConnectionCallback) { ... }

    fun connect() {
        // Call callback.onConnected() some time after this method returns.
    }

    fun disconnect() {
        // Call callback.onConnectionSuspended() some time after this method returns.
    }

    fun isConnected(): Boolean { ... }

    fun performAction(actionName: String, callback: ActionCallback) {
        // Perform a given action on the service, failing with a fatal exception if called when the service is not connected.
    }

    interface ConnectionCallback {
        fun onConnected() // May be called multiple times
        fun onConnectionSuspended() // May be called multiple times
        fun onConnectionFailed()
    }
}

I'd like to write a wrapper for that Service class (that I don't control) using Kotlin Coroutines. Here is a skeleton of ServiceWrapper:

class ServiceWrapper {
    private val service = Service(object : ConnectionCallback { ... })

    fun connect() {
        service.connect()
    }

    fun disconnect() {
        service.disconnect()
    }

    suspend fun performActionWhenConnected(actionName: String): ActionResult {
        suspendUntilConnected()

        return suspendCoroutine { continuation ->
            service.performAction(actionName, object : ActionCallback() {
                override fun onSuccess(result: ActionResult) {
                    continuation.resume(result)
                }

                override fun onError() {
                    continuation.resumeWithException(RuntimeException())
                }
            }
        }
    }
}

How can I implement this suspendUntilConnected() behavior using Coroutines ? Thanks in advance.

kotlin-coroutines kotlin
4 Answers

Here's how you can implement it:

class ServiceWrapper {
    @Volatile
    private var deferredUntilConnected = CompletableDeferred<Unit>()

    private val service = Service(object : ConnectionCallback {
        override fun onConnected() {
            deferredUntilConnected.complete(Unit)
        }

        override fun onConnectionSuspended() {
            deferredUntilConnected = CompletableDeferred()
        }
    })

    private suspend fun suspendUntilConnected() = deferredUntilConnected.await()

    ...
}

A general note: just because the service got connected at a certain point doesn't guarantee it will still be connected by the time you use it.

user_11038720

Another approach with StateFlow

  1. Define some "state" first
enum class ServiceState {
    CONNECTED, SUSPENDED, FAILED
}
  1. Convert the callback-based code to flow
val connectionState = MutableStateFlow(ServiceState.FAILED)

private val service = Service(object : ConnectionCallback {
    override fun onConnected() {
        connectionState.value = ServiceState.CONNECTED
    }

    override fun onConnectionSuspended() {
        connectionState.value = ServiceState.SUSPENDED
    }

    override fun onConnectionFailed() {
        connectionState.value = ServiceState.FAILED
    }
})
  1. Copy and paste these utils code
class ConditionalAwait<T>(
    private val stateFlow: StateFlow<T>,
    private val condition: (T) -> Boolean
) {
    suspend fun await(): T {
        val nowValue = stateFlow.value
        return if (condition(nowValue)) {
            nowValue
        } else {
            stateFlow.first { condition(it) }
        }
    }
}

suspend fun <T> StateFlow<T>.conditionalAwait(condition: (T) -> Boolean): T =
    ConditionalAwait(this, condition).await()
  1. USE IT~
suspend fun performActionWhenConnected() {
    connectionState.conditionalAwait { it == ServiceState.CONNECTED }
    // other actions when service is connected
}
  1. Advanced usage
suspend fun performActionWhenConnected() {
    val state = connectionState.conditionalAwait { 
        it == ServiceState.CONNECTED || it == ServiceState.FAILED
    } // keep suspended when Service.SUSPENDED

    if (state is ServiceState.CONNECTED) {
        // other actions when service is connected
    } else {
        // error handling
    }
}
user_29971280
  1. suspendCoroutine suspends the currently running coroutine, so no need for a pre-suspension.

Obtains the current continuation instance inside suspend functions and suspends the currently running coroutine.

  1. On a different note, I suggest using async and return a Deferred. It's not really the job of the ServiceWrapper to suspend. The caller should take control of when the result is need and therefore, when to suspend if needed.
class ServiceWrapper {

    ...

    fun performAction(actionName: String): Deferred<ActionResult> = coroutineScope.async {
        suspendCoroutine { continuation ->
            service.performAction(actionName, object : ActionCallback() {
                override fun onSuccess(result: ActionResult) {
                    continuation.resume(result)
                }

                override fun onError() {
                    continuation.resumeWithException(RuntimeException())
                }
            }
        }
    }
}
  1. If for some reason, you have to suspend before suspendCoroutine, I suggest using a CompletableJob. Deferred should return a result. Using Deferred<Unit> seems like an anti-pattern from my point of view.
user_28281000

You're in a suspended function, why not something like:

while (!service.isConnected()) {
    delay(1000)
}

You can put additional timeout conditions into this statement.

user_62758440
Related