More control on event publishing

MAIF · Nov 22, 2023 · 3c90408 · 3c90408
1 parent 1b485b3
commit 3c90408
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/...-core-reactor/src/main/java/fr/maif/reactor/eventsourcing/ReactorKafkaEventPublisher.java b/...-core-reactor/src/main/java/fr/maif/reactor/eventsourcing/ReactorKafkaEventPublisher.java
@@ -99,7 +99,9 @@ public <TxCtx> void start(EventStore<TxCtx, E, Meta, Context> eventStore, Concur
                                         LOGGER.error("Error republishing events for topic %s retrying for the %s time".formatted(topic, ctx.totalRetries()), ctx.failure());
                                     })
                             )
-                            .switchIfEmpty(Flux.just(Tuple.empty()));
+                            .collectList()
+                            .map(__ -> Tuple.empty())
+                            .switchIfEmpty(Mono.just(Tuple.empty()));
                 })
                 .concatMap(__ ->
                         this.eventsSource.transform(publishToKafka(