enable dynamic registry serveQuery handler

Author: juancgalvis

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

@@ -1,13 +1,16 @@
 package org.reactivecommons.async.api;
 
 import org.reactivecommons.async.api.handlers.EventHandler;
+import org.reactivecommons.async.api.handlers.QueryHandler;
 import reactor.core.publisher.Mono;
 
 public interface DynamicRegistry {
 
     @Deprecated
     <T> Mono<Void> listenEvent(String eventName, EventHandler<T> fn, Class<T> eventClass);
 
+    <T, R> void serveQuery(String resource, QueryHandler<T, R> handler, Class<R> queryClass);
+
     Mono<Void> startListeningEvent(String eventName);
 
     Mono<Void> stopListeningEvent(String eventName);

async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/DynamicRegistryImp.java

@@ -4,9 +4,11 @@
 import lombok.RequiredArgsConstructor;
 import org.reactivecommons.async.api.DynamicRegistry;
 import org.reactivecommons.async.api.handlers.EventHandler;
+import org.reactivecommons.async.api.handlers.QueryHandler;
 import org.reactivecommons.async.api.handlers.registered.RegisteredEventListener;
-import org.reactivecommons.async.rabbit.communications.TopologyCreator;
+import org.reactivecommons.async.api.handlers.registered.RegisteredQueryHandler;
 import org.reactivecommons.async.commons.config.IBrokerConfigProps;
+import org.reactivecommons.async.rabbit.communications.TopologyCreator;
 import reactor.core.publisher.Mono;
 import reactor.rabbitmq.BindingSpecification;
 
@@ -26,6 +28,11 @@ public <T> Mono<Void> listenEvent(String eventName, EventHandler<T> fn, Class<T>
                 .then();
     }
 
+    @Override
+    public <T, R> void serveQuery(String resource, QueryHandler<T, R> handler, Class<R> queryClass) {
+        resolver.addQueryHandler(new RegisteredQueryHandler<>(resource, (ignored, message) -> handler.handle(message), queryClass));
+    }
+
     @Override
     public Mono<Void> startListeningEvent(String eventName) {
         return topologyCreator.bind(buildBindingSpecification(eventName))

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

@@ -1,6 +1,7 @@
 package org.reactivecommons.async.rabbit;
 
 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;
@@ -13,6 +14,7 @@
 import java.util.Set;
 import java.util.function.Function;
 
+@Log
 @RequiredArgsConstructor
 public class HandlerResolver {
 
@@ -73,6 +75,13 @@ void addEventListener(RegisteredEventListener<?> listener) {
         eventListeners.put(listener.getPath(), listener);
     }
 
+    void addQueryHandler(RegisteredQueryHandler<?, ?> handler) {
+        if (handler.getPath().contains("*")) {
+            log.warning("avoid * in dynamic handlers, make sure you have no conflicts with cached patterns");
+        }
+        queryHandlers.put(handler.getPath(), handler);
+    }
+
     private <T> Function<String, T> getMatchHandler(Map<String, T> handlers) {
         return name -> {
             String matched = matcher.match(handlers.keySet(), name);

async/async-rabbit/src/test/java/org/reactivecommons/async/rabbit/DynamicRegistryImpTest.java

@@ -11,10 +11,8 @@
 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.rabbit.DynamicRegistryImp;
-import org.reactivecommons.async.rabbit.HandlerResolver;
-import org.reactivecommons.async.rabbit.communications.TopologyCreator;
 import org.reactivecommons.async.commons.config.IBrokerConfigProps;
+import org.reactivecommons.async.rabbit.communications.TopologyCreator;
 import reactor.core.publisher.Mono;
 import reactor.rabbitmq.BindingSpecification;
 import reactor.test.StepVerifier;
@@ -52,13 +50,12 @@ void setUp() {
         Map<String, RegisteredQueryHandler<?, ?>> queryHandlers = new ConcurrentHashMap<>();
         resolver = new HandlerResolver(queryHandlers, eventListeners, notificationEventListeners,
                 dynamicEventsHandlers, commandHandlers);
-        when(props.getDomainEventsExchangeName()).thenReturn("domainEx");
-        when(props.getEventsQueue()).thenReturn("events.queue");
         dynamicRegistry = new DynamicRegistryImp(resolver, topologyCreator, props);
     }
 
     @Test
     void registerEventListener() {
+        setupMock();
         when(topologyCreator.bind(any())).thenReturn(just(mock(BindOk.class)));
         dynamicRegistry.listenEvent("event1", message -> Mono.empty(), Long.class);
 
@@ -68,6 +65,7 @@ void registerEventListener() {
 
     @Test
     void declareBindingWhenRegisterEventListener() {
+        setupMock();
         ArgumentCaptor<BindingSpecification> captor = ArgumentCaptor.forClass(BindingSpecification.class);
         when(topologyCreator.bind(any())).thenReturn(just(mock(BindOk.class)));
 
@@ -82,6 +80,7 @@ void declareBindingWhenRegisterEventListener() {
 
     @Test
     void subscribeToResultWhenRegisterEventListener() {
+        setupMock();
         PublisherProbe<BindOk> probe = PublisherProbe.of(just(mock(BindOk.class)));
         when(topologyCreator.bind(any())).thenReturn(probe.mono());
 
@@ -94,6 +93,7 @@ void subscribeToResultWhenRegisterEventListener() {
 
     @Test
     void shouldBindDomainEventsToEventsQueueUsingEventName() {
+        setupMock();
         ArgumentCaptor<BindingSpecification> bindingSpecificationCaptor =
                 ArgumentCaptor.forClass(BindingSpecification.class);
 
@@ -119,6 +119,7 @@ void shouldBindDomainEventsToEventsQueueUsingEventName() {
 
     @Test
     void shouldUnbindDomainEventsToEventsQueueUsingEventName() {
+        setupMock();
         ArgumentCaptor<BindingSpecification> bindingSpecificationCaptor =
                 ArgumentCaptor.forClass(BindingSpecification.class);
 
@@ -142,5 +143,19 @@ void shouldUnbindDomainEventsToEventsQueueUsingEventName() {
         topologyCreatorProbe.assertWasSubscribed();
     }
 
+    @Test
+    void serveQueryShouldAddHandler() {
+        dynamicRegistry.serveQuery("dynamic.query", message -> Mono.just(message * 100), Integer.class);
+
+        final RegisteredQueryHandler<Integer, Integer> handler = resolver.getQueryHandler("dynamic.query");
+        StepVerifier.create(handler.getHandler().handle(null, 50))
+                .expectNext(5000)
+                .verifyComplete();
+    }
+
+    private void setupMock(){
+        when(props.getDomainEventsExchangeName()).thenReturn("domainEx");
+        when(props.getEventsQueue()).thenReturn("events.queue");
+    }
 
 }