Merge branch 'recovery'

Daniel Bustamante Ospina · Daniel Bustamante Ospina · commit 03aa80441652 · 2021-05-13T12:01:05.000-05:00
diff --git a/.gitignore b/.gitignore
@@ -642,4 +642,6 @@ MigrationBackup/
 .ionide/
 
 # End of https://www.toptal.com/developers/gitignore/api/macos,linux,windows,gradle,java,intellij,visualstudio,eclipse
-contiperf-report
+contiperf-report
+
+samples/async/local-example/
diff --git a/async/async-commons/src/main/java/org/reactivecommons/async/commons/ext/CustomErrorReporter.java b/async/async-commons/src/main/java/org/reactivecommons/async/commons/ext/CustomErrorReporter.java
diff --git a/async/async-commons/src/main/java/org/reactivecommons/async/commons/utils/ArrayUtils.java b/async/async-commons/src/main/java/org/reactivecommons/async/commons/utils/ArrayUtils.java
@@ -0,0 +1,16 @@
+package org.reactivecommons.async.commons.utils;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+
+public class ArrayUtils {
+
+    private ArrayUtils(){}
+
+    public static <E> Object[] prefixArray(E head, E[] tail) {
+        final ArrayList<E> objects = new ArrayList<>(1 + tail.length);
+        objects.add(head);
+        objects.addAll(Arrays.asList(tail));
+        return objects.toArray();
+    }
+}
diff --git a/async/async-commons/src/main/java/org/reactivecommons/async/commons/utils/LoggerSubscriber.java b/async/async-commons/src/main/java/org/reactivecommons/async/commons/utils/LoggerSubscriber.java
@@ -0,0 +1,48 @@
+package org.reactivecommons.async.commons.utils;
+
+import lombok.extern.java.Log;
+import reactor.core.publisher.BaseSubscriber;
+import reactor.core.publisher.SignalType;
+
+import java.util.logging.Level;
+
+import static org.reactivecommons.async.commons.utils.ArrayUtils.prefixArray;
+
+
+@Log
+public class LoggerSubscriber<T> extends BaseSubscriber<T> {
+
+    private final String flowName;
+    private static final String ON_COMPLETE_MSG = "%s: ##On Complete Hook!!";
+    private static final String ON_ERROR_MSG = "%s: ##On Error Hook!!";
+    private static final String ON_CANCEL_MSG = "%s: ##On Cancel Hook!!";
+    private static final String ON_FINALLY_MSG = "%s: ##On Finally Hook! Signal type: %s";
+
+    public LoggerSubscriber(String flowName) {
+        this.flowName = flowName;
+    }
+
+    @Override
+    protected void hookOnComplete() {
+        log.warning(format(ON_COMPLETE_MSG));
+    }
+
+    @Override
+    protected void hookOnError(Throwable throwable) {
+        log.log(Level.SEVERE, format(ON_ERROR_MSG), throwable);
+    }
+
+    @Override
+    protected void hookOnCancel() {
+        log.warning(format(ON_CANCEL_MSG));
+    }
+
+    @Override
+    protected void hookFinally(SignalType type) {
+        log.warning(format(ON_FINALLY_MSG, type.name()));
+    }
+
+    private String format(String msg, String... args) {
+        return String.format(msg, prefixArray(flowName, args));
+    }
+}
diff --git a/async/async-commons/src/test/java/org/reactivecommons/async/commons/utils/LoggerSubscriberTest.java b/async/async-commons/src/test/java/org/reactivecommons/async/commons/utils/LoggerSubscriberTest.java
@@ -0,0 +1,31 @@
+package org.reactivecommons.async.commons.utils;
+
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.SignalType;
+
+
+public class LoggerSubscriberTest {
+
+    private final LoggerSubscriber<String> subscriber = new LoggerSubscriber<>("testFlow");
+
+    @Test
+    public void shouldPrintOnCancelMessage() {
+        subscriber.hookOnCancel();
+    }
+
+    @Test
+    public void shouldPrintOnErrorMessage() {
+        subscriber.hookOnError(new RuntimeException());
+    }
+
+    @Test
+    public void shouldPrintOnFinallyMessage() {
+        subscriber.hookFinally(SignalType.ON_ERROR);
+    }
+
+    @Test
+    public void shouldPrintOnCompleteMessage() {
+        subscriber.hookOnComplete();
+    }
+
+}
diff --git a/async/async-rabbit-starter/src/test/java/org/reactivecommons/async/rabbit/config/CommandListenersConfigTest.java b/async/async-rabbit-starter/src/test/java/org/reactivecommons/async/rabbit/config/CommandListenersConfigTest.java
@@ -50,7 +50,7 @@ public void init() throws NoSuchFieldException, IllegalAccessException {
         when(creator.declare(any(ExchangeSpecification.class))).thenReturn(Mono.just(mock(AMQP.Exchange.DeclareOk.class)));
         when(creator.declareQueue(any(String.class), any())).thenReturn(Mono.just(mock(AMQP.Queue.DeclareOk.class)));
         when(listener.getTopologyCreator()).thenReturn(creator);
-        when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.empty());
+        when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.never());
         when(listener.getReceiver()).thenReturn(receiver);
         when(listener.getMaxConcurrency()).thenReturn(20);
     }
diff --git a/async/async-rabbit-starter/src/test/java/org/reactivecommons/async/rabbit/config/EventListenersConfigTest.java b/async/async-rabbit-starter/src/test/java/org/reactivecommons/async/rabbit/config/EventListenersConfigTest.java
@@ -44,7 +44,7 @@ public void init() {
         when(creator.declareQueue(any(String.class), any())).thenReturn(Mono.just(mock(AMQP.Queue.DeclareOk.class)));
         when(creator.declareQueue(any(String.class), any(String.class), any())).thenReturn(Mono.just(mock(AMQP.Queue.DeclareOk.class)));
         when(listener.getTopologyCreator()).thenReturn(creator);
-        when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.empty());
+        when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.never());
         when(listener.getReceiver()).thenReturn(receiver);
         when(listener.getMaxConcurrency()).thenReturn(20);
     }
diff --git a/async/async-rabbit-starter/src/test/java/org/reactivecommons/async/rabbit/config/NotificacionListenersConfigTest.java b/async/async-rabbit-starter/src/test/java/org/reactivecommons/async/rabbit/config/NotificacionListenersConfigTest.java
@@ -44,7 +44,7 @@ public void init() {
         when(creator.declareQueue(any(String.class), any())).thenReturn(Mono.just(mock(AMQP.Queue.DeclareOk.class)));
         when(creator.declareQueue(any(String.class), any(String.class), any())).thenReturn(Mono.just(mock(AMQP.Queue.DeclareOk.class)));
         when(listener.getTopologyCreator()).thenReturn(creator);
-        when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.empty());
+        when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.never());
         when(listener.getReceiver()).thenReturn(receiver);
         when(listener.getMaxConcurrency()).thenReturn(20);
     }
diff --git a/async/async-rabbit-starter/src/test/java/org/reactivecommons/async/rabbit/config/QueryListenerConfigTest.java b/async/async-rabbit-starter/src/test/java/org/reactivecommons/async/rabbit/config/QueryListenerConfigTest.java
@@ -46,7 +46,7 @@ public void init() {
         when(creator.declareQueue(any(String.class), any())).thenReturn(Mono.just(mock(AMQP.Queue.DeclareOk.class)));
         when(creator.declareQueue(any(String.class), any(String.class), any())).thenReturn(Mono.just(mock(AMQP.Queue.DeclareOk.class)));
         when(listener.getTopologyCreator()).thenReturn(creator);
-        when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.empty());
+        when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.never());
         when(listener.getReceiver()).thenReturn(receiver);
         when(listener.getMaxConcurrency()).thenReturn(20);
     }
diff --git a/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/listeners/ApplicationReplyListener.java b/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/listeners/ApplicationReplyListener.java
@@ -1,10 +1,13 @@
 package org.reactivecommons.async.rabbit.listeners;
 
+import com.rabbitmq.client.Delivery;
 import lombok.extern.java.Log;
+import org.reactivecommons.async.commons.utils.LoggerSubscriber;
 import org.reactivecommons.async.rabbit.RabbitMessage;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
 import org.reactivecommons.async.rabbit.communications.TopologyCreator;
 import org.reactivecommons.async.commons.reply.ReactiveReplyRouter;
+import reactor.core.publisher.Flux;
 import reactor.rabbitmq.Receiver;
 
 import java.util.logging.Level;
@@ -19,6 +22,7 @@ public class ApplicationReplyListener {
     private final Receiver receiver;
     private final TopologyCreator creator;
     private final String queueName;
+    private volatile Flux<Delivery> deliveryFlux;
 
     public ApplicationReplyListener(ReactiveReplyRouter router, ReactiveMessageListener listener, String queueName) {
         this.router = router;
@@ -28,7 +32,7 @@ public ApplicationReplyListener(ReactiveReplyRouter router, ReactiveMessageListe
     }
 
     public void startListening(String routeKey) {
-        creator.declare(exchange("globalReply").type("topic").durable(true))
+        deliveryFlux = creator.declare(exchange("globalReply").type("topic").durable(true))
             .then(creator.declare(queue(queueName).durable(false).autoDelete(true).exclusive(true)))
             .then(creator.bind(binding("globalReply", routeKey, queueName)))
             .thenMany(receiver.consumeAutoAck(queueName).doOnNext(delivery -> {
@@ -43,6 +47,13 @@ public void startListening(String routeKey) {
                 } catch (Exception e) {
                     log.log(Level.SEVERE, "Error in reply reception", e);
                 }
-            })).subscribe();
+            }));
+        onTerminate();
+    }
+
+
+    private void onTerminate() {
+        deliveryFlux.doOnTerminate(this::onTerminate)
+            .subscribe(new LoggerSubscriber<>(getClass().getName()));
     }
 }
diff --git a/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/listeners/GenericMessageListener.java b/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/listeners/GenericMessageListener.java
@@ -1,16 +1,20 @@
 package org.reactivecommons.async.rabbit.listeners;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Delivery;
 import lombok.extern.java.Log;
 import org.reactivecommons.async.commons.DiscardNotifier;
 import org.reactivecommons.async.commons.FallbackStrategy;
+import org.reactivecommons.async.commons.utils.LoggerSubscriber;
 import org.reactivecommons.async.rabbit.RabbitMessage;
 import org.reactivecommons.async.commons.communications.Message;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
 import org.reactivecommons.async.rabbit.communications.TopologyCreator;
 import org.reactivecommons.async.commons.ext.CustomReporter;
+import reactor.core.publisher.BaseSubscriber;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.SignalType;
 import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 import reactor.rabbitmq.AcknowledgableDelivery;
@@ -46,6 +50,7 @@ public abstract class GenericMessageListener {
     private final DiscardNotifier discardNotifier;
     private final String objectType;
     private final CustomReporter customReporter;
+    private volatile Flux<AcknowledgableDelivery> messageFlux;
 
     public GenericMessageListener(String queueName, ReactiveMessageListener listener, boolean useDLQRetries,
                                   long maxRetries, DiscardNotifier discardNotifier, String objectType, CustomReporter customReporter) {
@@ -74,10 +79,16 @@ public void startListener() {
         ConsumeOptions consumeOptions = new ConsumeOptions();
         consumeOptions.qos(messageListener.getPrefetchCount());
 
-        setUpBindings(messageListener.getTopologyCreator()).thenMany(
-                receiver.consumeManualAck(queueName, consumeOptions)
-                        .transform(this::consumeFaultTolerant))
-                .subscribe();
+        this.messageFlux = setUpBindings(messageListener.getTopologyCreator()).thenMany(
+            receiver.consumeManualAck(queueName, consumeOptions)
+                .transform(this::consumeFaultTolerant));
+        onTerminate();
+
+    }
+
+    private void onTerminate() {
+        messageFlux.doOnTerminate(this::onTerminate)
+            .subscribe(new LoggerSubscriber<>(getClass().getName()));
     }
 
 
diff --git a/async/async-rabbit/src/test/java/org/reactivecommons/async/rabbit/listeners/ApplicationCommandListenerPerfTest.java b/async/async-rabbit/src/test/java/org/reactivecommons/async/rabbit/listeners/ApplicationCommandListenerPerfTest.java
@@ -32,12 +32,14 @@
 import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
+import static reactor.core.publisher.Flux.defer;
 import static reactor.core.publisher.Flux.range;
 import static reactor.core.publisher.Mono.just;
 
@@ -95,7 +97,14 @@ public void shouldProcessMessagesInOptimalTime() throws JsonProcessingException,
         );
         messageListener = new StubGenericMessageListener("test-queue", reactiveMessageListener, true, 10, discardNotifier, "command", handlerResolver, messageConverter, errorReporter);
         Flux<AcknowledgableDelivery> messageFlux = createSource(messageCount);
-        when(receiver.consumeManualAck(Mockito.anyString(), any(ConsumeOptions.class))).thenReturn(messageFlux);
+        final AtomicBoolean flag = new AtomicBoolean(true);
+        when(receiver.consumeManualAck(Mockito.anyString(), any(ConsumeOptions.class))).thenAnswer(invocation -> defer(() -> {
+            if (flag.getAndSet(false)) {
+                return messageFlux;
+            } else {
+                return Flux.never();
+            }
+        }));
 
         messageListener.startListener();
         final long init = System.currentTimeMillis();
@@ -124,7 +133,14 @@ public void shouldProcessAsyncMessagesConcurrent() throws JsonProcessingExceptio
         );
         messageListener = new StubGenericMessageListener("test-queue", reactiveMessageListener, true, 10, discardNotifier, "command", handlerResolver, messageConverter, errorReporter);
         Flux<AcknowledgableDelivery> messageFlux = createSource(messageCount);
-        when(receiver.consumeManualAck(Mockito.anyString(), any(ConsumeOptions.class))).thenReturn(messageFlux);
+        final AtomicBoolean flag = new AtomicBoolean(true);
+        when(receiver.consumeManualAck(Mockito.anyString(), any(ConsumeOptions.class))).thenAnswer(invocation -> defer(() -> {
+            if (flag.getAndSet(false)) {
+                return messageFlux;
+            } else {
+                return Flux.never();
+            }
+        }));
         System.out.println("Permits before: " + semaphore.availablePermits());
         final long init = System.currentTimeMillis();
         messageListener.startListener();
@@ -189,7 +205,14 @@ public void shouldProcessCPUMessagesInParallel() throws JsonProcessingException,
         reactiveMessageListener = new ReactiveMessageListener(receiver, topologyCreator, 250, 250);
         messageListener = new StubGenericMessageListener("test-queue", reactiveMessageListener, true, 10, discardNotifier, "command", handlerResolver, messageConverter, errorReporter);
         Flux<AcknowledgableDelivery> messageFlux = createSource(messageCount);
-        when(receiver.consumeManualAck(Mockito.anyString(), any(ConsumeOptions.class))).thenReturn(messageFlux);
+        final AtomicBoolean flag = new AtomicBoolean(true);
+        when(receiver.consumeManualAck(Mockito.anyString(), any(ConsumeOptions.class))).thenAnswer(invocation -> defer(() -> {
+            if (flag.getAndSet(false)) {
+                return messageFlux;
+            } else {
+                return Flux.never();
+            }
+        }));
         System.out.println("Permits before: " + semaphore.availablePermits());
         final long init = System.currentTimeMillis();
         messageListener.startListener();
@@ -215,7 +238,14 @@ public void shouldProcessCPUWorkMessagesInParallel() throws JsonProcessingExcept
         reactiveMessageListener = new ReactiveMessageListener(receiver, topologyCreator, 500, 250);
         messageListener = new StubGenericMessageListener("test-queue", reactiveMessageListener, true, 10, discardNotifier, "command", handlerResolver, messageConverter, errorReporter);
         Flux<AcknowledgableDelivery> messageFlux = createSource(messageCount);
-        when(receiver.consumeManualAck(Mockito.anyString(), any(ConsumeOptions.class))).thenReturn(messageFlux);
+        final AtomicBoolean flag = new AtomicBoolean(true);
+        when(receiver.consumeManualAck(Mockito.anyString(), any(ConsumeOptions.class))).thenAnswer(invocation -> defer(() -> {
+            if (flag.getAndSet(false)) {
+                return messageFlux;
+            } else {
+                return Flux.never();
+            }
+        }));
         System.out.println("Permits before: " + semaphore.availablePermits());
         final long init = System.currentTimeMillis();
         messageListener.startListener();
@@ -241,7 +271,14 @@ public void shouldProcessPasiveBlockingMessagesInParallel() throws JsonProcessin
         reactiveMessageListener = new ReactiveMessageListener(receiver, topologyCreator, 500, 250);
         messageListener = new StubGenericMessageListener("test-queue", reactiveMessageListener, true, 10, discardNotifier, "command", handlerResolver, messageConverter, errorReporter);
         Flux<AcknowledgableDelivery> messageFlux = createSource(messageCount);
-        when(receiver.consumeManualAck(Mockito.anyString(), any(ConsumeOptions.class))).thenReturn(messageFlux);
+        final AtomicBoolean flag = new AtomicBoolean(true);
+        when(receiver.consumeManualAck(Mockito.anyString(), any(ConsumeOptions.class))).thenAnswer(invocation -> defer(() -> {
+            if (flag.getAndSet(false)) {
+                return messageFlux;
+            } else {
+                return Flux.never();
+            }
+        }));
         System.out.println("Permits before: " + semaphore.availablePermits());
         final long init = System.currentTimeMillis();
         messageListener.startListener();

Original file line number	Diff line number	Diff line change
`@@ -50,7 +50,7 @@ public void init() throws NoSuchFieldException, IllegalAccessException {`
`50`	`50`	`when(creator.declare(any(ExchangeSpecification.class))).thenReturn(Mono.just(mock(AMQP.Exchange.DeclareOk.class)));`
`51`	`51`	`when(creator.declareQueue(any(String.class), any())).thenReturn(Mono.just(mock(AMQP.Queue.DeclareOk.class)));`
`52`	`52`	`when(listener.getTopologyCreator()).thenReturn(creator);`
`53`		`- when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.empty());`
	`53`	`+ when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.never());`
`54`	`54`	`when(listener.getReceiver()).thenReturn(receiver);`
`55`	`55`	`when(listener.getMaxConcurrency()).thenReturn(20);`
`56`	`56`	`}`
Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,7 @@ public void init() {`
`44`	`44`	`when(creator.declareQueue(any(String.class), any())).thenReturn(Mono.just(mock(AMQP.Queue.DeclareOk.class)));`
`45`	`45`	`when(creator.declareQueue(any(String.class), any(String.class), any())).thenReturn(Mono.just(mock(AMQP.Queue.DeclareOk.class)));`
`46`	`46`	`when(listener.getTopologyCreator()).thenReturn(creator);`
`47`		`- when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.empty());`
	`47`	`+ when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.never());`
`48`	`48`	`when(listener.getReceiver()).thenReturn(receiver);`
`49`	`49`	`when(listener.getMaxConcurrency()).thenReturn(20);`
`50`	`50`	`}`
Original file line number	Diff line number	Diff line change
`@@ -46,7 +46,7 @@ public void init() {`
`46`	`46`	`when(creator.declareQueue(any(String.class), any())).thenReturn(Mono.just(mock(AMQP.Queue.DeclareOk.class)));`
`47`	`47`	`when(creator.declareQueue(any(String.class), any(String.class), any())).thenReturn(Mono.just(mock(AMQP.Queue.DeclareOk.class)));`
`48`	`48`	`when(listener.getTopologyCreator()).thenReturn(creator);`
`49`		`- when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.empty());`
	`49`	`+ when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.never());`
`50`	`50`	`when(listener.getReceiver()).thenReturn(receiver);`
`51`	`51`	`when(listener.getMaxConcurrency()).thenReturn(20);`
`52`	`52`	`}`