Kotlinx.coroutines: IllegalStateException: Already resumed

[Tuubz]

In the following piece of code, I would like to call resume(...) twice, but it leads to IllegalStateException: Already resumed.

To give you a bit of context, I'm creating a chat application:

1. Send a message by executing groupChannel.sendUserMessage(message)
2. It immediately returns a UserMessage object, called pendingUserMessage
3. Execute continuation.resume(pendingUserMessage) > This will add the object to the recycler-view adapter in a different class
4. Once the call-back finishes, it returns again another UserMessage object
5. Execute continuation.resume(userMessage) / continuation.resumeWithException(exception) > This will decide whether the message has been sent successfully or not, unfortunately this will lead to IllegalStateException: Already resumed

class UserMessageRepository {

    suspend fun sendUserMessage(
        groupChannel: GroupChannel,
        message: String
    ): UserMessage {
        return suspendCoroutine { continuation ->
            val pendingUserMessage = groupChannel.sendUserMessage(message) { userMessage, exception ->
                if (exception == null) {
                    continuation.resume(userMessage)
                } else {
                    continuation.resumeWithException(exception)
                }
            }
            continuation.resume(pendingUserMessage)
        }
    }
}

Any other way to achieve this? Thanks in advance 🙏

[elizarov]

Please, using either a Channel or a Flow to send multiple messages. See the corresponding sections in the docs:

• https://kotlinlang.org/docs/reference/coroutines/channels.html
• https://kotlinlang.org/docs/reference/coroutines/flow.html
  Does it help?

[Tuubz]

@elizarov Thank you for your quick reply, I searched about channels and flows, and how they are different, eventually I ended up finding out about callBackFlow:

class UserMessageRepository {

    @ExperimentalCoroutinesApi
    fun sendUserMessage(
        groupChannel: GroupChannel,
        message: String
    ): Flow<UserMessage> {
        return callbackFlow {
            val pendingUserMessage = groupChannel.sendUserMessage(message) { userMessage, exception ->
                if (exception == null) {
                    offer(userMessage)
                } else {
                    close(exception)
                }
            }
            offer(pendingUserMessage)
        }
    }
}

Could you tell me if I'm on the right track here? 🙏

[elizarov]

Yes. However, note that offer may lose messages. Use sendBlocking if you don't want messages to be lost.

[Tuubz]

@elizarov Hmmm, I've noticed that this part of @ExperimentalCoroutinesApi. Does it mean we should avoid making use of callbackFlow, at this current stage?

[elizarov]

callbackFlow is going to be stable soon.

[Tuubz]

Got it~ 🚀

This is my final solution:

  @ExperimentalCoroutinesApi
    override fun sendUserMessage(
        groupChannel: GroupChannel,
        message: String
    ): Flow<UserMessage> {
        return callbackFlow {
            val pendingUserMessage = groupChannelEntity.sendUserMessage(message) { userMessage, exception ->
                if (exception == null) {
                    offer(userMessage)
                    close()
                } else {
                    close(exception)
                }
            }
            offer(pendingUserMessage)
            awaitClose { cancel() }
        }
    }

I have one more question, if I may :

Is there a similar function that achieves the same behaviour as resumeWithException(...) > Continuation.kt ? I was hoping that close(exception) would do the same, but it does not. Any other suggestion is also welcome 🙇

Thanks in advance 🙏

[elizarov]

What's wrong with close(exception)?

[Tuubz]

Sorry for the misunderstanding, my idea was to throw an actual exception. I just found out I can use cancel() instead, as mentioned here: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/callback-flow.html