Initial api version, auto-recovery support

Author: Daniel Bustamante Ospina

async/async-commons-api/async-commons-api.gradle

@@ -2,7 +2,5 @@ dependencies {
     compile project(":domain-events-api")
     implementation 'io.projectreactor:reactor-core'
     implementation 'io.projectreactor.addons:reactor-extra'
-    implementation 'org.springframework.cloud:spring-cloud-stream-reactive'
-    implementation 'org.springframework.cloud:spring-cloud-starter-stream-rabbit'
     testImplementation 'io.projectreactor:reactor-test'
 }

async/async-commons-api/src/main/java/us/sofka/commons/reactive/async/api/AsyncQuery.java renamed to async/async-commons-api/src/main/java/org/reactivecommons/async/api/AsyncQuery.java

@@ -1,4 +1,4 @@
-package us.sofka.commons.reactive.async.api;
+package org.reactivecommons.async.api;
 
 import lombok.Data;
 

async/async-commons-api/src/main/java/us/sofka/commons/reactive/async/api/AsyncQueryHandler.java renamed to async/async-commons-api/src/main/java/org/reactivecommons/async/api/AsyncQueryHandler.java

@@ -1,4 +1,4 @@
-package us.sofka.commons.reactive.async.api;
+package org.reactivecommons.async.api;
 
 import reactor.core.publisher.Mono;
 

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

@@ -0,0 +1,7 @@
+package org.reactivecommons.async.api;
+
+import org.reactivecommons.async.api.handlers.CommandHandler;
+
+public interface DefaultCommandHandler<T> extends CommandHandler<T> {
+
+}

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

@@ -0,0 +1,7 @@
+package org.reactivecommons.async.api;
+
+import org.reactivecommons.async.api.handlers.QueryHandler;
+
+public interface DefaultQueryHandler<T, C> extends QueryHandler<T, C> {
+
+}

async/async-commons-api/src/main/java/us/sofka/commons/reactive/async/api/DirectAsyncGateway.java renamed to async/async-commons-api/src/main/java/org/reactivecommons/async/api/DirectAsyncGateway.java

@@ -1,5 +1,6 @@
-package us.sofka.commons.reactive.async.api;
+package org.reactivecommons.async.api;
 
+import org.reactivecommons.api.domain.Command;
 import reactor.core.publisher.Mono;
 
 public interface DirectAsyncGateway {

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

@@ -0,0 +1,85 @@
+package org.reactivecommons.async.api;
+
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.RequiredArgsConstructor;
+import org.reactivecommons.async.api.handlers.CommandHandler;
+import org.reactivecommons.async.api.handlers.EventHandler;
+import org.reactivecommons.async.api.handlers.QueryHandler;
+
+import java.lang.reflect.ParameterizedType;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+@Getter
+@NoArgsConstructor(access = AccessLevel.PACKAGE)
+public class HandlerRegistry {
+
+    public static HandlerRegistry register(){
+        return new HandlerRegistry();
+    }
+
+    private final List<RegisteredQueryHandler<?, ?>> handlers = new CopyOnWriteArrayList<>();
+    private final List<RegisteredEventListener> eventListeners = new CopyOnWriteArrayList<>();
+    private final List<RegisteredCommandHandler> commandHandlers = new CopyOnWriteArrayList<>();
+
+    public <T> HandlerRegistry listenEvent(String eventName, EventHandler<T> fn, Class<T> eventClass){
+        eventListeners.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;
+    }
+
+    @SuppressWarnings("unchecked")
+    public <T, R> HandlerRegistry serveQuery(String commandId, QueryHandler<T, R> handler){
+        return serveQuery(commandId, handler, inferGenericParameterType(handler));
+    }
+
+    @SuppressWarnings("unchecked")
+    private <T, R> Class<R> inferGenericParameterType(QueryHandler<T, R> handler){
+        try{
+            ParameterizedType genericSuperclass = (ParameterizedType) handler.getClass().getGenericInterfaces()[0];
+            return  (Class<R>) genericSuperclass.getActualTypeArguments()[1];
+        }catch (Exception e){
+            throw new RuntimeException("Fail to infer generic Query class, please use serveQuery(path, handler, class) instead");
+        }
+    }
+
+    public <T, R> HandlerRegistry serveQuery(String commandId, QueryHandler<T, R> handler, Class<R> queryClass){
+        handlers.add(new RegisteredQueryHandler<>(commandId, handler, queryClass));
+        return this;
+    }
+
+
+    @RequiredArgsConstructor
+    @Getter
+    public static class RegisteredQueryHandler<T, R> {
+        private final String path;
+        private final QueryHandler<T, R> handler;
+        private final Class<R> queryClass;
+    }
+
+    @RequiredArgsConstructor
+    @Getter
+    public static class RegisteredEventListener<T> {
+        private final String path;
+        private final EventHandler<T> handler;
+        private final Class<T> inputClass;
+    }
+
+    @RequiredArgsConstructor
+    @Getter
+    public static class RegisteredCommandHandler<T> {
+        private final String path;
+        private final CommandHandler<T> handler;
+        private final Class<T> inputClass;
+    }
+
+}
+
+
+

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

@@ -0,0 +1,20 @@
+package org.reactivecommons.async.api;
+
+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-commons-api/src/main/java/org/reactivecommons/async/api/MessageConversionException.java

@@ -0,0 +1,8 @@
+package org.reactivecommons.async.api;
+
+public class MessageConversionException extends RuntimeException {
+
+    public MessageConversionException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

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

@@ -0,0 +1,16 @@
+package org.reactivecommons.async.api;
+
+import org.reactivecommons.api.domain.Command;
+import org.reactivecommons.api.domain.DomainEvent;
+
+public interface MessageConverter {
+
+    <T> AsyncQuery<T> readAsyncQuery(Message message, Class<T> bodyClass);
+
+    <T> DomainEvent<T> readDomainEvent(Message message, Class<T> bodyClass);
+
+    <T> Command<T> readCommand(Message message, Class<T> bodyClass);
+
+    Message toMessage(Object object);
+
+}

async/async-commons-api/src/main/java/org/reactivecommons/async/api/handlers/CommandHandler.java

@@ -0,0 +1,6 @@
+package org.reactivecommons.async.api.handlers;
+
+import org.reactivecommons.api.domain.Command;
+
+public interface CommandHandler<T> extends GenericHandler<Void, Command<T>> {
+}

async/async-commons-api/src/main/java/org/reactivecommons/async/api/handlers/EventHandler.java

@@ -0,0 +1,6 @@
+package org.reactivecommons.async.api.handlers;
+
+import org.reactivecommons.api.domain.DomainEvent;
+
+public interface EventHandler<T> extends GenericHandler<Void, DomainEvent<T>> {
+}

async/async-commons-api/src/main/java/org/reactivecommons/async/api/handlers/GenericHandler.java

@@ -0,0 +1,7 @@
+package org.reactivecommons.async.api.handlers;
+
+import reactor.core.publisher.Mono;
+
+public interface GenericHandler<T, M> {
+    Mono<T> handle(M message);
+}

async/async-commons-api/src/main/java/org/reactivecommons/async/api/handlers/QueryHandler.java

@@ -0,0 +1,5 @@
+package org.reactivecommons.async.api.handlers;
+
+public interface QueryHandler<T, C> extends GenericHandler<T, C> {
+
+}

async/async-commons-api/src/main/java/us/sofka/commons/reactive/async/api/CommandHandler.java

@@ -3,11 +3,12 @@ dependencies {
     compile project(":domain-events-api")
 
     compile 'io.projectreactor:reactor-core'
-//    implementation 'io.projectreactor.addons:reactor-extra'
-//    compile 'org.springframework.cloud:spring-cloud-stream-reactive'
-//    compile 'org.springframework.cloud:spring-cloud-starter-stream-rabbit'
-    compile 'org.springframework.boot:spring-boot-starter-amqp'
+    implementation 'io.projectreactor.addons:reactor-extra'
+    compile('org.springframework.boot:spring-boot-starter')
+//    compile 'org.springframework.boot:spring-boot-starter-amqp'
+//    compile group: 'org.springframework.amqp', name: 'spring-amqp'
     compile "io.projectreactor.rabbitmq:reactor-rabbitmq:1.0.0.RC1"
+//    compile project(":async:libs:reactor-rabbitmq")
     compile 'com.fasterxml.jackson.core:jackson-databind:2.9.7'
     testImplementation 'io.projectreactor:reactor-test'
 }

async/async-commons-api/src/main/java/us/sofka/commons/reactive/async/api/DomainEvent.java

@@ -0,0 +1,23 @@
+package org.reactivecommons.async.impl;
+
+
+import org.reactivecommons.api.domain.Command;
+import org.reactivecommons.async.api.Message;
+import org.reactivecommons.async.api.handlers.CommandHandler;
+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-commons-api/src/main/java/us/sofka/commons/reactive/async/api/DomainEventBus.java

@@ -0,0 +1,23 @@
+package org.reactivecommons.async.impl;
+
+
+import org.reactivecommons.api.domain.DomainEvent;
+import org.reactivecommons.async.api.handlers.EventHandler;
+import org.reactivecommons.async.api.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-commons/async-commons.gradle

@@ -0,0 +1,36 @@
+package org.reactivecommons.async.impl;
+
+import lombok.RequiredArgsConstructor;
+import org.reactivecommons.async.api.handlers.QueryHandler;
+import org.reactivecommons.async.api.HandlerRegistry;
+
+import java.util.Collection;
+import java.util.Map;
+
+@RequiredArgsConstructor
+public class HandlerResolver {
+
+    private final Map<String, QueryHandler<?, ?>> queryHandlers;
+    private final Map<String, HandlerRegistry.RegisteredEventListener> eventListeners;
+    private final Map<String, HandlerRegistry.RegisteredCommandHandler> commandHandlers;
+
+
+    @SuppressWarnings("unchecked")
+    public <T, R> QueryHandler<T, R> getQueryHandler(String path) {
+        return (QueryHandler<T, R>) queryHandlers.get(path);
+    }
+
+    @SuppressWarnings("unchecked")
+    public <T> HandlerRegistry.RegisteredCommandHandler<T> getCommandHandler(String path) {
+        return commandHandlers.get(path);
+    }
+
+    @SuppressWarnings("unchecked")
+    public <T> HandlerRegistry.RegisteredEventListener<T> getEventListener(String path) {
+        return eventListeners.get(path);
+    }
+
+    public Collection<HandlerRegistry.RegisteredEventListener> getEventListeners() {
+        return eventListeners.values();
+    }
+}

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

@@ -0,0 +1,22 @@
+package org.reactivecommons.async.impl;
+
+
+import org.reactivecommons.async.api.handlers.QueryHandler;
+import org.reactivecommons.async.api.Message;
+import reactor.core.publisher.Mono;
+
+import java.util.function.Function;
+
+public class QueryExecutor<C, R> {
+    private final QueryHandler<R, C> queryHandler;
+    private final Function<Message, C> converter;
+
+    public QueryExecutor(QueryHandler<R, C> queryHandler, Function<Message, C> converter) {
+        this.queryHandler = queryHandler;
+        this.converter = converter;
+    }
+
+    public Mono<R> execute(Message rawMessage){
+        return queryHandler.handle(converter.apply(rawMessage));
+    }
+}