fix(performance): Return to specific scheduler used in v1.3.0 for mes…

…sage selector (#61)
bancolombia · Jul 4, 2024 · 0b3d56d · 0b3d56d
1 parent ec9e124
commit 0b3d56d
Show file tree

Hide file tree

Showing 9 changed files with 19 additions and 9 deletions.
diff --git a/...ns-jms-mq/src/main/java/co/com/bancolombia/commons/jms/mq/config/MQAutoconfiguration.java b/...ns-jms-mq/src/main/java/co/com/bancolombia/commons/jms/mq/config/MQAutoconfiguration.java
@@ -11,7 +11,7 @@
 import co.com.bancolombia.commons.jms.internal.models.MQListenerConfig;
 import co.com.bancolombia.commons.jms.internal.models.RetryableConfig;
 import co.com.bancolombia.commons.jms.mq.config.health.MQListenerHealthIndicator;
-import co.com.bancolombia.commons.jms.mq.listeners.MQExecutorService;
+import co.com.bancolombia.commons.jms.internal.listener.selector.MQExecutorService;
 import co.com.bancolombia.commons.jms.mq.utils.MQUtils;
 import co.com.bancolombia.commons.jms.utils.MQQueueUtils;
 import co.com.bancolombia.commons.jms.utils.MQQueuesContainerImp;

diff --git a/commons-jms-mq/src/main/java/co/com/bancolombia/commons/jms/mq/config/MQSpringResolver.java b/commons-jms-mq/src/main/java/co/com/bancolombia/commons/jms/mq/config/MQSpringResolver.java
@@ -7,7 +7,7 @@
 import co.com.bancolombia.commons.jms.internal.listener.selector.strategy.SelectorBuilder;
 import co.com.bancolombia.commons.jms.internal.models.RetryableConfig;
 import co.com.bancolombia.commons.jms.mq.config.senders.MQSenderContainer;
-import co.com.bancolombia.commons.jms.mq.listeners.MQExecutorService;
+import co.com.bancolombia.commons.jms.internal.listener.selector.MQExecutorService;
 import co.com.bancolombia.commons.jms.utils.ReactiveReplyRouter;
 import jakarta.jms.ConnectionFactory;
 import jakarta.jms.Message;

diff --git a/...-mq/src/main/java/co/com/bancolombia/commons/jms/mq/config/factory/MQReqReplyFactory.java b/...-mq/src/main/java/co/com/bancolombia/commons/jms/mq/config/factory/MQReqReplyFactory.java
@@ -8,6 +8,7 @@
 import co.com.bancolombia.commons.jms.api.MQQueuesContainer;
 import co.com.bancolombia.commons.jms.api.exceptions.MQHealthListener;
 import co.com.bancolombia.commons.jms.api.model.JmsMessage;
+import co.com.bancolombia.commons.jms.internal.listener.selector.MQExecutorService;
 import co.com.bancolombia.commons.jms.internal.listener.selector.MQMultiContextMessageSelectorListener;
 import co.com.bancolombia.commons.jms.internal.listener.selector.MQMultiContextMessageSelectorListenerSync;
 import co.com.bancolombia.commons.jms.internal.listener.selector.strategy.ContextPerMessageStrategy;
@@ -163,8 +164,9 @@ private static MQRequestReplySelector fixedQueueWithMessageSelector(ReqReply ann
                 selectorModeProvider,
                 queuesContainer);
         SelectorBuilder selector = resolver.getSelectorBuilder();
+        MQExecutorService executorService = resolver.getMqExecutorService();
         MQMessageSelectorListener reactiveSelectorListener = new MQMultiContextMessageSelectorListener(
-                selectorListener);
+                selectorListener, executorService);
         return new MQRequestReplySelector(
                 sender,
                 queuesContainer,

diff --git a/...ms-mq/src/test/java/co/com/bancolombia/commons/jms/mq/config/MQAutoconfiguracionTest.java b/...ms-mq/src/test/java/co/com/bancolombia/commons/jms/mq/config/MQAutoconfiguracionTest.java
@@ -9,7 +9,7 @@
 import co.com.bancolombia.commons.jms.api.exceptions.MQHealthListener;
 import co.com.bancolombia.commons.jms.internal.listener.selector.strategy.SelectorBuilder;
 import co.com.bancolombia.commons.jms.internal.models.RetryableConfig;
-import co.com.bancolombia.commons.jms.mq.listeners.MQExecutorService;
+import co.com.bancolombia.commons.jms.internal.listener.selector.MQExecutorService;
 import co.com.bancolombia.commons.jms.utils.ReactiveReplyRouter;
 import com.ibm.mq.jakarta.jms.MQQueue;
 import jakarta.jms.JMSContext;

diff --git a/...s-jms-mq/src/test/java/co/com/bancolombia/commons/jms/mq/config/MQSpringResolverTest.java b/...s-jms-mq/src/test/java/co/com/bancolombia/commons/jms/mq/config/MQSpringResolverTest.java
@@ -8,7 +8,7 @@
 import co.com.bancolombia.commons.jms.internal.listener.selector.strategy.SelectorBuilder;
 import co.com.bancolombia.commons.jms.internal.models.RetryableConfig;
 import co.com.bancolombia.commons.jms.mq.config.senders.MQSenderContainer;
-import co.com.bancolombia.commons.jms.mq.listeners.MQExecutorService;
+import co.com.bancolombia.commons.jms.internal.listener.selector.MQExecutorService;
 import co.com.bancolombia.commons.jms.utils.MQQueuesContainerImp;
 import co.com.bancolombia.commons.jms.utils.ReactiveReplyRouter;
 import jakarta.jms.ConnectionFactory;

diff --git a/...ms-mq/src/test/java/co/com/bancolombia/commons/jms/mq/config/utils/MQFactoryBeanTest.java b/...ms-mq/src/test/java/co/com/bancolombia/commons/jms/mq/config/utils/MQFactoryBeanTest.java
@@ -13,7 +13,7 @@
 import co.com.bancolombia.commons.jms.mq.config.MQProperties;
 import co.com.bancolombia.commons.jms.mq.config.MQSpringResolver;
 import co.com.bancolombia.commons.jms.mq.config.senders.MQSenderContainer;
-import co.com.bancolombia.commons.jms.mq.listeners.MQExecutorService;
+import co.com.bancolombia.commons.jms.internal.listener.selector.MQExecutorService;
 import co.com.bancolombia.commons.jms.utils.MQQueuesContainerImp;
 import jakarta.jms.ConnectionFactory;
 import jakarta.jms.JMSConsumer;

diff --git a/...s/jms/mq/listeners/MQExecutorService.java → .../listener/selector/MQExecutorService.java b/...s/jms/mq/listeners/MQExecutorService.java → .../listener/selector/MQExecutorService.java
@@ -1,4 +1,4 @@
-package co.com.bancolombia.commons.jms.mq.listeners;
+package co.com.bancolombia.commons.jms.internal.listener.selector;
 
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;

diff --git a/...olombia/commons/jms/internal/listener/selector/MQMultiContextMessageSelectorListener.java b/...olombia/commons/jms/internal/listener/selector/MQMultiContextMessageSelectorListener.java
@@ -15,6 +15,7 @@
 @AllArgsConstructor
 public class MQMultiContextMessageSelectorListener implements MQMessageSelectorListener {
     private final MQMessageSelectorListenerSync listenerSync; // MQMultiContextMessageSelectorListenerSync
+    private final MQExecutorService executorService;
 
     @Override
     public Mono<Message> getMessage(String correlationId) {
@@ -48,6 +49,6 @@ public Mono<Message> getMessageBySelector(String selector, long timeout, Destina
 
     private Mono<Message> doAsync(Supplier<Message> supplier) {
         return Mono.fromSupplier(supplier)
-                .subscribeOn(Schedulers.boundedElastic());
+                .subscribeOn(Schedulers.fromExecutor(executorService));
     }
 }
diff --git a/...bia/commons/jms/internal/listener/selector/MQMultiContextMessageSelectorListenerTest.java b/...bia/commons/jms/internal/listener/selector/MQMultiContextMessageSelectorListenerTest.java
@@ -25,6 +25,8 @@
 import reactor.test.StepVerifier;
 
 import java.util.UUID;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.TimeUnit;
 
 import static co.com.bancolombia.commons.jms.internal.listener.selector.MQContextMessageSelectorListenerSync.DEFAULT_TIMEOUT;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -37,6 +39,8 @@
 
 @ExtendWith(MockitoExtension.class)
 class MQMultiContextMessageSelectorListenerTest {
+    public static final int MAX_THREADS = 200;
+    public static final long KEEP_ALIVE_SECONDS = 5L;
     @Mock
     private ConnectionFactory connectionFactory;
     @Mock
@@ -73,7 +77,10 @@ void setup() {
         MQMessageSelectorListenerSync listenerSync =
                 new MQMultiContextMessageSelectorListenerSync(config, healthListener,
                         retryableConfig, provider, container);
-        listener = new MQMultiContextMessageSelectorListener(listenerSync);
+        MQExecutorService executorService =
+                new MQExecutorService(0, MAX_THREADS, KEEP_ALIVE_SECONDS, TimeUnit.SECONDS,
+                        new SynchronousQueue<>());
+        listener = new MQMultiContextMessageSelectorListener(listenerSync, executorService);
     }
 
     @Test