Fix test in CI

Author: Daniel Bustamante Ospina

async/async-rabbit/src/test/java/org/reactivecommons/async/rabbit/listeners/ApplicationCommandListenerPerfTest.java

@@ -24,6 +24,7 @@
 import org.reactivecommons.async.commons.converters.json.DefaultObjectMapperSupplier;
 import org.reactivecommons.async.rabbit.converters.json.JacksonMessageConverter;
 import org.reactivecommons.async.commons.ext.CustomReporter;
+import org.reactivecommons.async.utils.TestUtils;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.rabbitmq.*;
@@ -97,14 +98,7 @@ public void shouldProcessMessagesInOptimalTime() throws JsonProcessingException,
         );
         messageListener = new StubGenericMessageListener("test-queue", reactiveMessageListener, true, 10, discardNotifier, "command", handlerResolver, messageConverter, errorReporter);
         Flux<AcknowledgableDelivery> messageFlux = createSource(messageCount);
-        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();
-            }
-        }));
+        TestUtils.instructSafeReceiverMock(receiver, messageFlux);
 
         messageListener.startListener();
         final long init = System.currentTimeMillis();
@@ -133,14 +127,7 @@ public void shouldProcessAsyncMessagesConcurrent() throws JsonProcessingExceptio
         );
         messageListener = new StubGenericMessageListener("test-queue", reactiveMessageListener, true, 10, discardNotifier, "command", handlerResolver, messageConverter, errorReporter);
         Flux<AcknowledgableDelivery> messageFlux = createSource(messageCount);
-        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();
-            }
-        }));
+        TestUtils.instructSafeReceiverMock(receiver, messageFlux);
         System.out.println("Permits before: " + semaphore.availablePermits());
         final long init = System.currentTimeMillis();
         messageListener.startListener();
@@ -205,14 +192,9 @@ 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);
-        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();
-            }
-        }));
+
+        TestUtils.instructSafeReceiverMock(receiver, messageFlux);
+
         System.out.println("Permits before: " + semaphore.availablePermits());
         final long init = System.currentTimeMillis();
         messageListener.startListener();
@@ -238,14 +220,9 @@ 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);
-        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();
-            }
-        }));
+
+        TestUtils.instructSafeReceiverMock(receiver, messageFlux);
+
         System.out.println("Permits before: " + semaphore.availablePermits());
         final long init = System.currentTimeMillis();
         messageListener.startListener();
@@ -271,14 +248,9 @@ 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);
-        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();
-            }
-        }));
+
+        TestUtils.instructSafeReceiverMock(receiver, messageFlux);
+
         System.out.println("Permits before: " + semaphore.availablePermits());
         final long init = System.currentTimeMillis();
         messageListener.startListener();

async/async-rabbit/src/test/java/org/reactivecommons/async/rabbit/listeners/ListenerReporterTestSuperClass.java

@@ -23,13 +23,15 @@
 import org.reactivecommons.async.commons.converters.json.DefaultObjectMapperSupplier;
 import org.reactivecommons.async.rabbit.converters.json.JacksonMessageConverter;
 import org.reactivecommons.async.commons.ext.CustomReporter;
+import org.reactivecommons.async.utils.TestUtils;
 import reactor.core.publisher.Flux;
 import reactor.rabbitmq.*;
 
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -39,6 +41,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.*;
+import static reactor.core.publisher.Flux.defer;
 import static reactor.core.publisher.Mono.empty;
 import static reactor.core.publisher.Mono.just;
 
@@ -90,8 +93,7 @@ protected void assertSendErrorToCustomReporter(HandlerRegistry handlerRegistry,
 
         messageListener = createMessageListener(handlerResolver);
 
-        Flux<AcknowledgableDelivery> messageFlux = source;
-        when(receiver.consumeManualAck(Mockito.anyString(), any(ConsumeOptions.class))).thenReturn(messageFlux);
+        TestUtils.instructSafeReceiverMock(receiver, source);
 
         messageListener.startListener();
 

async/async-rabbit/src/test/java/org/reactivecommons/async/utils/TestUtils.java

@@ -0,0 +1,31 @@
+package org.reactivecommons.async.utils;
+
+import org.mockito.Mockito;
+import reactor.core.publisher.Flux;
+import reactor.rabbitmq.AcknowledgableDelivery;
+import reactor.rabbitmq.ConsumeOptions;
+import reactor.rabbitmq.Receiver;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+import static reactor.core.publisher.Flux.defer;
+
+public class TestUtils {
+
+    private TestUtils() {
+    }
+
+    public static void instructSafeReceiverMock(final Receiver receiver, final Flux<AcknowledgableDelivery> source) {
+        final AtomicBoolean flag = new AtomicBoolean(true);
+        when(receiver.consumeManualAck(Mockito.anyString(), any(ConsumeOptions.class))).thenAnswer(invocation -> defer(() -> {
+            if (flag.getAndSet(false)) {
+                return source;
+            } else {
+                return Flux.never();
+            }
+        }));
+    }
+
+}