feat(handler): add support for queue listeners and enhance HandlerRegistry

Author: luisgomez29

acceptance/async-tests/src/test/java/org/reactivecommons/test/QueryProcessPerfTest.java

@@ -32,8 +32,8 @@ class QueryProcessPerfTest {
     private static final String QUERY_NAME = "app.command.test";
     private static final int MESSAGE_COUNT = 40000;
     private static final Semaphore semaphore = new Semaphore(0);
-    private static final AtomicLong atomicLong = new AtomicLong(0);
-    private static final CountDownLatch latch = new CountDownLatch(12 + 1);
+    // private static final AtomicLong atomicLong = new AtomicLong(0);
+    // private static final CountDownLatch latch = new CountDownLatch(12 + 1);
 
     @Autowired
     private DirectAsyncGateway gateway;
@@ -75,7 +75,7 @@ private void assertMessageThroughput(long total, long messageCount, int reqMicro
     private Flux<AsyncQuery<DummyMessage>> createMessages(int count) {
         final List<AsyncQuery<DummyMessage>> queryList = IntStream.range(0, count)
                 .mapToObj(_v -> new AsyncQuery<>(QUERY_NAME, new DummyMessage()))
-                .collect(Collectors.toList());
+                .toList();
         return Flux.fromIterable(queryList);
     }
 
@@ -95,6 +95,7 @@ public HandlerRegistry registry() {
                             "app.command.name" + i, message -> Mono.empty(), Map.class
                     ))
                     .block();
+            assert registry != null;
             return registry
                     .serveQuery(QUERY_NAME, this::handleSimple, DummyMessage.class);
         }

async/async-commons-api/src/main/java/org/reactivecommons/async/api/HandlerRegistry.java

@@ -13,10 +13,13 @@
 import org.reactivecommons.async.api.handlers.QueryHandlerDelegate;
 import org.reactivecommons.async.api.handlers.RawCommandHandler;
 import org.reactivecommons.async.api.handlers.RawEventHandler;
+import org.reactivecommons.async.api.handlers.TopologyHandlerSetup;
 import org.reactivecommons.async.api.handlers.registered.RegisteredCommandHandler;
 import org.reactivecommons.async.api.handlers.registered.RegisteredDomainHandlers;
 import org.reactivecommons.async.api.handlers.registered.RegisteredEventListener;
 import org.reactivecommons.async.api.handlers.registered.RegisteredQueryHandler;
+import org.reactivecommons.async.api.handlers.registered.RegisteredQueueListener;
+import reactor.core.publisher.Mono;
 
 import java.lang.reflect.ParameterizedType;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -35,6 +38,8 @@ public final class HandlerRegistry {
     private final RegisteredDomainHandlers<RegisteredQueryHandler<?, ?>> handlers = new RegisteredDomainHandlers<>();
     private final RegisteredDomainHandlers<RegisteredCommandHandler<?, ?>> commandHandlers =
             new RegisteredDomainHandlers<>();
+    private final RegisteredDomainHandlers<RegisteredQueueListener> queueHandlers =
+            new RegisteredDomainHandlers<>();
 
 
     public static HandlerRegistry register() {
@@ -176,6 +181,26 @@ public HandlerRegistry serveCloudEventQuery(String resource, QueryHandlerDelegat
         return this;
     }
 
+    // Queues
+    public HandlerRegistry listenQueue(String queueName, RawEventHandler<RawMessage> handler) {
+        return listenQueue(queueName, handler, creator -> Mono.empty());
+    }
+
+    public HandlerRegistry listenQueue(String domain, String queueName, RawEventHandler<RawMessage> handler) {
+        return listenQueue(domain, queueName, handler, creator -> Mono.empty());
+    }
+
+    public HandlerRegistry listenQueue(String queueName, RawEventHandler<RawMessage> handler,
+                                       TopologyHandlerSetup topologyCreator) {
+        queueHandlers.add(DEFAULT_DOMAIN, new RegisteredQueueListener(queueName, handler, topologyCreator));
+        return this;
+    }
+
+    public HandlerRegistry listenQueue(String domain, String queueName, RawEventHandler<RawMessage> handler,
+                                       TopologyHandlerSetup topologyCreator) {
+        queueHandlers.add(domain, new RegisteredQueueListener(queueName, handler, topologyCreator));
+        return this;
+    }
 
     @Deprecated(forRemoval = true)
     public <T> HandlerRegistry listenEvent(String eventName, DomainEventHandler<T> handler) {

async/async-commons-api/src/main/java/org/reactivecommons/async/api/handlers/TopologyHandlerSetup.java

@@ -0,0 +1,11 @@
+package org.reactivecommons.async.api.handlers;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * Interface for setting up topology handlers.
+ */
+@FunctionalInterface
+public interface TopologyHandlerSetup {
+    Mono<Void> setup(Object topologyCreator);
+}

async/async-commons-api/src/main/java/org/reactivecommons/async/api/handlers/registered/RegisteredQueueListener.java

@@ -0,0 +1,11 @@
+package org.reactivecommons.async.api.handlers.registered;
+
+import org.reactivecommons.api.domain.RawMessage;
+import org.reactivecommons.async.api.handlers.RawEventHandler;
+import org.reactivecommons.async.api.handlers.TopologyHandlerSetup;
+
+
+public record RegisteredQueueListener(String queueName,
+                                      RawEventHandler<RawMessage> handler,
+                                      TopologyHandlerSetup topologyHandlerSetup) {
+}

async/async-commons/src/main/java/org/reactivecommons/async/commons/HandlerResolver.java

@@ -1,10 +1,12 @@
 package org.reactivecommons.async.commons;
 
+import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.java.Log;
 import org.reactivecommons.async.api.handlers.registered.RegisteredCommandHandler;
 import org.reactivecommons.async.api.handlers.registered.RegisteredEventListener;
 import org.reactivecommons.async.api.handlers.registered.RegisteredQueryHandler;
+import org.reactivecommons.async.api.handlers.registered.RegisteredQueueListener;
 import org.reactivecommons.async.commons.utils.matcher.KeyMatcher;
 import org.reactivecommons.async.commons.utils.matcher.Matcher;
 
@@ -23,6 +25,8 @@ public class HandlerResolver {
     private final Map<String, RegisteredEventListener<?, ?>> eventsToBind;
     private final Map<String, RegisteredEventListener<?, ?>> eventNotificationListeners;
     private final Map<String, RegisteredCommandHandler<?, ?>> commandHandlers;
+    @Getter
+    private final Map<String, RegisteredQueueListener> queueListeners;
 
     private final Matcher matcher = new KeyMatcher();
 

async/async-commons/src/main/java/org/reactivecommons/async/commons/utils/resolver/HandlerResolverBuilder.java

@@ -8,6 +8,7 @@
 import org.reactivecommons.async.api.handlers.registered.RegisteredCommandHandler;
 import org.reactivecommons.async.api.handlers.registered.RegisteredEventListener;
 import org.reactivecommons.async.api.handlers.registered.RegisteredQueryHandler;
+import org.reactivecommons.async.api.handlers.registered.RegisteredQueueListener;
 import org.reactivecommons.async.commons.HandlerResolver;
 
 import java.util.List;
@@ -42,7 +43,17 @@ public static HandlerResolver buildResolver(String domain, Map<String, HandlerRe
                         -> map.put(handler.path(), handler), ConcurrentHashMap::putAll
                 );
 
-        final ConcurrentMap<String, RegisteredEventListener<?, ?>> eventNotificationListener = registries
+        final ConcurrentMap<String, RegisteredQueueListener> queueListeners = registries
+                .values()
+                .stream()
+                .flatMap(r -> r.getQueueHandlers()
+                        .getOrDefault(domain, List.of())
+                        .stream())
+                .collect(ConcurrentHashMap::new, (map, handler)
+                        -> map.put(handler.queueName(), handler), ConcurrentHashMap::putAll
+                );
+
+        final ConcurrentMap<String, RegisteredEventListener<?, ?>> eventNotificationListeners = registries
                 .values()
                 .stream()
                 .flatMap(r -> r.getEventNotificationListener()
@@ -55,11 +66,11 @@ public static HandlerResolver buildResolver(String domain, Map<String, HandlerRe
         final ConcurrentMap<String, RegisteredEventListener<?, ?>> eventsToBind = getEventsToBind(domain,
                 registries);
 
-        final ConcurrentMap<String, RegisteredEventListener<?, ?>> eventHandlers =
+        final ConcurrentMap<String, RegisteredEventListener<?, ?>> eventListeners =
                 getEventHandlersWithDynamics(domain, registries);
 
-        return new HandlerResolver(queryHandlers, eventHandlers, eventsToBind, eventNotificationListener,
-                commandHandlers) {
+        return new HandlerResolver(queryHandlers, eventListeners, eventsToBind, eventNotificationListeners,
+                commandHandlers, queueListeners) {
             @Override
             @SuppressWarnings("unchecked")
             public <T, D> RegisteredCommandHandler<T, D> getCommandHandler(String path) {

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

@@ -64,6 +64,7 @@ public ApplicationCommandListener(ReactiveMessageListener listener,
         this.maxLengthBytes = maxLengthBytes;
     }
 
+    @Override
     protected Mono<Void> setUpBindings(TopologyCreator creator) {
         final Mono<AMQP.Exchange.DeclareOk> declareExchange = creator.declare(
                 ExchangeSpecification.exchange(directExchange).durable(true).type("direct")

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

@@ -61,6 +61,7 @@ public ApplicationEventListener(ReactiveMessageListener receiver,
         this.appName = appName;
     }
 
+    @Override
     protected Mono<Void> setUpBindings(TopologyCreator creator) {
         final Mono<AMQP.Exchange.DeclareOk> declareExchange = creator.declare(
                 ExchangeSpecification.exchange(eventsExchange).durable(true).type("topic")

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

@@ -50,6 +50,7 @@ public ApplicationNotificationListener(ReactiveMessageListener receiver,
         this.createTopology = createTopology;
     }
 
+    @Override
     protected Mono<Void> setUpBindings(TopologyCreator creator) {
 
         final Mono<AMQP.Queue.DeclareOk> declareQueue = creator.declare(

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

@@ -0,0 +1,58 @@
+package org.reactivecommons.async.rabbit.listeners;
+
+import lombok.extern.java.Log;
+import org.reactivecommons.async.api.handlers.registered.RegisteredQueueListener;
+import org.reactivecommons.async.commons.DiscardNotifier;
+import org.reactivecommons.async.commons.communications.Message;
+import org.reactivecommons.async.commons.ext.CustomReporter;
+import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
+import org.reactivecommons.async.rabbit.communications.TopologyCreator;
+import reactor.core.publisher.Mono;
+import reactor.rabbitmq.AcknowledgableDelivery;
+
+import java.util.function.Function;
+
+@Log
+public class ApplicationQueueListener extends GenericMessageListener {
+
+    private final RegisteredQueueListener registeredListener;
+
+    public ApplicationQueueListener(ReactiveMessageListener listener,
+                                    boolean withDLQRetry,
+                                    long maxRetries,
+                                    int retryDelay,
+                                    RegisteredQueueListener registeredListener,
+                                    DiscardNotifier discardNotifier,
+                                    CustomReporter errorReporter) {
+        super(registeredListener.queueName(), listener, withDLQRetry, true, maxRetries, retryDelay,
+                discardNotifier, "queue", errorReporter);
+        this.registeredListener = registeredListener;
+    }
+
+    @Override
+    protected Mono<Void> setUpBindings(TopologyCreator creator) {
+        return registeredListener.topologyHandlerSetup().setup(creator);
+    }
+
+    @Override
+    protected Function<Message, Mono<Object>> rawMessageHandler(String executorPath) {
+        return message -> registeredListener.handler().handle(message).cast(Object.class);
+    }
+
+    protected String getExecutorPath(AcknowledgableDelivery msj) {
+        return registeredListener.queueName();
+    }
+
+    @Override
+    protected Object parseMessageForReporter(Message message) {
+        return message;
+    }
+
+    @Override
+    protected String getKind() {
+        return "queue";
+    }
+}
+
+
+