chore(next): add native support for Cloud Events (#115)

* Add CloudEvent specification to commands and events listeners and gateways
* Exclude cloud events

---------

Co-authored-by: Angel David Santa Giraldo <[email protected]>
Co-authored-by: jcanacon <jcanacon>
Co-authored-by: Juan C Galvis <[email protected]>

Author: 3 people

acceptance/async-tests/src/test/java/org/reactivecommons/test/DynamicRegistryTest.java

@@ -4,6 +4,7 @@
 import org.reactivecommons.api.domain.DomainEvent;
 import org.reactivecommons.api.domain.DomainEventBus;
 import org.reactivecommons.async.api.DynamicRegistry;
+import org.reactivecommons.async.api.handlers.DomainEventHandler;
 import org.reactivecommons.async.api.handlers.EventHandler;
 import org.reactivecommons.async.impl.config.annotations.EnableDomainEventBus;
 import org.reactivecommons.async.impl.config.annotations.EnableMessageListeners;
@@ -36,7 +37,7 @@ class DynamicRegistryTest {
     @Test
     void shouldReceiveResponse() {
         UnicastProcessor<String> result = UnicastProcessor.create();
-        EventHandler<String> fn = message -> fromRunnable(() -> result.onNext(message.getData()));
+        DomainEventHandler<String> fn = message -> fromRunnable(() -> result.onNext(message.getData()));
 
         dynamicRegistry.listenEvent("test.event", fn, String.class).block();
         final Publisher<Void> emit = eventBus.emit(new DomainEvent<>("test.event", "42", "Hello"));

acceptance/async-tests/src/test/java/org/reactivecommons/test/SimpleDirectCommunicationTest.java

@@ -6,6 +6,7 @@
 import org.reactivecommons.async.api.DirectAsyncGateway;
 import org.reactivecommons.async.api.HandlerRegistry;
 import org.reactivecommons.async.api.handlers.CommandHandler;
+import org.reactivecommons.async.api.handlers.DomainCommandHandler;
 import org.reactivecommons.async.impl.config.annotations.EnableDirectAsyncGateway;
 import org.reactivecommons.async.impl.config.annotations.EnableMessageListeners;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -81,7 +82,7 @@ public UnicastProcessor<Command<Long>> listener() {
             return UnicastProcessor.create();
         }
 
-        private CommandHandler<Long> handle(UnicastProcessor<Command<Long>> listener) {
+        private DomainCommandHandler<Long> handle(UnicastProcessor<Command<Long>> listener) {
             return command -> {
                 listener.onNext(command);
                 return empty();

acceptance/async-tests/src/test/java/org/reactivecommons/test/SimpleEventNotificationTest.java

@@ -4,6 +4,7 @@
 import org.reactivecommons.api.domain.DomainEvent;
 import org.reactivecommons.api.domain.DomainEventBus;
 import org.reactivecommons.async.api.HandlerRegistry;
+import org.reactivecommons.async.api.handlers.DomainEventHandler;
 import org.reactivecommons.async.api.handlers.EventHandler;
 import org.reactivecommons.async.impl.config.annotations.EnableDomainEventBus;
 import org.reactivecommons.async.impl.config.annotations.EnableMessageListeners;
@@ -65,7 +66,7 @@ public UnicastProcessor<DomainEvent<Long>> listener() {
             return UnicastProcessor.create();
         }
 
-        private EventHandler<Long> handle(UnicastProcessor<DomainEvent<Long>> listener) {
+        private DomainEventHandler<Long> handle(UnicastProcessor<DomainEvent<Long>> listener) {
             return command -> {
                 listener.onNext(command);
                 return empty();

acceptance/async-tests/src/test/java/org/reactivecommons/test/perf/BlockingCommandHandlePerfTest.java

@@ -5,6 +5,7 @@
 import org.reactivecommons.async.api.DirectAsyncGateway;
 import org.reactivecommons.async.api.HandlerRegistry;
 import org.reactivecommons.async.api.handlers.CommandHandler;
+import org.reactivecommons.async.api.handlers.DomainCommandHandler;
 import org.reactivecommons.async.impl.config.annotations.EnableDirectAsyncGateway;
 import org.reactivecommons.async.impl.config.annotations.EnableMessageListeners;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -89,7 +90,7 @@ public UnicastProcessor<Command<Long>> listener() {
             return UnicastProcessor.create();
         }
 
-        private CommandHandler<Long> handle(UnicastProcessor<Command<Long>> listener) {
+        private DomainCommandHandler<Long> handle(UnicastProcessor<Command<Long>> listener) {
             return command -> {
                 return fromRunnable(() -> {
 //                    out.println("Received at: " + System.currentTimeMillis()/1000);

acceptance/async-tests/src/test/java/org/reactivecommons/test/perf/ParallelOnBlockingInSubscriptionTimeTest.java

@@ -6,6 +6,7 @@
 import org.reactivecommons.async.api.DirectAsyncGateway;
 import org.reactivecommons.async.api.HandlerRegistry;
 import org.reactivecommons.async.api.handlers.CommandHandler;
+import org.reactivecommons.async.api.handlers.DomainCommandHandler;
 import org.reactivecommons.async.impl.config.annotations.EnableDirectAsyncGateway;
 import org.reactivecommons.async.impl.config.annotations.EnableMessageListeners;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -91,7 +92,7 @@ public UnicastProcessor<Command<Long>> listener() {
             return UnicastProcessor.create();
         }
 
-        private CommandHandler<Long> handle(UnicastProcessor<Command<Long>> listener) {
+        private DomainCommandHandler<Long> handle(UnicastProcessor<Command<Long>> listener) {
             return command -> {
 //                out.println("Received at: " + System.currentTimeMillis()/1000);
                 try {

acceptance/async-tests/src/test/java/org/reactivecommons/test/perf/SimpleCommandHandlePerfTest.java

@@ -6,6 +6,7 @@
 import org.reactivecommons.async.api.DirectAsyncGateway;
 import org.reactivecommons.async.api.HandlerRegistry;
 import org.reactivecommons.async.api.handlers.CommandHandler;
+import org.reactivecommons.async.api.handlers.DomainCommandHandler;
 import org.reactivecommons.async.impl.config.annotations.EnableDirectAsyncGateway;
 import org.reactivecommons.async.impl.config.annotations.EnableMessageListeners;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -88,7 +89,7 @@ public UnicastProcessor<Command<Long>> listener() {
             return UnicastProcessor.create();
         }
 
-        private CommandHandler<Long> handle(UnicastProcessor<Command<Long>> listener) {
+        private DomainCommandHandler<Long> handle(UnicastProcessor<Command<Long>> listener) {
             return command -> {
                 out.println("Received at: " + System.currentTimeMillis() / 1000);
                 out.println("Received in: " + Thread.currentThread().getName());

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

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

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

@@ -17,8 +17,12 @@ public interface DirectAsyncGateway {
 
     Mono<Void> sendCommand(CloudEvent command, String targetName);
 
+    Mono<Void> sendCommand(CloudEvent command, String targetName, long delayMillis);
+
     Mono<Void> sendCommand(CloudEvent command, String targetName, String domain);
 
+    Mono<Void> sendCommand(CloudEvent command, String targetName, long delayMillis, String domain);
+
     <T, R> Mono<R> requestReply(AsyncQuery<T> query, String targetName, Class<R> type);
 
     <T, R> Mono<R> requestReply(AsyncQuery<T> query, String targetName, Class<R> type, String domain);

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

@@ -1,5 +1,6 @@
 package org.reactivecommons.async.api;
 
+import org.reactivecommons.async.api.handlers.DomainEventHandler;
 import org.reactivecommons.async.api.handlers.EventHandler;
 import org.reactivecommons.async.api.handlers.QueryHandler;
 import org.reactivecommons.async.api.handlers.QueryHandlerDelegate;
@@ -8,7 +9,7 @@
 public interface DynamicRegistry {
 
     @Deprecated
-    <T> Mono<Void> listenEvent(String eventName, EventHandler<T> fn, Class<T> eventClass);
+    <T> Mono<Void> listenEvent(String eventName, DomainEventHandler<T> fn, Class<T> eventClass);
 
     <T, R> void serveQuery(String resource, QueryHandler<T, R> handler, Class<R> queryClass);
 

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

@@ -1,13 +1,13 @@
 package org.reactivecommons.async.api;
 
 import io.cloudevents.CloudEvent;
-import io.cloudevents.core.provider.EventFormatProvider;
-import io.cloudevents.jackson.JsonFormat;
 import lombok.AccessLevel;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
-import org.reactivecommons.async.api.handlers.CommandHandler;
-import org.reactivecommons.async.api.handlers.EventHandler;
+import org.reactivecommons.async.api.handlers.CloudCommandHandler;
+import org.reactivecommons.async.api.handlers.CloudEventHandler;
+import org.reactivecommons.async.api.handlers.DomainCommandHandler;
+import org.reactivecommons.async.api.handlers.DomainEventHandler;
 import org.reactivecommons.async.api.handlers.QueryHandler;
 import org.reactivecommons.async.api.handlers.QueryHandlerDelegate;
 import org.reactivecommons.async.api.handlers.registered.RegisteredCommandHandler;
@@ -24,11 +24,11 @@
 @NoArgsConstructor(access = AccessLevel.PACKAGE)
 public class HandlerRegistry {
     public static final String DEFAULT_DOMAIN = "app";
-    private final Map<String, List<RegisteredEventListener<?>>> domainEventListeners = new ConcurrentHashMap<>();
-    private final List<RegisteredEventListener<?>> dynamicEventHandlers = new CopyOnWriteArrayList<>();
-    private final List<RegisteredEventListener<?>> eventNotificationListener = new CopyOnWriteArrayList<>();
+    private final Map<String, List<RegisteredEventListener<?, ?>>> domainEventListeners = new ConcurrentHashMap<>();
+    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<>();
+    private final List<RegisteredCommandHandler<?, ?>> commandHandlers = new CopyOnWriteArrayList<>();
 
 
     public static HandlerRegistry register() {
@@ -37,75 +37,103 @@ public static HandlerRegistry register() {
         return instance;
     }
 
-    public <T> HandlerRegistry listenDomainEvent(String domain, String eventName, EventHandler<T> handler, Class<T> eventClass) {
+    public <T> HandlerRegistry listenDomainEvent(String domain, String eventName, DomainEventHandler<T> handler, Class<T> eventClass) {
         domainEventListeners.computeIfAbsent(domain, ignored -> new CopyOnWriteArrayList<>())
-        .add(new RegisteredEventListener<>(eventName, handler, eventClass));
+                .add(new RegisteredEventListener<>(eventName, handler, eventClass));
         return this;
     }
 
-    public <T> HandlerRegistry listenEvent(String eventName, EventHandler<T> handler, Class<T> eventClass) {
+    public HandlerRegistry listenDomainCloudEvent(String domain, String eventName, CloudEventHandler handler) {
+        domainEventListeners.computeIfAbsent(domain, ignored -> new CopyOnWriteArrayList<>())
+                .add(new RegisteredEventListener<>(eventName, handler, CloudEvent.class));
+        return this;
+    }
+
+    public <T> HandlerRegistry listenEvent(String eventName, DomainEventHandler<T> handler, Class<T> eventClass) {
         domainEventListeners.computeIfAbsent(DEFAULT_DOMAIN, ignored -> new CopyOnWriteArrayList<>())
-        .add(new RegisteredEventListener<>(eventName, handler, eventClass));
+                .add(new RegisteredEventListener<>(eventName, handler, eventClass));
         return this;
     }
 
-    public <T> HandlerRegistry listenEvent(String eventName, EventHandler<T> handler) {
-        return listenEvent(eventName, handler, inferGenericParameterType(handler));
+    public HandlerRegistry listenCloudEvent(String eventName, CloudEventHandler handler) {
+        domainEventListeners.computeIfAbsent(DEFAULT_DOMAIN, ignored -> new CopyOnWriteArrayList<>())
+                .add(new RegisteredEventListener<>(eventName, handler, CloudEvent.class));
+        return this;
     }
 
-    public <T> HandlerRegistry listenNotificationEvent(String eventName, EventHandler<T> handler, Class<T> eventClass) {
+    public <T> HandlerRegistry listenNotificationEvent(String eventName, DomainEventHandler<T> handler, Class<T> eventClass) {
         eventNotificationListener.add(new RegisteredEventListener<>(eventName, handler, eventClass));
         return this;
     }
 
-    public <T> HandlerRegistry handleDynamicEvents(String eventNamePattern, EventHandler<T> handler, Class<T> eventClass) {
+    public HandlerRegistry listenNotificationCloudEvent(String eventName, CloudEventHandler handler) {
+        eventNotificationListener.add(new RegisteredEventListener<>(eventName, handler, CloudEvent.class));
+        return this;
+    }
+
+    public <T> HandlerRegistry handleDynamicEvents(String eventNamePattern, DomainEventHandler<T> handler, Class<T> eventClass) {
         dynamicEventHandlers.add(new RegisteredEventListener<>(eventNamePattern, handler, eventClass));
         return this;
     }
 
-    public <T> HandlerRegistry handleDynamicEvents(String eventNamePattern, EventHandler<T> handler) {
-        return handleDynamicEvents(eventNamePattern, handler, inferGenericParameterType(handler));
+    public HandlerRegistry handleDynamicCloudEvents(String eventNamePattern, CloudEventHandler handler) {
+        dynamicEventHandlers.add(new RegisteredEventListener<>(eventNamePattern, handler, CloudEvent.class));
+        return this;
     }
 
-    public <T> HandlerRegistry handleCommand(String commandName, CommandHandler<T> fn, Class<T> commandClass) {
+    public <T> HandlerRegistry handleCommand(String commandName, DomainCommandHandler<T> fn, Class<T> commandClass) {
         commandHandlers.add(new RegisteredCommandHandler<>(commandName, fn, commandClass));
         return this;
     }
 
-    public <T> HandlerRegistry handleCommand(String commandName, CommandHandler<T> fn) {
-        commandHandlers.add(new RegisteredCommandHandler<>(commandName, fn, inferGenericParameterType(fn)));
+    public HandlerRegistry handleCloudEventCommand(String commandName, CloudCommandHandler handler) {
+        commandHandlers.add(new RegisteredCommandHandler<>(commandName, handler, CloudEvent.class));
         return this;
     }
 
-    public <T, R> HandlerRegistry serveQuery(String resource, QueryHandler<T, R> handler) {
-        return serveQuery(resource, handler, inferGenericParameterType(handler));
+    public <T, R> HandlerRegistry serveQuery(String resource, QueryHandler<T, R> handler, Class<R> queryClass) {
+        handlers.add(new RegisteredQueryHandler<>(resource, (ignored, message) -> handler.handle(message), queryClass));
+        return this;
     }
 
-    public <T, R> HandlerRegistry serveQuery(String resource, QueryHandler<T, R> handler, Class<R> queryClass) {
-        if(queryClass == CloudEvent.class){
-            handlers.add(new RegisteredQueryHandler<>(resource, (ignored, message) ->
-            {
-                CloudEvent query = EventFormatProvider
-                        .getInstance()
-                        .resolveFormat(JsonFormat.CONTENT_TYPE)
-                        .deserialize(message);
+    public <R> HandlerRegistry serveQuery(String resource, QueryHandlerDelegate<Void, R> handler, Class<R> queryClass) {
+        handlers.add(new RegisteredQueryHandler<>(resource, handler, queryClass));
+        return this;
+    }
 
-                return handler.handle((R) query);
+    public <R> HandlerRegistry serveCloudEventQuery(String resource, QueryHandler<R, CloudEvent> handler) {
+        handlers.add(new RegisteredQueryHandler<>(resource, (ignored, message) -> handler.handle(message), CloudEvent.class));
+        return this;
+    }
 
-            } , byte[].class));
-        }
-        else{
-            handlers.add(new RegisteredQueryHandler<>(resource, (ignored, message) -> handler.handle(message), queryClass));
-        }
+    public <R> HandlerRegistry serveCloudEventQuery(String resource, QueryHandlerDelegate<Void, CloudEvent> handler) {
+        handlers.add(new RegisteredQueryHandler<>(resource, handler, CloudEvent.class));
         return this;
     }
 
-    public <R> HandlerRegistry serveQuery(String resource, QueryHandlerDelegate<Void, R> handler, Class<R> queryClass) {
-        handlers.add(new RegisteredQueryHandler<>(resource, handler, queryClass));
+
+    @Deprecated(forRemoval = true)
+    public <T> HandlerRegistry listenEvent(String eventName, DomainEventHandler<T> handler) {
+        return listenEvent(eventName, handler, inferGenericParameterType(handler));
+    }
+
+    @Deprecated(forRemoval = true)
+    public <T> HandlerRegistry handleDynamicEvents(String eventNamePattern, DomainEventHandler<T> handler) {
+        return handleDynamicEvents(eventNamePattern, handler, inferGenericParameterType(handler));
+    }
+
+    @Deprecated(forRemoval = true)
+    public <T> HandlerRegistry handleCommand(String commandName, DomainCommandHandler<T> handler) {
+        commandHandlers.add(new RegisteredCommandHandler<>(commandName, handler, inferGenericParameterType(handler)));
         return this;
     }
 
+    @Deprecated(forRemoval = true)
+    public <T, R> HandlerRegistry serveQuery(String resource, QueryHandler<T, R> handler) {
+        return serveQuery(resource, handler, inferGenericParameterType(handler));
+    }
 
+    @Deprecated(forRemoval = true)
     @SuppressWarnings("unchecked")
     private <T, R> Class<R> inferGenericParameterType(QueryHandler<T, R> handler) {
         try {
@@ -117,8 +145,9 @@ private <T, R> Class<R> inferGenericParameterType(QueryHandler<T, R> handler) {
         }
     }
 
+    @Deprecated(forRemoval = true)
     @SuppressWarnings("unchecked")
-    private <T> Class<T> inferGenericParameterType(CommandHandler<T> handler) {
+    private <T> Class<T> inferGenericParameterType(DomainCommandHandler<T> handler) {
         try {
             ParameterizedType genericSuperclass = (ParameterizedType) handler.getClass().getGenericInterfaces()[0];
             return (Class<T>) genericSuperclass.getActualTypeArguments()[0];
@@ -128,7 +157,8 @@ private <T> Class<T> inferGenericParameterType(CommandHandler<T> handler) {
         }
     }
 
-    private <T> Class<T> inferGenericParameterType(EventHandler<T> handler) {
+    @Deprecated(forRemoval = true)
+    private <T> Class<T> inferGenericParameterType(DomainEventHandler<T> handler) {
         try {
             ParameterizedType genericSuperclass = (ParameterizedType) handler.getClass().getGenericInterfaces()[0];
             return (Class<T>) genericSuperclass.getActualTypeArguments()[0];

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

@@ -0,0 +1,6 @@
+package org.reactivecommons.async.api.handlers;
+
+import io.cloudevents.CloudEvent;
+
+public interface CloudCommandHandler extends CommandHandler<CloudEvent> {
+}

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

@@ -0,0 +1,6 @@
+package org.reactivecommons.async.api.handlers;
+
+import io.cloudevents.CloudEvent;
+
+public interface CloudEventHandler extends EventHandler<CloudEvent> {
+}

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

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

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

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

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

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

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

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

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

@@ -6,8 +6,8 @@
 
 @RequiredArgsConstructor
 @Getter
-public class RegisteredCommandHandler<T> {
+public class RegisteredCommandHandler<T,D> {
     private final String path;
-    private final CommandHandler<T> handler;
+    private final CommandHandler<D> handler;
     private final Class<T> inputClass;
 }

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

@@ -6,8 +6,8 @@
 
 @RequiredArgsConstructor
 @Getter
-public class RegisteredEventListener<T> {
+public class RegisteredEventListener<T, D> {
     private final String path;
-    private final EventHandler<T> handler;
+    private final EventHandler<D> handler;
     private final Class<T> inputClass;
 }