fix(retries): local retries

fix(retries): local retries

Author: juancgalvis

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

@@ -17,6 +17,7 @@
 import reactor.rabbitmq.AcknowledgableDelivery;
 import reactor.rabbitmq.ConsumeOptions;
 import reactor.rabbitmq.Receiver;
+import reactor.util.retry.Retry;
 
 import java.time.Duration;
 import java.time.Instant;
@@ -69,6 +70,10 @@ private static long resolveRetries(boolean useDLQRetries, long maxRetries) {
         return useDLQRetries && maxRetries == -1 ? DEFAULT_RETRIES_DLQ : maxRetries;
     }
 
+    private boolean hasLocalRetries() {
+        return !useDLQRetries && maxRetries != -1;
+    }
+
     protected Mono<Void> setUpBindings(TopologyCreator creator) {
         return Mono.empty();
     }
@@ -109,9 +114,12 @@ protected Mono<AcknowledgableDelivery> handle(AcknowledgableDelivery msj, Instan
             final Function<Message, Mono<Object>> handler = getExecutor(executorPath);
             final Message message = RabbitMessage.fromDelivery(msj);
 
-            return defer(() -> handler.apply(message))
-                    .transform(enrichPostProcess(message))
-                    .doOnSuccess(o -> logExecution(executorPath, initTime, true))
+            Mono<Object> flow = defer(() -> handler.apply(message))
+                    .transform(enrichPostProcess(message));
+            if (hasLocalRetries()) {
+                flow = flow.retryWhen(Retry.fixedDelay(maxRetries, retryDelay));
+            }
+            return flow.doOnSuccess(o -> logExecution(executorPath, initTime, true))
                     .subscribeOn(scheduler).thenReturn(msj);
         } catch (Exception e) {
             log.log(Level.SEVERE, format("ATTENTION !! Outer error protection reached for %s, in Async Consumer!! Severe Warning! ", msj.getProperties().getMessageId()));
@@ -194,16 +202,16 @@ private Mono<AcknowledgableDelivery> requeueOrAck(AcknowledgableDelivery msj, Th
         final boolean redeliver = msj.getEnvelope().isRedeliver();
         reportErrorMetric(msj, init);
         sendErrorToCustomReporter(err, rabbitMessage, redeliver || retryNumber > 0);
-        if (maxRetries != -1 && retryNumber >= maxRetries) {
+        if (hasLocalRetries() || retryNumber >= maxRetries) { // Discard
             logError(err, msj, FallbackStrategy.DEFINITIVE_DISCARD);
             return discardNotifier
                     .notifyDiscard(rabbitMessage)
                     .doOnSuccess(_a -> msj.ack()).thenReturn(msj);
-        } else if (useDLQRetries) {
+        } else if (useDLQRetries) { // DLQ retries
             logError(err, msj, FallbackStrategy.RETRY_DLQ);
             msj.nack(false);
             return Mono.just(msj);
-        } else {
+        } else { // infinity fast retries
             logError(err, msj, FallbackStrategy.FAST_RETRY);
             return Mono.just(msj).delayElement(retryDelay).doOnNext(m -> m.nack(true));
         }

docs/docs/reactive-commons/10-wildcards.md

@@ -99,6 +99,14 @@ The next code will help you to avoid unexpected behaviors, which indicates you t
     }
 ```
 
+## Possible issues
+
+Unprocessed events will be notified with an event, so please handle it properly and ensure you don't have wildcards that match all events, because it can cause unexpected behaviors.
+
+For example if you have a handler for `my.event.#` and you start listening to `my.event`, it will match all events that start with `my.event`, so it will match `my.event.any` if your handler has dlq retries or has local retries, when your handler fails the retry limit, an event with name `my.event.any.dlq` will be sent.
+
+And if you have the same binding `my.event.#` the event will be handled by the handler, so you will have two different event structures for the same handler, and it can cause unexpected behaviors.
+
 ## Example
 
 You can see a real example at [samples/async/async-receiver-responder](https://github.com/reactive-commons/reactive-commons-java/tree/master/samples/async/async-receiver-responder)

samples/async/async-receiver-responder/src/main/java/sample/HandlersConfig.java

@@ -25,7 +25,7 @@ public class HandlersConfig {
     public HandlerRegistry handlerRegistrySubs(UseCase useCase) {
         animalEventEventHandler = event -> {
             log.info(event);
-            return Mono.empty();
+            return Mono.error(new RuntimeException("Not implemented"));
         };
         return HandlerRegistry.register()
                 .listenEvent(Constants.MEMBER_REMOVED, useCase::removeMember, RemovedMemberEvent.class)

samples/async/async-receiver-responder/src/main/resources/application.yaml

@@ -7,4 +7,5 @@ app:
   async:
     createTopology: true
     listenReplies: false
+    maxRetries: 3