I have defined a `callbackFlow` looking somewhat ...

# coroutines

[niqo01]

I have defined a callbackFlow looking somewhat like this:

callbackFlow {
 val callback = object : Callback { // Implementation of some callback interface
        override fun onProgress(value: Int) {
            trySendBlocking(UploadStatus.Progress(value))

        }
        override fun onError(cause: Throwable) {
            trySendBlocking(UploadStatus.Error(cause))
            close()
        }
        override fun onCompleted(key: String){
            trySendBlocking(UploadStatus.Completed(key))
            close()
        }
    }
    api.register(callback)

    awaitClose { api.unregister(callback) }
}

My issue is that the UploadStatus.Completed or the UploadStatus.Error are not received by the collector. If I remove the close() those values are emitted but then my flow never stop.
What am I misunderstanding?

[Steven Veltema]

Try replacing close() with channel.close() .

[niqo01]

Same issue.

[Sam]

I don't see anything obvious in the code you shared that would cause the behaviour you describe. It looks like close() can only ever be called after those events have been sent. Maybe there is something going wrong in the way the flow is later transformed or collected?

[ross_a]

what's the return result of trySendBlocking?

[ross_a]

And is there a .conflate or .buffer between the callbackFlow and collector?

[niqo01]

Thanks to you all advice, I was able to figure it out.
the culprit is the sample(2.seconds) operator I am using later downstream.
I was also able to reproduce like so:

flow {
        repeat(10) {
            emit(it)
            delay(110)
        }
    }
        .sample(200)
        .onEach { println(it) }
        .collect()

It will print:

1
3
5
7
8

Missing the last value of 9.
The interesting thing is that the example above is taken from the official Kotlin doc: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/sample.html

In there example the output is

1, 3, 5, 7, 9

I think it's a mistake on their side.

[Sam]

Huh, that is indeed confusing. I can just about see why the example could produce different output if the timings end up slightly off, but... what on earth does "the latest element is not emitted if it does not fit into the sampling window" mean? Surely if an element overflows the sampling period, it just rolls into the next period?

[Sam]

Or is it trying to say that the final period is truncated as soon as the upstream flow terminates, and any elements that fall within the incomplete period are discarded? 🤔

[ross_a]

I think it's saying the latter, but not in the clearest manner

[ross_a]

And it lines up with the issue that was experienced here