Refactor commons

Author: andmagom

async/async-parent/async-parent.gradle

@@ -0,0 +1,19 @@
+plugins {
+    id 'java-library'
+    id 'maven'
+    id 'maven-publish'
+}
+
+test.onlyIf { true }
+
+repositories {
+    // Use jcenter for resolving dependencies.
+    // You can declare any Maven/Ivy/file repository here.
+    jcenter()
+}
+
+dependencies {
+    api project(":async-commons")
+
+    testImplementation 'io.projectreactor:reactor-test'
+}

async/async-parent/src/main/java/org/reactivecommons/async/parent/CommandExecutor.java

@@ -0,0 +1,23 @@
+package org.reactivecommons.async.parent;
+
+
+import org.reactivecommons.api.domain.Command;
+import org.reactivecommons.async.api.handlers.CommandHandler;
+import org.reactivecommons.async.impl.communications.Message;
+import reactor.core.publisher.Mono;
+
+import java.util.function.Function;
+
+public class CommandExecutor<T> {
+    private final CommandHandler<T> eventHandler;
+    private final Function<Message, Command<T>> converter;
+
+    public CommandExecutor(CommandHandler<T> eventHandler, Function<Message, Command<T>> converter) {
+        this.eventHandler = eventHandler;
+        this.converter = converter;
+    }
+
+    public Mono<Void> execute(Message rawMessage){
+        return eventHandler.handle(converter.apply(rawMessage));
+    }
+}

async/async-parent/src/main/java/org/reactivecommons/async/parent/DiscardNotifier.java

@@ -0,0 +1,10 @@
+package org.reactivecommons.async.parent;
+
+import org.reactivecommons.async.impl.communications.Message;
+import reactor.core.publisher.Mono;
+
+public interface DiscardNotifier {
+
+    Mono<Void> notifyDiscard(Message message);
+
+}

async/async-parent/src/main/java/org/reactivecommons/async/parent/EventExecutor.java

@@ -0,0 +1,23 @@
+package org.reactivecommons.async.parent;
+
+
+import org.reactivecommons.api.domain.DomainEvent;
+import org.reactivecommons.async.api.handlers.EventHandler;
+import org.reactivecommons.async.impl.communications.Message;
+import reactor.core.publisher.Mono;
+
+import java.util.function.Function;
+
+public class EventExecutor<T> {
+    private final EventHandler<T> eventHandler;
+    private final Function<Message, DomainEvent<T>> converter;
+
+    public EventExecutor(EventHandler<T> eventHandler, Function<Message, DomainEvent<T>> converter) {
+        this.eventHandler = eventHandler;
+        this.converter = converter;
+    }
+
+    public Mono<Void> execute(Message rawMessage){
+        return eventHandler.handle(converter.apply(rawMessage));
+    }
+}

async/async-parent/src/main/java/org/reactivecommons/async/parent/FallbackStrategy.java

@@ -0,0 +1,12 @@
+package org.reactivecommons.async.parent;
+
+public enum FallbackStrategy {
+    FAST_RETRY("ATTENTION!! Fast retry message to same Queue: %s"),
+    DEFINITIVE_DISCARD("ATTENTION!! DEFINITIVE DISCARD!! of the message: %s"),
+    RETRY_DLQ("ATTENTION!! Sending message to Retry DLQ: %s");
+
+    public final String message;
+    FallbackStrategy(String message){
+        this.message = message;
+    }
+}

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

@@ -0,0 +1,68 @@
+package org.reactivecommons.async.parent;
+
+import lombok.RequiredArgsConstructor;
+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 java.util.Collection;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+@RequiredArgsConstructor
+public class HandlerResolver {
+
+    private final Map<String, RegisteredQueryHandler<?, ?>> queryHandlers;
+    private final Map<String, RegisteredEventListener<?>> eventListeners;
+    private final Map<String, RegisteredEventListener<?>> eventNotificationListeners;
+    private final Map<String, RegisteredEventListener<?>> dynamicEventsHandlers;
+    private final Map<String, RegisteredCommandHandler<?>> commandHandlers;
+
+    @SuppressWarnings("unchecked")
+    public <T, M> RegisteredQueryHandler<T, M> getQueryHandler(String path) {
+        return (RegisteredQueryHandler<T, M>) queryHandlers.get(path);
+    }
+
+    @SuppressWarnings("unchecked")
+    public <T> RegisteredCommandHandler<T> getCommandHandler(String path) {
+        return (RegisteredCommandHandler<T>) commandHandlers.get(path);
+    }
+
+    @SuppressWarnings("unchecked")
+    public <T> RegisteredEventListener<T> getEventListener(String path) {
+        return (RegisteredEventListener<T>) eventListeners.get(path);
+    }
+
+    @SuppressWarnings("unchecked")
+    public <T> RegisteredEventListener<T> getDynamicEventsHandler(String path) {
+        return (RegisteredEventListener<T>) dynamicEventsHandlers.get(path);
+    }
+
+    public Collection<RegisteredEventListener<?>> getNotificationListeners() {
+        return eventNotificationListeners.values();
+    }
+
+    @SuppressWarnings("unchecked")
+    public <T> RegisteredEventListener<T> getNotificationListener(String path) {
+        return (RegisteredEventListener<T>) eventNotificationListeners.get(path);
+    }
+
+    public Collection<RegisteredEventListener<?>> getEventListeners() {
+        return eventListeners.values();
+    }
+
+    public Set<String> getToListenEventNames() {
+        Set<String> toListenEventNames = new HashSet<>(eventListeners.size() +
+                dynamicEventsHandlers.size());
+
+        toListenEventNames.addAll(eventListeners.keySet());
+        toListenEventNames.addAll(dynamicEventsHandlers.keySet());
+
+        return toListenEventNames;
+    }
+
+    void addEventListener(RegisteredEventListener listener) {
+        eventListeners.put(listener.getPath(), listener);
+    }
+}

async/async-parent/src/main/java/org/reactivecommons/async/parent/QueryExecutor.java

@@ -0,0 +1,29 @@
+package org.reactivecommons.async.parent;
+
+
+import org.reactivecommons.async.api.From;
+import org.reactivecommons.async.api.handlers.QueryHandlerDelegate;
+import org.reactivecommons.async.impl.communications.Message;
+import reactor.core.publisher.Mono;
+
+import java.util.function.Function;
+
+import static org.reactivecommons.async.impl.Headers.CORRELATION_ID;
+import static org.reactivecommons.async.impl.Headers.REPLY_ID;
+
+public class QueryExecutor<T, M> {
+    private final QueryHandlerDelegate<T, M> queryHandler;
+    private final Function<Message, M> converter;
+
+    public QueryExecutor(QueryHandlerDelegate<T, M> queryHandler, Function<Message, M> converter) {
+        this.queryHandler = queryHandler;
+        this.converter = converter;
+    }
+
+    public Mono<T> execute(Message rawMessage) {
+        From from = new From();
+        from.setCorrelationID(rawMessage.getProperties().getHeaders().getOrDefault(CORRELATION_ID, "").toString());
+        from.setReplyID(rawMessage.getProperties().getHeaders().getOrDefault(REPLY_ID, "").toString());
+        return queryHandler.handle(from, converter.apply(rawMessage));
+    }
+}

async/async-parent/src/main/java/org/reactivecommons/async/parent/communications/Message.java

@@ -0,0 +1,20 @@
+package org.reactivecommons.async.parent.communications;
+
+import java.util.Map;
+
+/**
+ * Simple Internal Message representation
+ * @author Daniel Bustamante Ospina
+ */
+public interface Message {
+
+    byte[] getBody();
+    Properties getProperties();
+
+    interface Properties {
+        String getContentType();
+        String getContentEncoding();
+        long getContentLength();
+        Map<String, Object> getHeaders();
+    }
+}

async/async-parent/src/main/java/org/reactivecommons/async/parent/config/BrokerConfig.java

@@ -0,0 +1,47 @@
+package org.reactivecommons.async.parent.config;
+
+import java.time.Duration;
+import java.util.UUID;
+
+public class BrokerConfig {
+    private final String routingKey = UUID.randomUUID().toString().replaceAll("-", "");
+    private final boolean persistentQueries;
+    private final boolean persistentCommands;
+    private final boolean persistentEvents;
+    private final Duration replyTimeout;
+
+    public BrokerConfig() {
+        this.persistentQueries = false;
+        this.persistentCommands = true;
+        this.persistentEvents = true;
+        this.replyTimeout = Duration.ofSeconds(15);
+    }
+
+    public BrokerConfig(boolean persistentQueries, boolean persistentCommands, boolean persistentEvents, Duration replyTimeout) {
+        this.persistentQueries = persistentQueries;
+        this.persistentCommands = persistentCommands;
+        this.persistentEvents = persistentEvents;
+        this.replyTimeout = replyTimeout;
+    }
+
+    public boolean isPersistentQueries() {
+        return persistentQueries;
+    }
+
+    public boolean isPersistentCommands() {
+        return persistentCommands;
+    }
+
+    public boolean isPersistentEvents() {
+        return persistentEvents;
+    }
+
+    public Duration getReplyTimeout() {
+        return replyTimeout;
+    }
+
+    public String getRoutingKey() {
+        return routingKey;
+    }
+
+}

async/async-parent/src/main/java/org/reactivecommons/async/parent/config/IBrokerConfigProps.java

@@ -0,0 +1,19 @@
+package org.reactivecommons.async.parent.config;
+
+public interface IBrokerConfigProps {
+    String getEventsQueue();
+
+    String getQueriesQueue();
+
+    String getCommandsQueue();
+
+    String getReplyQueue();
+
+    String getAppName();
+
+    String getDomainEventsExchangeName();
+
+    String getDirectMessagesExchangeName();
+
+    java.util.concurrent.atomic.AtomicReference<String> getReplyQueueName();
+}