Support for ordered merge

[ivansenic]

As I understood by reading the guides, there is only option to merge and concatenate Multis. There is no option to do an ordered merge.

I know that in RxJava, this is available using the rxjava3-extensions and the Flowables class.

Is this a planned feature maybe? If not, is there any suggestion on what would be the best way to implement this by my own?

[cescoffier]

Hello!
Yes, at the moment, we only have 2 strategies.

I need to understand your use case. Ordering items can be problematic, as it could require that all the publishers emit an item first.

To implement this feature, you just need to implement the operator. The RX Java 3 code can give a good starting point.

[ayhanap]

I have a similar use case and struggling to achieve it with current operators. So my use case is as follows;

I have a Multi<> of objects that I want to enrich by doing db calls for each item. I also need to preserve the order.

1. (With concatenate) I use .onItem().transformToUniAndConcatenate() on the original multi. This keeps the order but does not call the transforming unis until the previous one completes. Makes total time consumed increase by the number of items.

2. (With merge) It does call the transforming unis asynchronously and sends them to downstream as the complete.

What I want to achieve is as items come from the original Multi, call the transforming unis asynchronously in parallel and merge/concatenate results into a new Multi preserving the order. Which will make the total time consumed equal to the longest single uni transformation operation.

[jponge]

That's to be expected: if you want to keep order then you can't just throw tons of concurrent async jobs in parallel and buffer results. If you want to improve the speedup then you should lean towards the merge operator and possibly tune the concurrency level.

[ayhanap]

What I ended up is to memoize the order before mapping the items on multi then collect and reorder the results. Sharing this as a reference.

Having the same object type through the transformations might be a specific use case but I believe it can be done with transformed objects using a custom operator.

        Multi<Feedback> queryMulti = feedbackRepository.queryFeedbacks(feedbackFilterQuery.getQueryString(), feedbackFilterQuery.getParameters(),feedbackFilter.getSortBy(), page);
        
        Multi<Feedback> replayedQuery = Multi.createBy().replaying().ofMulti(queryMulti);
        //I memoize the id's to reorder at the end
        Multi<ObjectId> resultOrder = replayedQuery.map(Feedback::getId);

        //These are the async operations
        Multi<Feedback> queryTransformations = replayedQuery
                .onItem().transformToUniAndMerge(this::fillFeedbackUserInfo)
                .onItem().transformToUniAndMerge(this::fillStatusInfo)
                .onItem().transformToUniAndMerge(this::fillTags);

        //Here I collect items from multi and return the previously meoized Multi containing id's mapped to actual Objects
        Multi<FeedbackDTO> orderedQueryResult = queryTransformations.collect().asMap((Feedback::getId))
                .toMulti()
                .flatMap(objectIdFeedbackMap -> resultOrder.map(objectIdFeedbackMap::get))
                .map(feedBackMapper::entityToDTO);