listener-matcher c69f7ef

author gf7 <godfat7@gmail.com> 1591241642 -0500
committer gf7 <godfat7@gmail.com> 1596560939 -0500

Adds event notification listener and functionality to find most accurate listener for an specific event

Adds notification event listener

Author: kwah07

README.md

@@ -11,23 +11,23 @@ To include all (API and implementation) (Spring boot Starter):
 ```groovy
 
     dependencies {
-      compile 'org.reactivecommons:async-commons-starter:0.4.6'
+      compile 'org.reactivecommons:async-commons-starter:0.5.0'
     }
 ```
 
 To include only domain events API:
 
 ```groovy
     dependencies {
-      compile 'org.reactivecommons:domain-events-api:0.4.6'
+      compile 'org.reactivecommons:domain-events-api:0.5.0'
     }
 ```
 
 To include only async commons API:
 
 ```groovy
     dependencies {
-      compile 'org.reactivecommons:async-commons-api:0.4.6'
+      compile 'org.reactivecommons:async-commons-api:0.5.0'
     }
 ```
 
@@ -128,6 +128,31 @@ Don't forget to add the implementation dependency to the main spring boot module
     }
 ```
 
+### Domain Event-Listener
+Reactive commons has four types of listeners implemented, available to be registered in the application via the **HandlerRegistry**, each of them is designed to tackle   
+common requirements for listeners in event based applications and abstracts the behavior of event flow in every situation (Varying for example in retrying strategy, dead letter events, sources and so on).
+
+The available event listeners are:
+- Domain Event Listener
+- Query Event Listener
+- Command Listener
+- Notification Listener
+
+Example Code:
+```java
+        public HandlerRegistry notificationEvents() {
+            return HandlerRegistry.register()
+                    .listenNotificationEvent("gatewayRouteAdded", message -> {
+                        System.out.println("Refreshing instance");
+                        return Mono.empty();
+                    },GatewayEvent.class);
+        }
+```
+
+The above code shows how to handle a notification event (Notification event: an event that should be handled by
+every running instance of a microservice, e.g: notify to every instance that a configuration setting has changed
+  and has to do a hot reload from a persistent source of that data).
+
 ### Request-Reply
 Example Code:
 

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

@@ -25,6 +25,7 @@ public static HandlerRegistry register(){
     private final List<RegisteredQueryHandler<?, ?>> handlers = new CopyOnWriteArrayList<>();
     private final List<RegisteredEventListener> eventListeners = new CopyOnWriteArrayList<>();
     private final List<RegisteredCommandHandler> commandHandlers = new CopyOnWriteArrayList<>();
+    private final List<RegisteredEventListener> eventNotificationListener = new CopyOnWriteArrayList<>();
 
     public <T> HandlerRegistry listenEvent(String eventName, EventHandler<T> fn, Class<T> eventClass){
         eventListeners.add(new RegisteredEventListener<>(eventName, fn,  eventClass));
@@ -36,6 +37,11 @@ public <T> HandlerRegistry listenEvent(String eventName, EventHandler<T> handler
         return this;
     }
 
+    public <T> HandlerRegistry listenNotificationEvent(String eventName, EventHandler<T> fn, Class<T> eventClass){
+        eventNotificationListener.add(new RegisteredEventListener<>(eventName, fn,  eventClass));
+        return this;
+    }
+
     public <T> HandlerRegistry handleCommand(String commandName, CommandHandler<T> fn, Class<T> commandClass){
         commandHandlers.add(new RegisteredCommandHandler<>(commandName, fn,  commandClass));
         return this;

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

@@ -17,6 +17,7 @@
 import org.reactivecommons.async.impl.converters.MessageConverter;
 import org.reactivecommons.async.impl.listeners.ApplicationCommandListener;
 import org.reactivecommons.async.impl.listeners.ApplicationEventListener;
+import org.reactivecommons.async.impl.listeners.ApplicationNotificationListener;
 import org.reactivecommons.async.impl.listeners.ApplicationQueryListener;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@@ -48,9 +49,21 @@ public ApplicationEventListener eventListener(HandlerResolver resolver, MessageC
                 appName + ".subsEvents", resolver, asyncProps.getDomain().getEvents().getExchange(),
                 messageConverter, asyncProps.getWithDLQRetry(), asyncProps.getMaxRetries(), asyncProps.getRetryDelay(),
                 discardNotifier);
-
         listener.startListener();
+        return listener;
+    }
 
+    @Bean
+    public ApplicationNotificationListener eventNotificationListener(HandlerResolver resolver, MessageConverter messageConverter,
+                                                         ReactiveMessageListener receiver, DiscardNotifier discardNotifier) {
+        final ApplicationNotificationListener listener = new ApplicationNotificationListener(
+                receiver,
+                asyncProps.getDomain().getEvents().getExchange(),
+                asyncProps.getNotificationProps().getQueueName(appName),
+                resolver,
+                messageConverter,
+                discardNotifier);
+        listener.startListener();
         return listener;
     }
 
@@ -62,9 +75,7 @@ public ApplicationQueryListener queryListener(MessageConverter converter, Handle
                 appName + ".query", resolver, sender, asyncProps.getDirect().getExchange(), converter,
                 "globalReply", asyncProps.getWithDLQRetry(), asyncProps.getMaxRetries(),
                 asyncProps.getRetryDelay(), discardNotifier);
-
         listener.startListener();
-
         return listener;
     }
 
@@ -75,9 +86,7 @@ public ApplicationCommandListener applicationCommandListener(ReactiveMessageList
         ApplicationCommandListener commandListener = new ApplicationCommandListener(listener, appName, resolver,
                 asyncProps.getDirect().getExchange(), converter, asyncProps.getWithDLQRetry(), asyncProps.getMaxRetries(),
                 asyncProps.getRetryDelay(), discardNotifier);
-
         commandListener.startListener();
-
         return commandListener;
     }
 
@@ -108,7 +117,14 @@ public HandlerResolver resolver(ApplicationContext context, DefaultCommandHandle
                 .collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
                         ConcurrentHashMap::putAll);
 
-        return new HandlerResolver(handlers, eventListeners, commandHandlers) {
+        final ConcurrentMap<String, RegisteredEventListener> eventNotificationListener = registries
+                .values()
+                .stream()
+                .flatMap(r -> r.getEventNotificationListener().stream())
+                .collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
+                        ConcurrentHashMap::putAll);
+
+        return new HandlerResolver(handlers, eventListeners, commandHandlers, eventNotificationListener) {
             @Override
             @SuppressWarnings("unchecked")
             public <T> RegisteredCommandHandler<T> getCommandHandler(String path) {
@@ -125,7 +141,6 @@ public DefaultQueryHandler defaultHandler() {
                 Mono.error(new RuntimeException("No Handler Registered"));
     }
 
-
     @Bean
     @ConditionalOnMissingBean
     public DefaultCommandHandler defaultCommandHandler() {

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

@@ -20,12 +20,14 @@ public class AsyncProps {
     @NestedConfigurationProperty
     private DirectProps direct = new DirectProps();
 
+    @NestedConfigurationProperty
+    private NotificationProps notificationProps = new NotificationProps();
+
     private Integer maxRetries = 10;
 
     private Integer prefetchCount = 250;
 
     private Integer retryDelay = 1000;
 
     private Boolean withDLQRetry = false;
-
 }

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

@@ -3,6 +3,7 @@
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import org.reactivecommons.async.impl.config.IBrokerConfigProps;
+import org.reactivecommons.async.impl.utils.NameGenerator;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.util.Base64Utils;
@@ -19,9 +20,7 @@ public class BrokerConfigProps implements IBrokerConfigProps {
 
     @Value("${spring.application.name}")
     private String appName;
-
     private final AsyncProps asyncProps;
-
     private final AtomicReference<String> replyQueueName = new AtomicReference<>();
 
     @Override
@@ -43,7 +42,7 @@ public String getCommandsQueue() {
     public String getReplyQueue() {
         final String name = replyQueueName.get();
         if (name == null) {
-            final String replyName = newRandomQueueName();
+            final String replyName = NameGenerator.generateNameFrom(appName);
             if (replyQueueName.compareAndSet(null, replyName)) {
                 return replyName;
             } else {
@@ -62,14 +61,4 @@ public String getDomainEventsExchangeName() {
     public String getDirectMessagesExchangeName() {
         return asyncProps.getDirect().getExchange();
     }
-
-    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("=", "");
-    }
-
 }

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

@@ -0,0 +1,28 @@
+package org.reactivecommons.async.impl.config.props;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import org.reactivecommons.async.impl.utils.NameGenerator;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+@Getter
+@RequiredArgsConstructor
+public class NotificationProps {
+
+    private final AtomicReference<String> queueName = new AtomicReference<>();
+    private final String queueSuffix = "notification";
+
+    public String getQueueName(String applicationName) {
+        final String name = this.queueName.get();
+        if(name == null) return getGeneratedName(applicationName);
+        return name;
+    }
+
+    private String getGeneratedName(String applicationName) {
+        String generatedName = NameGenerator.generateNameFrom(applicationName, queueSuffix);
+        return this.queueName
+                .compareAndSet(null, generatedName) ?
+                generatedName : this.queueName.get();
+    }
+}

async/async-commons/async-commons.gradle

@@ -79,4 +79,6 @@ dependencies {
     api "io.projectreactor.rabbitmq:reactor-rabbitmq"
     api 'com.fasterxml.jackson.core:jackson-databind'
     testImplementation 'io.projectreactor:reactor-test'
+    testCompile group: 'org.databene', name: 'contiperf', version: '2.3.4'
+
 }

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

@@ -14,7 +14,7 @@ public class HandlerResolver {
     private final Map<String, RegisteredQueryHandler> queryHandlers;
     private final Map<String, RegisteredEventListener> eventListeners;
     private final Map<String, RegisteredCommandHandler> commandHandlers;
-
+    private final Map<String , RegisteredEventListener> eventNotificationListeners;
 
     @SuppressWarnings("unchecked")
     public <T, R> RegisteredQueryHandler<T, R> getQueryHandler(String path) {
@@ -31,6 +31,14 @@ public <T> RegisteredEventListener<T> getEventListener(String path) {
         return eventListeners.get(path);
     }
 
+    public Collection<RegisteredEventListener> getNotificationListeners() {
+        return eventNotificationListeners.values();
+    }
+
+    public <T> RegisteredEventListener<T> getNotificationListener(String path) {
+        return eventNotificationListeners.get(path);
+    }
+
     public Collection<RegisteredEventListener> getEventListeners() {
         return eventListeners.values();
     }

async/async-commons/src/main/java/org/reactivecommons/async/impl/listeners/ApplicationEventListener.java

@@ -13,14 +13,18 @@
 import org.reactivecommons.async.impl.HandlerResolver;
 import org.reactivecommons.async.impl.communications.ReactiveMessageListener;
 import org.reactivecommons.async.impl.communications.TopologyCreator;
+import org.reactivecommons.async.impl.utils.matcher.KeyMatcher;
+import org.reactivecommons.async.impl.utils.matcher.Matcher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.rabbitmq.AcknowledgableDelivery;
 import reactor.rabbitmq.BindingSpecification;
 import reactor.rabbitmq.ExchangeSpecification;
 import reactor.rabbitmq.QueueSpecification;
 
+import java.util.Set;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 
 import static reactor.core.publisher.Flux.fromIterable;
 
@@ -32,14 +36,24 @@ public class ApplicationEventListener extends GenericMessageListener {
     private final String eventsExchange;
     private final boolean withDLQRetry;
     private final int retryDelay;
-
-    public ApplicationEventListener(ReactiveMessageListener receiver, String queueName, HandlerResolver resolver, String eventsExchange, MessageConverter messageConverter, boolean withDLQRetry, long maxRetries, int retryDelay, DiscardNotifier discardNotifier) {
+    private final Matcher keyMatcher;
+
+    public ApplicationEventListener(ReactiveMessageListener receiver,
+                                    String queueName,
+                                    HandlerResolver resolver,
+                                    String eventsExchange,
+                                    MessageConverter messageConverter,
+                                    boolean withDLQRetry,
+                                    long maxRetries,
+                                    int retryDelay,
+                                    DiscardNotifier discardNotifier) {
         super(queueName, receiver, withDLQRetry, maxRetries, discardNotifier, "event");
         this.retryDelay = retryDelay;
         this.withDLQRetry = withDLQRetry;
         this.resolver = resolver;
         this.eventsExchange = eventsExchange;
         this.messageConverter = messageConverter;
+        this.keyMatcher = new KeyMatcher();
     }
 
     protected Mono<Void> setUpBindings(TopologyCreator creator) {
@@ -48,7 +62,8 @@ protected Mono<Void> setUpBindings(TopologyCreator creator) {
             final Mono<AMQP.Exchange.DeclareOk> declareExchangeDLQ = creator.declare(ExchangeSpecification.exchange(eventsExchange+".DLQ").durable(true).type("topic"));
             final Mono<AMQP.Queue.DeclareOk> declareDLQ = creator.declareDLQ(queueName, eventsExchange, retryDelay);
             final Mono<AMQP.Queue.DeclareOk> declareQueue = creator.declareQueue(queueName, eventsExchange+".DLQ");
-            final Flux<AMQP.Queue.BindOk> bindings = fromIterable(resolver.getEventListeners()).flatMap(listener -> creator.bind(BindingSpecification.binding(eventsExchange, listener.getPath(), queueName)));
+            final Flux<AMQP.Queue.BindOk> bindings = fromIterable(resolver.getEventListeners())
+                    .flatMap(listener -> creator.bind(BindingSpecification.binding(eventsExchange, listener.getPath(), queueName)));
             final Flux<AMQP.Queue.BindOk> bindingDLQ = fromIterable(resolver.getEventListeners()).flatMap(listener -> creator.bind(BindingSpecification.binding(eventsExchange+".DLQ", listener.getPath(), queueName + ".DLQ")));
             return declareExchange.then(declareExchangeDLQ).then(declareQueue).then(declareDLQ).thenMany(bindings).thenMany(bindingDLQ).then();
         } else {
@@ -63,26 +78,24 @@ protected Mono<Void> setUpBindings(TopologyCreator creator) {
 
     @Override
     protected Function<Message, Mono<Object>> rawMessageHandler(String executorPath) {
-        final RegisteredEventListener<Object> handler = resolver.getEventListener(executorPath);
+        final Set<String> listenerKeys = resolver.getEventListeners()
+                .stream()
+                .map(RegisteredEventListener::getPath)
+                .collect(Collectors.toSet());
+        final String matchedKey = keyMatcher.match(listenerKeys, executorPath);
+        final RegisteredEventListener<Object> handler = resolver.getEventListener(matchedKey);
         final Class<Object> eventClass = handler.getInputClass();
         Function<Message, DomainEvent<Object>> converter = msj -> messageConverter.readDomainEvent(msj, eventClass);
         final EventExecutor<Object> executor = new EventExecutor<>(handler.getHandler(), converter);
-        return msj -> executor.execute(msj).cast(Object.class);
+        return msj -> executor
+                .execute(msj)
+                .cast(Object.class);
     }
 
     protected String getExecutorPath(AcknowledgableDelivery msj) {
         return msj.getEnvelope().getRoutingKey();
     }
 
-
-    @Data
-    private static class DomainEventInt {
-        private String name;
-        private String eventId;
-        private JsonNode data;
-    }
-
-
 }