Multi Distinguishing Unique and Duplicate Items.

[prakashelango]

I would like to identify duplicate and unique items separately and go through a different workflow.

        .select().distinct()
        .collect().asList()
        .await().indefinitely();
        
       Input ->  {1, 1, 2, 3, 4, 5, 5, 6, 1, 4, 4} 
       
       Current Output -> {1, 2, 3, 4, 5, 6}
       
       Expected 
       Duplicates -> {1, 5, 4} or in any order stating duplicate items
       Unique -> {1, 2, 3, 4, 5, 6}
        ```
        
from documentation https://smallrye.io/smallrye-mutiny/latest/guides/eliminate-duplicates-and-repetitions/ 
I don't find a way to classify both and accumulate them.

Currently we are accumulating into a hashmap and reiteration.

If some one can shed some lights it would be really grateful.

Let me know if way to go would be a custom operators and some insights on achieving it.

[jponge]

The distinct operator does eliminate duplicates, hence what you see is expected.

If you want to create a custom operator, see https://github.com/smallrye/smallrye-mutiny/blob/main/workshop-examples/src/main/java/_08_misc/_01_Multi_Custom_Operator.java and the plug operator.

Note that it will be your own duty to make sure that the operator respects the Reactive Streams protocol.

[prakashelango]

Going through quick explorations. Can you help on this please

public static void main(String[] args) {
        System.out.println("⚡️ Custom operator, pass through items and capture duplicates");

        Multi.createFrom().items("q", "b", "c","d", "q", "b", "c", "d")
                .plug(DuplicateItem::new)
                .map(i -> i +1)
                .subscribe().with(System.out::println);
    }

    static class DuplicateItem<T> extends AbstractMultiOperator<T, T> {
        public DuplicateItem(Multi<? extends T> upstream) {
            super(upstream);
        }

        @Override
        public void subscribe(MultiSubscriber<? super T> downstream) {
            upstream.subscribe().withSubscriber(new DuplicateItemProcessor(downstream));
        }

        private class DuplicateItemProcessor extends MultiOperatorProcessor<T, T> {
            private final Map<T, Integer> seenItems = new HashMap<>();
            private final Set<T> uniqueItems = new HashSet<>();
            private final Set<T> duplicateItems = new HashSet<>();

            DuplicateItemProcessor(MultiSubscriber<? super T> downstream) {
                super(downstream);
            }

            @Override
            public void onItem(T item) {
                seenItems.put(item, seenItems.getOrDefault(item, 0) + 1);
                if (seenItems.get(item) == 1) {
                    uniqueItems.add(item);
                } else if (seenItems.get(item) == 2) {
                    uniqueItems.remove(item);
                    duplicateItems.add(item);
                    System.out.println("Duplicate item: " + item);
                }
                super.onItem(item);
            }
        }
    }

1. Unsure to get uniqueItems and duplicateItems to process it accordingly in pipeline back.
2. Is there a way to make sure on sticking to Reactive Streams protocol?
3. will there be consideration on including this into Multi in long term?

[jponge]

1. You can always use the downstream reference to send items whenever you want
2. Yes: https://github.com/reactive-streams/reactive-streams-jvm/blob/master/tck-flow/README.md
3. No, and for in-memory stream processing we don't recommend using Mutiny anyway (java.util.stream provides everything you need for CPU-bound stream processing)

[prakashelango]

Thank you for the quick response.

As suggested I skipped all together and handled in stream processing.
3. Do you mind pointing to documentation, let me learn more.

One more ask
Is that reactive and imperative style of code in quarkus world will be treated as same performant in-par or will there be performance gains on reactive approach (on IO intensive and cpu bound task or overall)?
we have most of our code in reactive way.