Add DynamicRegistry in order to handle new events in runtime and DynamicRegistryTest to demonstrate his usage

Fixes: gh-9

Author: Daniel Bustamante Ospina

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

@@ -0,0 +1,71 @@
+package org.reactivecommons.test;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.reactivecommons.api.domain.DomainEvent;
+import org.reactivecommons.api.domain.DomainEventBus;
+import org.reactivecommons.async.api.AsyncQuery;
+import org.reactivecommons.async.api.DirectAsyncGateway;
+import org.reactivecommons.async.api.DynamicRegistry;
+import org.reactivecommons.async.api.HandlerRegistry;
+import org.reactivecommons.async.api.handlers.EventHandler;
+import org.reactivecommons.async.impl.config.annotations.EnableDirectAsyncGateway;
+import org.reactivecommons.async.impl.config.annotations.EnableDomainEventBus;
+import org.reactivecommons.async.impl.config.annotations.EnableMessageListeners;
+import org.reactivestreams.Publisher;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.test.context.junit4.SpringRunner;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.UnicastProcessor;
+import reactor.test.StepVerifier;
+
+import java.time.Duration;
+
+import static reactor.core.publisher.Mono.*;
+
+@SpringBootTest
+@RunWith(SpringRunner.class)
+public class DynamicRegistryTest {
+
+    @Autowired
+    private DomainEventBus eventBus;
+
+    @Autowired
+    private DynamicRegistry dynamicRegistry;
+
+    @Value("${spring.application.name}")
+    private String appName;
+
+    @Test
+    public void shouldReceiveResponse() {
+        UnicastProcessor<String> result = UnicastProcessor.create();
+        EventHandler<String> fn = message -> fromRunnable(() -> result.onNext(message.getData()));
+
+        dynamicRegistry.listenEvent("test.event", fn, String.class).block();
+        final Publisher<Void> emit = eventBus.emit(new DomainEvent<>("test.event", "42", "Hello"));
+        from(emit).block();
+
+        StepVerifier.create(result.next().timeout(Duration.ofSeconds(10)))
+            .expectNext("Hello")
+            .verifyComplete();
+
+
+    }
+
+
+    @SpringBootApplication
+    @EnableMessageListeners
+    @EnableDomainEventBus
+    static class App{
+        public static void main(String[] args) {
+            SpringApplication.run(App.class, args);
+        }
+
+    }
+}

acceptance/async-tests/src/test/resources/application.properties

@@ -1 +1 @@
-spring.application.name=testApp
+spring.application.name=test-app

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

@@ -0,0 +1,8 @@
+package org.reactivecommons.async.api;
+
+import org.reactivecommons.async.api.handlers.EventHandler;
+import reactor.core.publisher.Mono;
+
+public interface DynamicRegistry {
+    <T> Mono<Void> listenEvent(String eventName, EventHandler<T> fn, Class<T> eventClass);
+}

async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/DirectAsyncGatewayConfig.java

@@ -1,42 +1,32 @@
 package org.reactivecommons.async.impl.config;
 
-import lombok.AllArgsConstructor;
-import lombok.NoArgsConstructor;
+import lombok.RequiredArgsConstructor;
 import org.reactivecommons.async.impl.RabbitDirectAsyncGateway;
 import org.reactivecommons.async.impl.communications.ReactiveMessageListener;
 import org.reactivecommons.async.impl.communications.ReactiveMessageSender;
+import org.reactivecommons.async.impl.config.props.BrokerConfigProps;
 import org.reactivecommons.async.impl.converters.MessageConverter;
 import org.reactivecommons.async.impl.listeners.ApplicationReplyListener;
 import org.reactivecommons.async.impl.reply.ReactiveReplyRouter;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
-import org.springframework.util.Base64Utils;
-
-import java.nio.ByteBuffer;
-import java.util.UUID;
 
 @Configuration
 @Import(RabbitMqConfig.class)
-@AllArgsConstructor
-@NoArgsConstructor
+@RequiredArgsConstructor
 public class DirectAsyncGatewayConfig {
 
-    @Value("${app.async.direct.exchange:directMessages}")
-    private String directMessagesExchangeName;
-
-    @Value("${spring.application.name}")
-    private String appName;
+    private final BrokerConfigProps props;
 
     @Bean
     public RabbitDirectAsyncGateway rabbitDirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router, ReactiveMessageSender rSender, MessageConverter converter) throws Exception {
-        return new RabbitDirectAsyncGateway(config, router, rSender, directMessagesExchangeName, converter);
+        return new RabbitDirectAsyncGateway(config, router, rSender, props.getDirectMessagesExchangeName(), converter);
     }
 
     @Bean
     public ApplicationReplyListener msgListener(ReactiveReplyRouter router, BrokerConfig config, ReactiveMessageListener listener)  {
-        final ApplicationReplyListener replyListener = new ApplicationReplyListener(router, listener, generateName());
+        final ApplicationReplyListener replyListener = new ApplicationReplyListener(router, listener, props.getReplyQueue());
         replyListener.startListening(config.getRoutingKey());
         return replyListener;
     }
@@ -53,14 +43,5 @@ public ReactiveReplyRouter router() {
         return new ReactiveReplyRouter();
     }
 
-    public String generateName() {
-        UUID uuid = UUID.randomUUID();
-        ByteBuffer bb = ByteBuffer.wrap(new byte[16]);
-        bb.putLong(uuid.getMostSignificantBits())
-            .putLong(uuid.getLeastSignificantBits());
-        // Convert to base64 and remove trailing =
-        return this.appName + Base64Utils.encodeToUrlSafeString(bb.array())
-            .replaceAll("=", "");
-    }
 
 }

async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/EventBusConfig.java

@@ -3,7 +3,7 @@
 import org.reactivecommons.api.domain.DomainEventBus;
 import org.reactivecommons.async.impl.RabbitDomainEventBus;
 import org.reactivecommons.async.impl.communications.ReactiveMessageSender;
-import org.springframework.beans.factory.annotation.Value;
+import org.reactivecommons.async.impl.config.props.BrokerConfigProps;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
@@ -14,12 +14,10 @@
 @Import(RabbitMqConfig.class)
 public class EventBusConfig {
 
-    @Value("${app.async.domain.events.exchange:domainEvents}")
-    private String domainEventsExchangeName;
-
     @Bean
-    public DomainEventBus domainEventBus(ReactiveMessageSender sender) {
-        sender.getTopologyCreator().declare(exchange(domainEventsExchangeName).durable(true).type("topic")).subscribe();
-        return new RabbitDomainEventBus(sender, domainEventsExchangeName);
+    public DomainEventBus domainEventBus(ReactiveMessageSender sender, BrokerConfigProps props) {
+        final String exchangeName = props.getDomainEventsExchangeName();
+        sender.getTopologyCreator().declare(exchange(exchangeName).durable(true).type("topic")).subscribe();
+        return new RabbitDomainEventBus(sender, exchangeName);
     }
 }

async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/MessageConfig.java

@@ -2,14 +2,18 @@
 
 import org.reactivecommons.async.api.DefaultCommandHandler;
 import org.reactivecommons.async.api.DefaultQueryHandler;
+import org.reactivecommons.async.api.DynamicRegistry;
 import org.reactivecommons.async.api.HandlerRegistry;
 import org.reactivecommons.async.api.handlers.QueryHandler;
 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.impl.DynamicRegistryImp;
 import org.reactivecommons.async.impl.HandlerResolver;
 import org.reactivecommons.async.impl.communications.ReactiveMessageListener;
 import org.reactivecommons.async.impl.communications.ReactiveMessageSender;
+import org.reactivecommons.async.impl.communications.TopologyCreator;
+import org.reactivecommons.async.impl.config.props.BrokerConfigProps;
 import org.reactivecommons.async.impl.converters.MessageConverter;
 import org.reactivecommons.async.impl.listeners.ApplicationCommandListener;
 import org.reactivecommons.async.impl.listeners.ApplicationEventListener;
@@ -25,6 +29,7 @@
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 
 @Configuration
 @Import(RabbitMqConfig.class)
@@ -60,23 +65,28 @@ public ApplicationCommandListener applicationCommandListener(ReactiveMessageList
         return commandListener;
     }
 
+    @Bean
+    public DynamicRegistry dynamicRegistry(HandlerResolver resolver, ReactiveMessageListener listener, IBrokerConfigProps props) {
+        return new DynamicRegistryImp(resolver, listener.getTopologyCreator(), props);
+    }
+
     @Bean
     public HandlerResolver resolver(ApplicationContext context, DefaultQueryHandler defaultHandler, Environment env, DefaultCommandHandler defaultCommandHandler) {
         final Map<String, HandlerRegistry> registries = context.getBeansOfType(HandlerRegistry.class);
 
-        final ConcurrentHashMap<String, RegisteredQueryHandler> handlers = registries
+        final ConcurrentMap<String, RegisteredQueryHandler> handlers = registries
             .values().stream()
             .flatMap(r -> r.getHandlers().stream())
             .collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
                 ConcurrentHashMap::putAll);
 
-        final Map<String, RegisteredEventListener> eventListeners = registries
+        final ConcurrentMap<String, RegisteredEventListener> eventListeners = registries
             .values().stream()
             .flatMap(r -> r.getEventListeners().stream())
             .collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
                 ConcurrentHashMap::putAll);
 
-        final Map<String, RegisteredCommandHandler> commandHandlers = registries
+        final ConcurrentMap<String, RegisteredCommandHandler> commandHandlers = registries
             .values().stream()
             .flatMap(r -> r.getCommandHandlers().stream())
             .collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),

async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/MessageListenersConfig.java

@@ -2,20 +2,19 @@
 
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
-import lombok.AllArgsConstructor;
-import lombok.NoArgsConstructor;
 import lombok.extern.java.Log;
 import org.reactivecommons.async.impl.communications.ReactiveMessageListener;
 import org.reactivecommons.async.impl.communications.ReactiveMessageSender;
 import org.reactivecommons.async.impl.communications.TopologyCreator;
+import org.reactivecommons.async.impl.config.props.BrokerConfigProps;
 import org.reactivecommons.async.impl.converters.JacksonMessageConverter;
 import org.reactivecommons.async.impl.converters.MessageConverter;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.boot.context.properties.PropertyMapper;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
@@ -26,19 +25,15 @@
 
 @Log
 @Configuration
-@NoArgsConstructor
-@AllArgsConstructor
 @EnableConfigurationProperties(RabbitProperties.class)
+@Import(BrokerConfigProps.class)
 public class RabbitMqConfig {
 
-    @Value("${spring.application.name}")
-    private String appName;
-
     @Bean
-    public ReactiveMessageSender messageSender(ConnectionFactoryProvider provider, MessageConverter converter){
+    public ReactiveMessageSender messageSender(ConnectionFactoryProvider provider, MessageConverter converter, BrokerConfigProps props){
         final Mono<Connection> senderConnection = createSenderConnectionMono(provider.getConnectionFactory(), "sender");
         final Sender sender = RabbitFlux.createSender(new SenderOptions().connectionMono(senderConnection));
-        return new ReactiveMessageSender(sender, appName, converter, new TopologyCreator(senderConnection));
+        return new ReactiveMessageSender(sender, props.getAppName(), converter, new TopologyCreator(senderConnection));
     }
 
     @Bean

async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/RabbitMqConfig.java

@@ -0,0 +1,67 @@
+package org.reactivecommons.async.impl.config.props;
+
+import lombok.Getter;
+import org.reactivecommons.async.impl.config.IBrokerConfigProps;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.util.Base64Utils;
+
+import java.nio.ByteBuffer;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
+
+
+@Getter
+@Configuration
+public class BrokerConfigProps implements IBrokerConfigProps {
+
+    @Value("${spring.application.name}")
+    private String appName;
+
+    @Value("${app.async.domain.events.exchange:domainEvents}")
+    private String domainEventsExchangeName;
+
+    @Value("${app.async.direct.exchange:directMessages}")
+    private String directMessagesExchangeName;
+
+    private final AtomicReference<String> replyQueueName = new AtomicReference<>();
+
+    @Override
+    public String getEventsQueue() {
+        return appName + ".subsEvents";
+    }
+
+    @Override
+    public String getQueriesQueue() {
+        return appName + ".query";
+    }
+
+    @Override
+    public String getCommandsQueue() {
+        return appName;
+    }
+
+    @Override
+    public String getReplyQueue() {
+        final String name = replyQueueName.get();
+        if (name == null) {
+            final String replyName =  newRandomQueueName();
+            if (replyQueueName.compareAndSet(null, replyName)) {
+                return replyName;
+            } else {
+                return replyQueueName.get();
+            }
+        }
+        return name;
+    }
+
+    private String newRandomQueueName() {
+        UUID uuid = UUID.randomUUID();
+        ByteBuffer bb = ByteBuffer.wrap(new byte[16]);
+        bb.putLong(uuid.getMostSignificantBits())
+            .putLong(uuid.getLeastSignificantBits());
+        return appName + Base64Utils.encodeToUrlSafeString(bb.array())
+            .replaceAll("=", "");
+    }
+
+}