test(retries): local retries

test(retries): local retries

Author: juancgalvis

async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/listeners/GenericMessageListener.java

@@ -117,7 +117,14 @@ protected Mono<AcknowledgableDelivery> handle(AcknowledgableDelivery msj, Instan
             Mono<Object> flow = defer(() -> handler.apply(message))
                     .transform(enrichPostProcess(message));
             if (hasLocalRetries()) {
-                flow = flow.retryWhen(Retry.fixedDelay(maxRetries, retryDelay));
+                flow = flow.retryWhen(Retry.fixedDelay(maxRetries, retryDelay))
+                        .onErrorMap(err -> {
+                            if (err.getMessage() != null && err.getMessage().contains("Retries exhausted")) {
+                                log.warning(err.getMessage());
+                                return err.getCause();
+                            }
+                            return err;
+                        });
             }
             return flow.doOnSuccess(o -> logExecution(executorPath, initTime, true))
                     .subscribeOn(scheduler).thenReturn(msj);