I'm converting a callback-based API into a flow. I...

# coroutines

[diesieben07]

I'm converting a callback-based API into a flow. I need to signal to the API if the flow is cancelled or fails with an exception.
I know about onCompletion, but it has no information about if the flow was completely consumed, or if consumption was cancelled using e.g. take operator.
onCompletion also has no context information:

flow {
  val callback = MyCallbackHandler(this@flow)
  api.call(callback)
}.onCompletion { t ->
  // Was the stream cancelled? I don't know.
  // and how do I access callback here to (maybe) tell it to cancel the request
}

[Dominaezzz]

You might have to use callbackFlow with awaitClose.
How are you able to keep your flow suspended btw?

[Dominaezzz]

Oh it's you. You can't wrap your while loop in a try/finally. Assuming you haven't changed your code.

[diesieben07]

Yes, I know I can't do that. well, finally is okay, but not catch, because that breaks flow cancellation.
But with finally I still don't know whether the stream got cancelled or completed fully. And I don't have the exception that may have occurred either.
callbackFlow + awaitClose does not work. The awaitClose code is not called if the flow is cancelled (e.g. using take(1))

[Dominaezzz]

It's not called?!

[diesieben07]

val myFlow = callbackFlow {
        send(1)
        send(2)
        delay(200)
        send(3)
        close()
        awaitClose {
            println("await close")
        }
    }.onCompletion {
        println("on completion!")
        it?.printStackTrace()
    }

    myFlow.take(1).collect {
        println("got value $it")
    }

Prints:

got value 1
on completion!

[Dominaezzz]

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/await-close.html

[diesieben07]

I know the documentation. It seems to be wrong for channels in flows...

[Dominaezzz]

Yeah, but you close it though, before awaitClose was called.

[diesieben07]

If I remove close the flow never finishes.

[diesieben07]

Which makes sense, because the channel does not get closed.

[Dominaezzz]

Even with the take ?

[diesieben07]

Well, take cancels the flow so it will finish. But awaitClose is still not called (how could it, the producing lambda hasn't even gotten there...

[diesieben07]

I can surround my producer with try-finally, but then I am back to square one (don't know if it was cancelled, don't have the failing exception).

[Dominaezzz]

Use try/catch and rethrow the exception.

[diesieben07]

Still no way to tell if it was simply cancelled or if it's a genuine exception. There is no way to check for the cancellation exception, because it's class is marked internal.

[diesieben07]

val myFlow = callbackFlow {
        try {
            send(1)
            send(2)
            delay(200)
            send(3)
            close()
        } catch (e: Exception) {
            println("Failure! ... not really")
            e.printStackTrace()
        }
    }

    myFlow.take(1).collect {
        println("got value $it")
    }

Shows:

got value 1
Failure! ... not really
kotlinx.coroutines.flow.internal.AbortFlowException: Flow was aborted, no more elements needed

[diesieben07]

And there is no way to check for AbortFlowException, because it's internal.

[Dominaezzz]

On second thought, does the difference matter? Why does the producer need to know how (as supposed to just when) the consumer failed?

[diesieben07]

apiCall.cancel() as opposed to apiCall.fail(exception).

[diesieben07]

The first one is a perfectly normal condition "I don't need more elements, thanks".
The 2nd one is "something went wrong"

[Dominaezzz]

I still don't understand why it needs to know? As far as Flow is concerned here, only the first one is possible. Flow just doesn't provide the second one.

[Dominaezzz]

I mean, what if your Flow had multiple consumers and one fails, should the rest fail because of it?

[diesieben07]

Each flow only has one collector... You can't call two terminal operators on one flow.

[diesieben07]

Well, you can, but they will just make your emitter code run twice.

[Dominaezzz]

flow {}.broadcastIn() ...... , which can have multiple consumers. There will be a share operator coming soon also, that will basically allow flows have multiple collectors.

[diesieben07]

Yes, but broadcastIn is still one terminal operator, which collects the flow once.

[diesieben07]

As far as the flow is concerned, there is one consumer, which just happens to broadcast things.

[Dominaezzz]

Well, yes, but you could say that about literally every operator. Every operator is a consumer which happens to emit things as another Flow.

[Dominaezzz]

Regardless, why should you get downstream exceptions?

[diesieben07]

I don't know. Should I? I am not sure I understand the point you are making.

[Dominaezzz]

Sorry. I'm not very good at explaining things.

[Dominaezzz]

Flow implementations never catch or handle exceptions that occur in downstream flows.

You shouldn't handle the exceptions.

[diesieben07]

Okay. So just cancel without any information

[Dominaezzz]

Yes

[diesieben07]

This logs a warning from the API...

Cancelling without a message or cause is suboptimal

[Dominaezzz]

🤷🏼 You can pass the cancellation exception as a cause I guess, idk. Weird API imo.

[diesieben07]

grpc 😄

[diesieben07]

val myFlow = flow {
        var done = false
        try {
            emit(1)
            emit(2)
            delay(200)
            emit(3)
            println("closed")
            done = true
        } finally {
            if (!done) {
                println("cancelled!")
            }
        }
    }

Seems to work