Hey guys, does somebody know why `capacity` and `o...

# coroutines

[David Bieregger]

Hey guys,
does somebody know why capacity and onBufferOverflow is not exposed in channelFlow ? Following usecase:

suspend fun receiveOutgoingTransactionsAsFlow(capacity: Int): Flow<EventTransaction> = channelFlow(capacity = capacity /* this parameter doesn't exist */) {
        launch {
            sharedFlow.collect { item ->
                trySend(item).getOrElse {
                    close(Error("Private buffer of size $capacity overflowed."))
                }
            }
        }
    }

This logic ensures that subscribers of a SharedFlow can process independently, but also accounts for the fact that the collector might be too slow. The capacity parameter would be really useful in this case or am I overcomplicating things?

[mbonnin]

Adjacent applications of channelFlow, flowOn, buffer, and produceIn are always fused so that only one properly configured channel is used for execution

https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/channel-flow.html

[mbonnin]

Sorry for the mobile formating but you can call .buffer() on the resulting flow

[David Bieregger]

Sorry for the confusion I might have created by using channel. . This means I didn't create another channel and therefore I see no contradiction to this statement.

Unfortunately .buffer() doesn't allow me to handle the buffer overflow myself.

[ross_a]

I think you can use onUndeliveredElement with DROP_LATEST

[ross_a]

though you might not know the specific reason - e.g. closure

[ross_a]

I have ended up with code like the above to detect slow consumers however