I m converting a callback based API into a flow I need to si

Question

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 gt Was the stream cancelled I don t know and how do I access callback here to maybe tell it to cancel the request ```

Dominaezzz · Answer

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

Dominaezzz · Answer

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

diesieben07 · Answer

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 · Answer

It s not called

diesieben07 · Answer

``` 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 · Answer

diesieben07 · Answer

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

Dominaezzz · Answer

Yeah but you close it though before awaitClose was called

diesieben07 · Answer

If I remove `close` the flow never finishes

diesieben07 · Answer

Which makes sense because the channel does not get closed

Dominaezzz · Answer

Even with the `take`

diesieben07 · Answer

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 · Answer

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 · Answer

Use try catch and rethrow the exception

diesieben07 · Answer

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 · Answer

``` 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 · Answer

And there is no way to check for `AbortFlowException` because it s internal

Dominaezzz · Answer

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

diesieben07 · Answer

`apiCall cancel ` as opposed to `apiCall fail exception `

diesieben07 · Answer

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

Dominaezzz · Answer

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 · Answer

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

diesieben07 · Answer

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

diesieben07 · Answer

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

Dominaezzz · Answer

`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 · Answer

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

diesieben07 · Answer

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

Dominaezzz · Answer

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 · Answer

Regardless why should you get downstream exceptions

diesieben07 · Answer

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

Dominaezzz · Answer

Sorry I m not very good at explaining things

Dominaezzz · Answer

gt Flow implementations never catch or handle exceptions that occur in downstream flows You shouldn t handle the exceptions

diesieben07 · Answer

Okay So just cancel without any information

Dominaezzz · Answer

Yes

diesieben07 · Answer

This logs a warning from the API ```Cancelling without a message or cause is suboptimal```

Dominaezzz · Answer

shrug skin tone 3 You can pass the cancellation exception as a cause I guess idk Weird API imo

diesieben07 · Answer

grpc smile

diesieben07 · Answer

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