Implement recovery of lost consumer for tmp queues

Author: Daniel Bustamante Ospina

async/async-commons/src/main/java/org/reactivecommons/async/commons/ext/CustomErrorReporter.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();
+    }
+}

async/async-commons/src/main/java/org/reactivecommons/async/commons/utils/ArrayUtils.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));
+    }
+}

async/async-commons/src/main/java/org/reactivecommons/async/commons/utils/LoggerSubscriber.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();
+    }
+
+}

async/async-commons/src/test/java/org/reactivecommons/async/commons/utils/LoggerSubscriberTest.java

@@ -1,22 +1,18 @@
 package org.reactivecommons.async.rabbit.listeners;
 
 import com.rabbitmq.client.Delivery;
-import com.rabbitmq.client.UnblockedCallback;
 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 org.reactivestreams.Subscription;
-import reactor.core.publisher.BaseSubscriber;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.SignalType;
 import reactor.rabbitmq.Receiver;
 
 import java.util.logging.Level;
 
 import static org.reactivecommons.async.commons.Headers.*;
-import static reactor.core.publisher.Signal.subscribe;
 import static reactor.rabbitmq.ResourcesSpecification.*;
 
 @Log
@@ -52,37 +48,12 @@ public void startListening(String routeKey) {
                     log.log(Level.SEVERE, "Error in reply reception", e);
                 }
             }));
-        //deliveryFlux.subscribe();
         onTerminate();
     }
 
 
     private void onTerminate() {
-        deliveryFlux.doOnTerminate(this::onTerminate).subscribe(new BaseSubscriber<Delivery>() {
-            @Override
-            protected void hookOnNext(Delivery value) {
-                log.info("#On NextHook");
-            }
-
-            @Override
-            protected void hookOnComplete() {
-                log.warning("#On Complete Hook!!");
-            }
-
-            @Override
-            protected void hookOnError(Throwable throwable) {
-                log.log(Level.SEVERE, "#Hook On Error!!", throwable);
-            }
-
-            @Override
-            protected void hookOnCancel() {
-                log.warning("#On Cancel Hook!!");
-            }
-
-            @Override
-            protected void hookFinally(SignalType type) {
-                log.warning("#On Finally Hook!! " + type.name());
-            }
-        });
+        deliveryFlux.doOnTerminate(this::onTerminate)
+            .subscribe(new LoggerSubscriber<>(getClass().getName()));
     }
 }

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

@@ -5,6 +5,7 @@
 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;
@@ -86,29 +87,8 @@ public void startListener() {
     }
 
     private void onTerminate() {
-        log.info("Hard Subscription 2 to " + this.getClass().getName());
-        messageFlux.doOnTerminate(this::onTerminate).subscribe(new BaseSubscriber<Delivery>() {
-
-            @Override
-            protected void hookOnComplete() {
-                log.warning("##On Complete Hook!!");
-            }
-
-            @Override
-            protected void hookOnError(Throwable throwable) {
-                log.log(Level.SEVERE, "##Hook On Error!!", throwable);
-            }
-
-            @Override
-            protected void hookOnCancel() {
-                log.warning("##On Cancel Hook!!");
-            }
-
-            @Override
-            protected void hookFinally(SignalType type) {
-                log.warning("##On Finally Hook!! " + type.name());
-            }
-        });
+        messageFlux.doOnTerminate(this::onTerminate)
+            .subscribe(new LoggerSubscriber<>(getClass().getName()));
     }