Remove usage of operator onErrorContinue for internal error/retry handling due to some cases of inconsistent behaviour.

Author: Daniel Bustamante Ospina

async/async-commons/src/main/java/org/reactivecommons/async/impl/listeners/GenericMessageListener.java

@@ -71,21 +71,25 @@ public void startListener() {
 
         setUpBindings(messageListener.getTopologyCreator()).thenMany(
                 receiver.consumeManualAck(queueName, consumeOptions)
-                        .transform(this::consumeFaultTolerant)
-                        .transform(this::outerFailureProtection))
+                        .transform(this::consumeFaultTolerant))
                 .subscribe();
     }
 
 
     private Mono<AcknowledgableDelivery> handle(AcknowledgableDelivery msj) {
-        String executorPath = getExecutorPath(msj);
-        final Function<Message, Mono<Object>> handler = getExecutor(executorPath);
-        final Message message = RabbitMessage.fromDelivery(msj);
+        try {
+            String executorPath = getExecutorPath(msj);
+            final Function<Message, Mono<Object>> handler = getExecutor(executorPath);
+            final Message message = RabbitMessage.fromDelivery(msj);
 
-        return defer(() -> handler.apply(message))
+            return defer(() -> handler.apply(message))
                 .transform(enrichPostProcess(message))
                 .transform(logExecution(executorPath))
                 .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()));
+            return Mono.error(e);
+        }
     }
 
     private Function<Mono<Object>, Mono<Object>> logExecution(String executorPath) {
@@ -102,22 +106,6 @@ private Function<Mono<Object>, Mono<Object>> logExecution(String executorPath) {
         };
     }
 
-    private <T> Flux<T> outerFailureProtection(Flux<T> messageFlux) {
-        return messageFlux.onErrorContinue(t -> true, (throwable, elem) -> {
-            if (elem instanceof AcknowledgableDelivery) {
-                try {
-                    String messageID = ((AcknowledgableDelivery) elem).getProperties().getMessageId();
-                    log.log(Level.SEVERE, format("ATTENTION !! Outer error protection reached for %s, in Async Consumer!! Severe Warning! ", messageID));
-                    requeueOrAck((AcknowledgableDelivery) elem, throwable).subscribe();
-                } catch (Exception e) {
-                    log.log(Level.SEVERE, "Error returning message in failure!", e);
-                }
-            } else {
-                throw new RuntimeException(throwable);
-            }
-        });
-    }
-
     private Flux<AcknowledgableDelivery> consumeFaultTolerant(Flux<AcknowledgableDelivery> messageFlux) {
         return messageFlux.flatMap(msj ->
                         handle(msj)