ChannelFlow instances do not prefer conflation over buffering when fused

[kevincianfarini]

Describe the bug

The Flow.conflate() documentation states that:

Adjacent applications of conflate/buffer, channelFlow, flowOn and produceIn are always fused so that only one properly configured channel is used for execution. Conflation takes precedence over buffer() calls with any other capacity.

This is not the behavior I've observed when using a callbackFlow within my library.

What happened? What should have happened instead?

While working on a library I publish, I wanted to ensure that a flow I return from my public API is always conflated. It does not make sense for events to be buffered because, even though I was using a callbackFlow, the items within that flow describe state and not events. I thought I had achieved this behavior by applying the Flow.conflate operator to the flow I return, however I was able to disprove that this unconditionally always configures the underlying channel to be conflated in some unit tests. In my experience, adjacent applications of the Flow.buffer operator would reconfigure the underlying channel's capacity, but would not reconfigure the buffer overflow strategy.

Instead, I would expect calls to Flow.conflate to always take precedence over other fused operators as the documentation leads me to believe.

I am hoping that this isn't just treated as unclear documentation. From my perspective, I don't see much utility in allowing Flow consumers to buffer a flow which has previously been conflated.

Provide a Reproducer

// The flow that's returned from my public API. 
val original = channelFlow { awaitCancellation() }.conflate()

// The flow that's altered by the API consumer. 
val altered = original.buffer(3, onBufferOverflow = BufferOverflow.SUSPEND)

The above snippet will configure the underlying channel to have a capacity of three and a buffer overflow strategy of DROP_OLDEST. It seems like partial retention of the conflated policy is successful since DROP_OLDEST is retained, however the final configured capacity is inconsistent with what's documented.

[dkhalanskyjb]

A simplified answer: buffer(3) is simply a very fast consumer of your elements. You can't stop a consumer from consuming elements as quickly as you produce them, even if you think the older elements are not worth consuming. It's not exactly what happens when fusing operators, but it's close.

So, yes, this is not a bug but unclear documentation.

To elaborate, the mental model of operator chaining is that of a pipeline through which data moves.

.conflate() is a small basket with enough space for only one piece of data, and this basket is operated by a person that observes the incoming data, and if something arrives, throws away the contents of the basket (if there are any) and puts the new piece of data in its place.

.buffer(3, BufferOverflow.SUSPEND) is a larger basket, with a strong, strict person operating it. The person observes the incoming data; if there is enough space in the basket for the new data, the person puts the data into the basket, but if not, the person yells: "STOP", and the incoming data stops.

What happens when the large basket is put right behind the small basket? If there is space in the large basket, the strict person moves the data from the small basket; if not, the strict person forbids new data from entering the small basket, so the operator of the small basket throws away the incoming data when both baskets are full. So, what we observe is: the oldest element gets dropped, and 4 elements fit into the two baskets.

What fusion does is it allows throwing away all basket but the last one while effectively preserving the basket operators: if you are given a flow with a basket of size 64, you can demand that it's replaced by a small basket.

If your flow pipeline ends with a basket, a consumer of your flow can replace your basket with a larger or a smaller one, but you get to decide what the operator of the basket does to your flow.

[kevincianfarini]

I think on a philosophical level that line of reasoning makes some sense, however what are some real world use cases for this behavior?

There's at least one scenario I can think of that the following statement doesn't apply to:

buffer(3) is simply a very fast consumer of your elements. You can't stop a consumer from consuming elements as quickly as you produce them, even if you think the older elements are not worth consuming. It's not exactly what happens when fusing operators, but it's close.

I'm currently using a callbackFlow which I conflate in my library to deal with synchronously emitted elements from callbacks which are sent using trySend. In this scenario it's possible that my callback could by synchronously invoked multiple times before coroutine dispatch occurs. In such a case, even if I had a downstream consumer working as quickly as possible, these elements would still be conflated because they are sent to the channel before dispatch can occur. Within the above line of reasoning they'd be buffered.

I'm not trying to be a jerk and nitpick your reasoning, either. This is a real scenario I ran into when writing some unit tests for my library which then uncovered this behavior in the first place.