ReactorMergeOperations { public static <K> Mono<Map<K, Optional<Throwable>>> merge(Map<K, Mono<Void>> monos, int concurrencyLimit, Scheduler scheduler) { List<Flux<Pair<K, Optional<Throwable>>>> m2 = new ArrayList<>(); monos.forEach((key, mono) -> { Flux<Pair<K, Optional<Throwable>>> x = mono.toProcessor().ignoreElement().materialize().map(result -> { Optional<Throwable> error = result.getType() == SignalType.ON_ERROR ? Optional.of(result.getThrowable()) : Optional.empty(); return Pair.of(key, error); } ).flux(); m2.add(x); }); return Flux.merge(Flux.fromIterable(m2), concurrencyLimit) .subscribeOn(scheduler) .collectList() .map(list -> list.stream().collect(Collectors.toMap(Pair::getLeft, Pair::getRight))); }  static Mono<Map<K, Optional<Throwable>>> merge(Map<K, Mono<Void>> monos, int concurrencyLimit, Scheduler scheduler);  }
@Test public void testMonoMerge() { Set<String> resultCollector = new HashSet<>(); Map<String, Mono<Void>> monoMap = ImmutableMap.<String, Mono<Void>>builder() .put("mono1", Mono.defer(() -> { resultCollector.add("mono1"); return Mono.empty(); })) .put("mono2", Mono.defer(() -> { resultCollector.add("mono2"); return Mono.empty(); })) .build(); Map<String, Optional<Throwable>> executionResult = ReactorExt.merge(monoMap, 2, Schedulers.parallel()).block(); assertThat(executionResult).hasSize(2); assertThat(executionResult.get("mono1")).isEmpty(); assertThat(executionResult.get("mono2")).isEmpty(); assertThat(resultCollector).contains("mono1", "mono2"); }
@Test public void testMonoMergeWithError() { Map<String, Mono<Void>> monoMap = ImmutableMap.<String, Mono<Void>>builder() .put("mono1", Mono.defer(Mono::empty)) .put("mono2", Mono.defer(() -> Mono.error(new RuntimeException("Simulated error")))) .put("mono3", Mono.defer(Mono::empty)) .build(); Map<String, Optional<Throwable>> executionResult = ReactorExt.merge(monoMap, 2, Schedulers.parallel()).block(); assertThat(executionResult).hasSize(3); assertThat(executionResult.get("mono1")).isEmpty(); assertThat(executionResult.get("mono2")).containsInstanceOf(RuntimeException.class); assertThat(executionResult.get("mono3")).isEmpty(); }