I have this method. It sends a message out over a ...

# coroutines

[Chris Fillmore]

I have this method. It sends a message out over a websocket and waits for a reply.

suspend fun sendAndWait(message: Message): Message = withTimeout(5000) {
  launch {
    send(message)
  }
  // 'incoming' is a Flow<Message>
  incoming.first { /* logic here which matches outbound messages with inbound ones by some matching id */ }
}

The timeout never works. It never times out, I don’t get an exception. There are two suspension points: send(message) and incoming.first { }. Is there something I’m misunderstanding about timeouts? withTimeout has otherwise worked for me; I’m not sure what’s different in this case. Any ideas?

[Jacob]

I don't think send(message) counts as a suspension point when it's being `launch`ed

[Chris Fillmore]

Yeah I wasn’t sure about that. But calling Flow.first() should work, I thought

[simon.vergauwen]

Where does incoming come from? If the Flow<Message> is not cancellable, then`withTimeout` will not return before it's finished.

[Chris Fillmore]

incoming is a SharedFlow with incoming websocket frames. I am using Ktor/OkHttp websockets.

I understand that cancellation can occur at any suspension point. But I guess I am confused by the statement in the docs:

All the suspending functions in kotlinx.coroutines are cancellable.

I am expecting Flow.first() to be cancellable. But perhaps I need a workaround like

suspend fun sendAndWait(message: Message): Message = withTimeout(5000) {
  val job = async {
    launch {
      send(message)
    }
    
    incoming.first()
  }
  job.await()
}

[Chris Fillmore]

Btw thank you for responding, I appreciate the help

[Chris Fillmore]

The following simple program fails as I expect

fun main(args: Array<String>) = runBlocking {
    val incoming = MutableSharedFlow<Unit>()
    val result = withTimeout(3000) {
        incoming.first()
    }
    println("Got result: $result")
}

Exception in thread "main" kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 3000 ms

[simon.vergauwen]

I am expecting Flow.first() to be cancellable. But perhaps I need a workaround like

That might depend, for example I think the following is prone to hanging.

flow<Int> {
   withContext(NonCancellable) {
      delay(Long.MAX_VALUE)
      emit(1)
   }
}.first()

[julian]

@Chris Fillmore Are you able to reproduce this behavior in some standalone code i.e. outside of the project you're working on?

[Chris Fillmore]

No. The simple example I wrote above fails as expected

[Chris Fillmore]

If I spent more time on it I could perhaps repro, but I don’t have the time right now

[julian]

Got it. It's puzzling behavior. Seems there's something going on "outside the frame" of what we're looking at that we're not seeing, and that's causing this. I tried some variations on what you're trying to do and couldn't repro either.