Merge pull request #65 from reactive-commons/fix/dynamic-event-binding

fix event bindings for dynamic events

Author: Daniel Bustamante Ospina

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

@@ -20,6 +20,7 @@
 public class HandlerRegistry {
 
     private final List<RegisteredEventListener<?>> eventListeners = new CopyOnWriteArrayList<>();
+    private final List<RegisteredEventListener<?>> dynamicEventHandlers = new CopyOnWriteArrayList<>();
     private final List<RegisteredEventListener<?>> eventNotificationListener = new CopyOnWriteArrayList<>();
     private final List<RegisteredQueryHandler<?, ?>> handlers = new CopyOnWriteArrayList<>();
     private final List<RegisteredCommandHandler<?>> commandHandlers = new CopyOnWriteArrayList<>();
@@ -42,27 +43,11 @@ public <T> HandlerRegistry listenNotificationEvent(String eventName, EventHandle
         return this;
     }
 
-    /**
-     * @param eventNamePattern
-     * @param handler
-     * @param eventClass
-     * @param <T>
-     * @return HandlerRegistry
-     * Avoid use this deprecated method please use listenEvent
-     */
-    @Deprecated
     public <T> HandlerRegistry handleDynamicEvents(String eventNamePattern, EventHandler<T> handler, Class<T> eventClass) {
-        return listenEvent(eventNamePattern, handler, eventClass);
+        dynamicEventHandlers.add(new RegisteredEventListener<>(eventNamePattern, handler, eventClass));
+        return this;
     }
 
-    /**
-     * @param eventNamePattern
-     * @param handler
-     * @param <T>
-     * @return HandlerRegistry
-     * Avoid use this deprecated method please use listenEvent
-     */
-    @Deprecated
     public <T> HandlerRegistry handleDynamicEvents(String eventNamePattern, EventHandler<T> handler) {
         return handleDynamicEvents(eventNamePattern, handler, inferGenericParameterType(handler));
     }

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

@@ -7,7 +7,11 @@
 import org.reactivecommons.api.domain.Command;
 import org.reactivecommons.api.domain.DomainEvent;
 import org.reactivecommons.api.domain.DomainEventBus;
-import org.reactivecommons.async.api.*;
+import org.reactivecommons.async.api.AsyncQuery;
+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.registered.RegisteredCommandHandler;
 import org.reactivecommons.async.api.handlers.registered.RegisteredEventListener;
 import org.reactivecommons.async.api.handlers.registered.RegisteredQueryHandler;
@@ -38,7 +42,15 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import reactor.core.publisher.Mono;
-import reactor.rabbitmq.*;
+import reactor.rabbitmq.ChannelPool;
+import reactor.rabbitmq.ChannelPoolFactory;
+import reactor.rabbitmq.ChannelPoolOptions;
+import reactor.rabbitmq.RabbitFlux;
+import reactor.rabbitmq.Receiver;
+import reactor.rabbitmq.ReceiverOptions;
+import reactor.rabbitmq.Sender;
+import reactor.rabbitmq.SenderOptions;
+import reactor.rabbitmq.Utils;
 import reactor.util.retry.Retry;
 
 import java.security.KeyManagementException;
@@ -48,6 +60,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.logging.Level;
+import java.util.stream.Stream;
 
 import static reactor.rabbitmq.ExchangeSpecification.exchange;
 
@@ -198,12 +211,19 @@ public HandlerResolver resolver(ApplicationContext context, DefaultCommandHandle
                 .collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
                         ConcurrentHashMap::putAll);
 
-        final ConcurrentMap<String, RegisteredEventListener<?>> eventListeners = registries
+        final ConcurrentMap<String, RegisteredEventListener<?>> eventsToBind = registries
                 .values().stream()
                 .flatMap(r -> r.getEventListeners().stream())
                 .collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
                         ConcurrentHashMap::putAll);
 
+        // event handlers and dynamic handlers
+        final ConcurrentMap<String, RegisteredEventListener<?>> eventHandlers = registries
+                .values().stream()
+                .flatMap(r -> Stream.concat(r.getEventListeners().stream(), r.getDynamicEventHandlers().stream()))
+                .collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
+                        ConcurrentHashMap::putAll);
+
         final ConcurrentMap<String, RegisteredCommandHandler<?>> commandHandlers = registries
                 .values().stream()
                 .flatMap(r -> r.getCommandHandlers().stream())
@@ -217,7 +237,7 @@ public HandlerResolver resolver(ApplicationContext context, DefaultCommandHandle
                 .collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
                         ConcurrentHashMap::putAll);
 
-        return new HandlerResolver(queryHandlers, eventListeners, eventNotificationListener, commandHandlers) {
+        return new HandlerResolver(queryHandlers, eventHandlers, eventsToBind, eventNotificationListener, commandHandlers) {
             @Override
             @SuppressWarnings("unchecked")
             public <T> RegisteredCommandHandler<T> getCommandHandler(String path) {

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

@@ -18,6 +18,7 @@ public class HandlerResolver {
 
     private final Map<String, RegisteredQueryHandler<?, ?>> queryHandlers;
     private final Map<String, RegisteredEventListener<?>> eventListeners;
+    private final Map<String, RegisteredEventListener<?>> eventsToBind;
     private final Map<String, RegisteredEventListener<?>> eventNotificationListeners;
     private final Map<String, RegisteredCommandHandler<?>> commandHandlers;
     private final Matcher matcher = new KeyMatcher();
@@ -53,8 +54,9 @@ public <T> RegisteredEventListener<T> getNotificationListener(String path) {
                 .computeIfAbsent(path, getMatchHandler(eventNotificationListeners));
     }
 
+    // Returns only the listenEvent not the handleDynamicEvents
     public Collection<RegisteredEventListener<?>> getEventListeners() {
-        return eventListeners.values();
+        return eventsToBind.values();
     }
 
     void addEventListener(RegisteredEventListener<?> listener) {

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

@@ -9,8 +9,6 @@
 import org.reactivecommons.async.commons.communications.Message;
 import org.reactivecommons.async.commons.converters.MessageConverter;
 import org.reactivecommons.async.commons.ext.CustomReporter;
-import org.reactivecommons.async.commons.utils.matcher.KeyMatcher;
-import org.reactivecommons.async.commons.utils.matcher.Matcher;
 import org.reactivecommons.async.rabbit.HandlerResolver;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
 import org.reactivecommons.async.rabbit.communications.TopologyCreator;
@@ -35,7 +33,6 @@ public class ApplicationEventListener extends GenericMessageListener {
     private final boolean withDLQRetry;
     private final int retryDelay;
     private final Optional<Integer> maxLengthBytes;
-    private final Matcher keyMatcher;
     private final String appName;
 
 
@@ -57,7 +54,6 @@ public ApplicationEventListener(ReactiveMessageListener receiver,
         this.eventsExchange = eventsExchange;
         this.messageConverter = messageConverter;
         this.maxLengthBytes = maxLengthBytes;
-        this.keyMatcher = new KeyMatcher();
         this.appName = appName;
     }
 

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

@@ -45,9 +45,10 @@ class DynamicRegistryImpTest {
     void setUp() {
         Map<String, RegisteredCommandHandler<?>> commandHandlers = new ConcurrentHashMap<>();
         Map<String, RegisteredEventListener<?>> eventListeners = new ConcurrentHashMap<>();
+        Map<String, RegisteredEventListener<?>> eventsToBind = new ConcurrentHashMap<>();
         Map<String, RegisteredEventListener<?>> notificationEventListeners = new ConcurrentHashMap<>();
         Map<String, RegisteredQueryHandler<?, ?>> queryHandlers = new ConcurrentHashMap<>();
-        resolver = new HandlerResolver(queryHandlers, eventListeners, notificationEventListeners, commandHandlers);
+        resolver = new HandlerResolver(queryHandlers, eventListeners, eventsToBind, notificationEventListeners, commandHandlers);
         dynamicRegistry = new DynamicRegistryImp(resolver, topologyCreator, props);
     }
 

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

@@ -0,0 +1,57 @@
+package org.reactivecommons.async.rabbit;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+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 reactor.core.publisher.Mono;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+class HandlerResolverTest {
+    private HandlerResolver resolver;
+
+    @BeforeEach
+    void setup() {
+        Map<String, RegisteredCommandHandler<?>> commandHandlers = new ConcurrentHashMap<>();
+        Map<String, RegisteredEventListener<?>> eventListeners = new ConcurrentHashMap<>();
+        eventListeners.put("event.name", new RegisteredEventListener<>("event.name", message -> Mono.empty(), String.class));
+        eventListeners.put("event.name2", new RegisteredEventListener<>("event.name2", message -> Mono.empty(), String.class));
+        eventListeners.put("some.*", new RegisteredEventListener<>("some.*", message -> Mono.empty(), String.class));
+        Map<String, RegisteredEventListener<?>> eventsToBind = new ConcurrentHashMap<>();
+        eventsToBind.put("event.name", new RegisteredEventListener<>("event.name", message -> Mono.empty(), String.class));
+        eventsToBind.put("event.name2", new RegisteredEventListener<>("event.name2", message -> Mono.empty(), String.class));
+        Map<String, RegisteredEventListener<?>> notificationEventListeners = new ConcurrentHashMap<>();
+        Map<String, RegisteredQueryHandler<?, ?>> queryHandlers = new ConcurrentHashMap<>();
+        resolver = new HandlerResolver(queryHandlers, eventListeners, eventsToBind, notificationEventListeners, commandHandlers);
+    }
+
+    @Test
+    void shouldGetOnlyTheBindingEvents() {
+        // Act
+        Collection<RegisteredEventListener<?>> eventListener = resolver.getEventListeners();
+        // Assert
+        Assertions.assertThat(eventListener.size()).isEqualTo(2);
+    }
+
+    @Test
+    void shouldMatchForAWildcardEvent() {
+        // Act
+        RegisteredEventListener<Object> eventListener = resolver.getEventListener("some.sample");
+        // Assert
+        Assertions.assertThat(eventListener.getPath()).isEqualTo("some.*");
+    }
+
+    @Test
+    void shouldMatchForAnExactEvent() {
+        // Act
+        RegisteredEventListener<Object> eventListener = resolver.getEventListener("event.name");
+        // Assert
+        Assertions.assertThat(eventListener.getPath()).isEqualTo("event.name");
+    }
+
+}

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

@@ -271,7 +271,7 @@ private HandlerResolver createHandlerResolver(final HandlerRegistry initialRegis
         final HandlerRegistry registry = range(0, 20).reduce(initialRegistry, (r, i) -> r.handleCommand("app.command.name" + i, message -> Mono.empty(), Map.class)).block();
         final ConcurrentMap<String, RegisteredCommandHandler<?>> commandHandlers = registry.getCommandHandlers().stream()
                 .collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler), ConcurrentHashMap::putAll);
-        return new HandlerResolver(null, null, null, commandHandlers) {
+        return new HandlerResolver(null, null, null, null, commandHandlers) {
             @Override
             @SuppressWarnings("unchecked")
             public RegisteredCommandHandler<Object> getCommandHandler(String path) {

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

@@ -80,7 +80,7 @@ public void setUp() {
         QueryHandler<String, SampleClass> handler = (message) -> just("OK");
         handlers.put("queryDirect", new RegisteredQueryHandler<>("queryDirect",
                 (from, message) -> handler.handle(message), SampleClass.class));
-        HandlerResolver resolver = new HandlerResolver(handlers, null, null, null);
+        HandlerResolver resolver = new HandlerResolver(handlers, null,null, null, null);
         applicationQueryListener = new ApplicationQueryListener(reactiveMessageListener, "queue", resolver, sender,
                 "directExchange", messageConverter, "replyExchange", false,
                 1, 100, maxLengthBytes, true, discardNotifier, errorReporter);

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

@@ -24,9 +24,18 @@
 import org.reactivecommons.async.rabbit.converters.json.JacksonMessageConverter;
 import org.reactivecommons.async.utils.TestUtils;
 import reactor.core.publisher.Flux;
-import reactor.rabbitmq.*;
-
-import java.util.*;
+import reactor.rabbitmq.AcknowledgableDelivery;
+import reactor.rabbitmq.BindingSpecification;
+import reactor.rabbitmq.ConsumeOptions;
+import reactor.rabbitmq.ExchangeSpecification;
+import reactor.rabbitmq.Receiver;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Random;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
@@ -38,7 +47,9 @@
 import static java.util.stream.Collectors.toMap;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 import static reactor.core.publisher.Mono.empty;
 import static reactor.core.publisher.Mono.just;
 
@@ -120,13 +131,15 @@ protected <T> Flux<AcknowledgableDelivery> createSource(Function<T, String> rout
     protected abstract GenericMessageListener createMessageListener(final HandlerResolver handlerResolver);
 
     private HandlerResolver createHandlerResolver(final HandlerRegistry registry) {
-        final Map<String, RegisteredEventListener<?>> eventHandlers = registry.getEventListeners().stream().collect(toMap(RegisteredEventListener::getPath, identity()));
+        final Map<String, RegisteredEventListener<?>> eventHandlers = Stream.concat(registry.getDynamicEventHandlers().stream(), registry.getEventListeners().stream()).collect(toMap(RegisteredEventListener::getPath, identity()));
+        final Map<String, RegisteredEventListener<?>> eventsToBind = registry.getEventListeners().stream().collect(toMap(RegisteredEventListener::getPath, identity()));
         final Map<String, RegisteredEventListener<?>> notificationHandlers = registry.getEventNotificationListener().stream().collect(toMap(RegisteredEventListener::getPath, identity()));
         final Map<String, RegisteredQueryHandler<?, ?>> queryHandlers = registry.getHandlers().stream().collect(toMap(RegisteredQueryHandler::getPath, identity()));
         final Map<String, RegisteredCommandHandler<?>> commandHandlers = registry.getCommandHandlers().stream().collect(toMap(RegisteredCommandHandler::getPath, identity()));
         return new HandlerResolver(
                 new ConcurrentHashMap<>(queryHandlers),
                 new ConcurrentHashMap<>(eventHandlers),
+                new ConcurrentHashMap<>(eventsToBind),
                 new ConcurrentHashMap<>(notificationHandlers),
                 new ConcurrentHashMap<>(commandHandlers));
     }

gradle.properties

@@ -1,4 +1,4 @@
-version=1.0.3
+version=1.0.4
 springBootVersion=2.4.2
 gradleVersionsVersion=0.36.0
 toPublish=async-commons,async-commons-api,async-commons-rabbit-standalone,async-commons-rabbit-starter,domain-events-api,async-rabbit