how to do buffering on multi (by size and by time)?

[bensilvan]

I have a stream of messages, for example from Kafka, that after some transformations I want to create batches by size or time
to be able to use bulk operations of other services (like DBs for example)

for example, I want that every 100 messages will be collected into a list, that I will be able to call some IO bulk operation on it.
it is like Multi.Collect() but suitable for infinity stream.

Can I achieve that behavior in the current API? Or what is the alternative for that in Multiny?
I didn't find in the documentation how do that and I know that this behavior exists in Project Reactor, even with
the option to wait for size or a time (with respecting backpressure)

[jponge]

You might look into https://javadoc.io/doc/io.smallrye.reactive/mutiny/latest/io.smallrye.mutiny/io/smallrye/mutiny/groups/MultiGroup.html (multi.group()...).

Having more advanced windowing operators is on the radar, see #1279

[bensilvan]

Thanks!
So I can use Multi.group().intoLists().of(100) and by that turn a Multi<message> to Multi<List<Messages> and then perform a bulk operation on each list, is that correct?

and I saw this function in the documentation:
Multi<java.util.List<T>> of​(int size, java.time.Duration maximumDelay), this function is exactly what I was looking for
but is it respecting back pressure? for example, if a consumer is not able to handle this message rate and will not request enough
messages for that specified message rate (for example the maximumDelay happens before the consumer requested more elements and not enough messages were collected)
what will be the expected behavior?

[jponge]

Thanks! So I can use Multi.group().intoLists().of(100) and by that turn a Multi<message> to Multi<List<Messages> and then perform a bulk operation on each list, is that correct?

Yes, you will get chunks of up to 100 items lists.

and I saw this function in the documentation: Multi<java.util.List<T>> of​(int size, java.time.Duration maximumDelay), this function is exactly what I was looking for but is it respecting back pressure? for example, if a consumer is not able to handle this message rate and will not request enough messages for that specified message rate (for example the maximumDelay happens before the consumer requested more elements and not enough messages were collected) what will be the expected behavior?

It does respect back-pressure.