Merge branch 'master' into separate-messagelistenersconfig

Author: Daniel Bustamante Ospina

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'
     }
 ```
 

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/listeners/ApplicationEventListener.java

@@ -13,6 +13,8 @@
 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;
@@ -21,7 +23,9 @@
 import reactor.rabbitmq.QueueSpecification;
 
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 
 import static reactor.core.publisher.Flux.fromIterable;
 
@@ -34,15 +38,28 @@ public class ApplicationEventListener extends GenericMessageListener {
     private final boolean withDLQRetry;
     private final int retryDelay;
     private final Optional<Integer> maxLengthBytes;
+    private final Matcher keyMatcher;
 
-    public ApplicationEventListener(ReactiveMessageListener receiver, String queueName, HandlerResolver resolver, String eventsExchange, MessageConverter messageConverter, boolean withDLQRetry, long maxRetries, int retryDelay, Optional<Integer> maxLengthBytes, DiscardNotifier discardNotifier) {
+
+    
+
+    public ApplicationEventListener(ReactiveMessageListener receiver, 
+                                    String queueName, 
+                                    HandlerResolver resolver, 
+                                    String eventsExchange, 
+                                    MessageConverter messageConverter, 
+                                    boolean withDLQRetry, 
+                                    long maxRetries, int retryDelay, 
+                                    Optional<Integer> maxLengthBytes, 
+                                    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.maxLengthBytes = maxLengthBytes;
+        this.keyMatcher = new KeyMatcher();
     }
 
     protected Mono<Void> setUpBindings(TopologyCreator creator) {
@@ -66,26 +83,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;
-    }
-
-
 }
 
 

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

@@ -112,6 +112,8 @@ private <T> Flux<T> outerFailureProtection(Flux<T> messageFlux) {
                 } catch (Exception e) {
                     log.log(Level.SEVERE, "Error returning message in failure!", e);
                 }
+            } else {
+                throw new RuntimeException(throwable);
             }
         });
     }

async/async-commons/src/main/java/org/reactivecommons/async/impl/utils/matcher/Candidate.java

@@ -0,0 +1,22 @@
+package org.reactivecommons.async.impl.utils.matcher;
+
+import lombok.Data;
+
+import java.util.Comparator;
+
+@Data
+public class Candidate implements Comparable<Candidate>, Comparator<Candidate> {
+    private final String key;
+    private final long score;
+
+    @Override
+    public int compareTo(Candidate o) {
+        return (int) (this.score - o.score);
+    }
+
+    @Override
+    public int compare(Candidate o1, Candidate o2) {
+        return (int) (o1.score - o2.score);
+    }
+}
+

async/async-commons/src/main/java/org/reactivecommons/async/impl/utils/matcher/KeyMatcher.java

@@ -0,0 +1,70 @@
+package org.reactivecommons.async.impl.utils.matcher;
+
+import lombok.Data;
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuples;
+
+import java.util.Set;
+import java.util.stream.IntStream;
+
+@Data
+public class KeyMatcher implements Matcher {
+    private static final String WILD_CARD = "*";
+    private static final String SEGMENT_DELIMITER_REGEX = "\\.";
+
+    private String[] getSegments(String s) {
+        return s.split(SEGMENT_DELIMITER_REGEX);
+    }
+
+    private boolean isBoundedToTarget(String[] current, String[] target) {
+        final String lastSegment = current[current.length -1];
+        return (current.length == target.length ||
+                (current.length < target.length && WILD_CARD.equals(lastSegment)));
+    }
+
+    private boolean isRootSegmentCoincident(String[] current, String[] target) {
+        final String currentFirstSegment = current[0];
+        final String targetFirstSegment = target[0];
+        return currentFirstSegment.equalsIgnoreCase(targetFirstSegment);
+    }
+
+    private boolean isCandidate(String[] current, String[] target ) {
+        return isRootSegmentCoincident(current, target) &&
+                isBoundedToTarget(current,target);
+    }
+
+    private long calculateMatchingScore(String[] current, String[] target) {
+        return IntStream.range(0, Math.min(current.length, target.length))
+                .filter(segment -> current[segment].equals(WILD_CARD) ||
+                        current[segment].equalsIgnoreCase(target[segment]))
+                .count();
+    }
+
+    private long getMatchingScore(String[] current, String[] target) {
+        final long totalMatches = calculateMatchingScore(current,target);
+        final boolean allSegmentsMatched = totalMatches == current.length;
+        return allSegmentsMatched ? totalMatches: 0;
+    }
+
+    private Candidate getScoredCandidate(Tuple2<String,String[]> candidate, String[] target) {
+        final long matchingScore = getMatchingScore(candidate.getT2(), target);
+        return new Candidate(candidate.getT1(), matchingScore);
+    }
+
+    private String matchMissingKey(Set<String> sources, String target ) {
+        final String[] targetSegments = getSegments(target);
+        return  sources.stream()
+                .map(source -> Tuples.of(source, getSegments(source)))
+                .filter(source -> isCandidate(source.getT2(), targetSegments))
+                .map(candidate -> getScoredCandidate(candidate, targetSegments))
+                .max(Candidate::compareTo)
+                .map(Candidate::getKey)
+                .orElse(target);
+    }
+
+    public String match(Set<String> sources, String target ) {
+        return sources.contains(target) || sources.isEmpty() ?
+                target :
+                matchMissingKey(sources,target);
+    }
+}

async/async-commons/src/main/java/org/reactivecommons/async/impl/utils/matcher/Matcher.java

@@ -0,0 +1,7 @@
+package org.reactivecommons.async.impl.utils.matcher;
+
+import java.util.Set;
+
+public interface Matcher {
+    String match(Set<String> sources, String target);
+}

async/async-commons/src/test/java/org/reactivecommons/async/impl/utils/matcher/KeyMatcherPerformanceTest.java

@@ -0,0 +1,63 @@
+package org.reactivecommons.async.impl.utils.matcher;
+
+import org.databene.contiperf.PerfTest;
+import org.databene.contiperf.Required;
+import org.databene.contiperf.junit.ContiPerfRule;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@ContextConfiguration(classes = KeyMatcher.class)
+@RunWith(SpringJUnit4ClassRunner.class)
+public class KeyMatcherPerformanceTest {
+
+    @Rule
+    public ContiPerfRule contiPerfRule = new ContiPerfRule();
+    Map<String, String> candidates = new HashMap<>();
+
+    private KeyMatcher keyMatcher = new KeyMatcher();
+
+
+    @Before
+    public void init() {
+        ClassLoader classLoader = getClass().getClassLoader();
+        File file = new File(classLoader.getResource("candidateNamesForMatching.txt").getFile());
+        try {
+             Set<String> names =  new HashSet<>(Files
+                    .readAllLines(Paths.get(file.getAbsolutePath())));
+            candidates = names.stream()
+                    .collect(Collectors.toMap(name -> name, name -> name));
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+
+    @Test
+    @PerfTest(threads = 4, duration = 4000)
+    @Required(average = 200, max = 10000, throughput = 200)
+    public void getFromMapTest() {
+        String existentKey = "System.segment.3073.event.action";
+        candidates.get(existentKey);
+    }
+
+
+    @Test
+    @PerfTest(threads = 4, duration = 4000)
+    @Required(average = 200, max = 10000, throughput = 200)
+    public void matchNameBeforeTest() {
+        keyMatcher.match(candidates.keySet(), "System.segment.1.event.action");
+    }
+}

async/async-commons/src/test/java/org/reactivecommons/async/impl/utils/matcher/KeyMatcherTest.java

@@ -0,0 +1,82 @@
+package org.reactivecommons.async.impl.utils.matcher;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.junit.Assert.*;
+
+public class KeyMatcherTest {
+    private KeyMatcher keyMatcher;
+    private Set<String> listeners;
+
+    @Before
+    public void init() {
+        keyMatcher =  new KeyMatcher();
+        listeners = new HashSet<>();
+        listeners.add("A.*");
+        listeners.add("A.B");
+        listeners.add("A.B.*");
+        listeners.add("A.B.C");
+        listeners.add("A.B.*.D");
+        listeners.add("A.B.C.D");
+    }
+
+    @Test
+    public void matchNonExistentFirstLevel() {
+        String nonExistentTarget = "A.X";
+        final String match = keyMatcher.match(listeners, nonExistentTarget);
+        assertEquals("A.*", match);
+    }
+
+    @Test
+    public void matchExistentFirstLevel() {
+        String existentTarget = "A.B";
+        final String match = keyMatcher.match(listeners, existentTarget);
+        assertEquals("A.B", match);
+    }
+
+    @Test
+    public void matchNonExistentSecondLevel() {
+        String nonExistentTarget = "A.B.X";
+        final String match = keyMatcher.match(listeners, nonExistentTarget);
+        assertEquals("A.B.*", match);
+    }
+
+    @Test
+    public void matchExistentSecondLevel() {
+        String existentTarget = "A.B.C";
+        final String match = keyMatcher.match(listeners, existentTarget);
+        assertEquals("A.B.C", match);
+    }
+
+    @Test
+    public void matchNonExistentThirdLevel() {
+        String nonExistentTarget = "A.B.X.D";
+        final String match = keyMatcher.match(listeners, nonExistentTarget);
+        assertEquals("A.B.*.D", match);
+    }
+
+    @Test
+    public void matchExistentThirdLevel() {
+        String existentTarget = "A.B.C.D";
+        final String match = keyMatcher.match(listeners, existentTarget);
+        assertEquals("A.B.C.D", match);
+    }
+
+    @Test
+    public void matchDefaultForNonExistent() {
+        String nonExistentTarget = "A.W.X.Y.Z";
+        final String match = keyMatcher.match(listeners, nonExistentTarget);
+        assertEquals("A.*", match);
+    }
+
+    @Test
+    public void matchDefaultForNonExistentSecondLevel() {
+        String nonExistentTarget = "A.B.X.Y.Z";
+        final String match = keyMatcher.match(listeners, nonExistentTarget);
+        assertEquals("A.B.*", match);
+    }
+}