'Filter a Flux with a HashSet cellected from another Flux

I have two flux coming from ReactiveMongoDB API: the first will gets the data of date1 the second one gets the data of date2 by the end i want to exclude elements that still appearing from the date1 in date2, In order to get those that are disappeared and save them in another collection. To deal with That I used a HashSet to collect the element of the second date (date2), so I can filter the flux of the first date (date2) by excluding the element in the hashset.

Flux<Report> reportOfCurrentDate = reportRepository.get(LocalDate.now());
Flux<Report> reportOfPrevDate = reportRepository.get(LocalDate.now().minusDays(1));

reportOfCurrentDate.log("insert In Set :").collect(Collectors.toSet());

// filtering 
reportOfPrevDate
.filterWhen(
                report -> resourceIds
                        .map(installedResId -> !installedResId.contains(report.getInstalledResInstalledResourceId()))
)
.transform(rereports -> resolvedRepository.insertAll(reports.collectList())
.blockLast();

Note that the flux is getting the data form a very large collection The problem is i get just one flux running (the one that insert in a set) it doesn't seem reactive the behavior expected is that the two fluxs will run in parallel, filtering data, and insert in the end in a new collection. any one can help me with that, or suggest another way of doing it ?



Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution Source