merge

jespinosas · jespinosas · commit b2b7017a5c89 · 2023-06-08T11:23:39.000-05:00
diff --git a/async/async-commons-api/async-commons-api.gradle b/async/async-commons-api/async-commons-api.gradle
@@ -7,4 +7,6 @@ dependencies {
     api project(':domain-events-api')
     compileOnly 'io.projectreactor:reactor-core'
     testImplementation 'io.projectreactor:reactor-test'
+    implementation 'io.cloudevents:cloudevents-json-jackson:2.5.0'
+
 }
diff --git a/async/async-commons-api/src/main/java/org/reactivecommons/async/api/DirectAsyncGateway.java b/async/async-commons-api/src/main/java/org/reactivecommons/async/api/DirectAsyncGateway.java
@@ -1,10 +1,13 @@
 package org.reactivecommons.async.api;
 
+import io.cloudevents.CloudEvent;
 import org.reactivecommons.api.domain.Command;
 import reactor.core.publisher.Mono;
 
 public interface DirectAsyncGateway {
     <T> Mono<Void> sendCommand(Command<T> command, String targetName);
+    Mono<Void> sendCommand(CloudEvent command, String targetName);
     <T, R> Mono<R> requestReply(AsyncQuery<T> query, String targetName, Class<R> type);
+    <R extends CloudEvent> Mono<R> requestReply(CloudEvent query, String targetName, Class<R> type);
     <T> Mono<Void> reply(T response, From from);
 }
diff --git a/async/async-commons-api/src/main/java/org/reactivecommons/async/api/HandlerRegistry.java b/async/async-commons-api/src/main/java/org/reactivecommons/async/api/HandlerRegistry.java
@@ -1,5 +1,8 @@
 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;
@@ -26,6 +29,7 @@ public class HandlerRegistry {
     private final List<RegisteredQueryHandler<?, ?>> handlers = new CopyOnWriteArrayList<>();
     private final List<RegisteredCommandHandler<?>> commandHandlers = new CopyOnWriteArrayList<>();
 
+
     public static HandlerRegistry register() {
         return new HandlerRegistry();
     }
@@ -75,7 +79,21 @@ public <T, R> HandlerRegistry serveQuery(String resource, QueryHandler<T, R> han
     }
 
     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));
+        if(queryClass == CloudEvent.class){
+            handlers.add(new RegisteredQueryHandler<>(resource, (ignored, message) ->
+            {
+                CloudEvent query = EventFormatProvider
+                        .getInstance()
+                        .resolveFormat(JsonFormat.CONTENT_TYPE)
+                        .deserialize(message);
+
+                return handler.handle((R) query);
+
+            } , byte[].class));
+        }
+        else{
+            handlers.add(new RegisteredQueryHandler<>(resource, (ignored, message) -> handler.handle(message), queryClass));
+        }
         return this;
     }
 
diff --git a/async/async-commons/src/main/java/org/reactivecommons/async/commons/converters/MessageConverter.java b/async/async-commons/src/main/java/org/reactivecommons/async/commons/converters/MessageConverter.java
@@ -1,5 +1,6 @@
 package org.reactivecommons.async.commons.converters;
 
+import io.cloudevents.CloudEvent;
 import org.reactivecommons.api.domain.Command;
 import org.reactivecommons.api.domain.DomainEvent;
 import org.reactivecommons.async.api.AsyncQuery;
diff --git a/async/async-rabbit/async-rabbit.gradle b/async/async-rabbit/async-rabbit.gradle
@@ -14,4 +14,5 @@ dependencies {
     api 'com.rabbitmq:amqp-client'
     api 'com.fasterxml.jackson.core:jackson-databind'
     testImplementation 'io.projectreactor:reactor-test'
+    implementation 'io.cloudevents:cloudevents-json-jackson:2.5.0'
 }
diff --git a/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/RabbitDirectAsyncGateway.java b/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/RabbitDirectAsyncGateway.java
@@ -1,5 +1,8 @@
 package org.reactivecommons.async.rabbit;
 
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.provider.EventFormatProvider;
+import io.cloudevents.jackson.JsonFormat;
 import io.micrometer.core.instrument.MeterRegistry;
 import org.reactivecommons.api.domain.Command;
 import org.reactivecommons.async.api.AsyncQuery;
@@ -56,6 +59,11 @@ public <T> Mono<Void> sendCommand(Command<T> command, String targetName) {
         return sender.sendWithConfirm(command, exchange, targetName, Collections.emptyMap(), persistentCommands);
     }
 
+    @Override
+    public Mono<Void> sendCommand(CloudEvent command, String targetName) {
+        return sendCommand(new Command<>(command.getType(), command.getId(), command), targetName);
+    }
+
     public <T> Flux<OutboundMessageResult> sendCommands(Flux<Command<T>> commands, String targetName) {
         return sender.sendWithConfirmBatch(commands, exchange, targetName, Collections.emptyMap(), persistentCommands);
     }
@@ -84,6 +92,11 @@ public <T, R> Mono<R> requestReply(AsyncQuery<T> query, String targetName, Class
                 .tap(Micrometer.metrics(meterRegistry));
     }
 
+    @Override
+    public <R extends CloudEvent> Mono<R> requestReply(CloudEvent query, String targetName, Class<R> type) {
+        return requestReply(new AsyncQuery<>(query.getType(), query), targetName, type);
+    }
+
     @Override
     public <T> Mono<Void> reply(T response, From from) {
         final HashMap<String, Object> headers = new HashMap<>();
diff --git a/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/RabbitDomainEventBus.java b/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/RabbitDomainEventBus.java
@@ -1,7 +1,9 @@
 package org.reactivecommons.async.rabbit;
 
+import io.cloudevents.CloudEvent;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
 import org.reactivecommons.async.commons.config.BrokerConfig;
+import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
 import org.reactivecommons.api.domain.DomainEvent;
 import org.reactivecommons.api.domain.DomainEventBus;
@@ -31,8 +33,8 @@ public <T> Mono<Void> emit(DomainEvent<T> event) {
     }
 
     @Override
-    public <T> Mono<Void> emit(CloudEvent<T> event) {
-        return emit(new DomainEvent(event.getName(), event.getId(), event))
+    public Publisher<Void> emit(CloudEvent event) {
+        return emit(new DomainEvent<>(event.getType(), event.getId(), event));
     }
 
 }
diff --git a/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/converters/json/CloudEventBuilderExt.java b/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/converters/json/CloudEventBuilderExt.java
@@ -0,0 +1,15 @@
+package org.reactivecommons.async.rabbit.converters.json;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.SneakyThrows;
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
+public class CloudEventBuilderExt {
+    private static final ObjectMapper mapper = new ObjectMapper();
+
+    @SneakyThrows
+    public static byte[] asBytes(Object object) {
+        return mapper.writeValueAsBytes(object);
+    }
+}
diff --git a/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/converters/json/JacksonMessageConverter.java b/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/converters/json/JacksonMessageConverter.java
@@ -1,7 +1,11 @@
 package org.reactivecommons.async.rabbit.converters.json;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.provider.EventFormatProvider;
+import io.cloudevents.jackson.JsonFormat;
 import lombok.Data;
 import org.reactivecommons.api.domain.Command;
 import org.reactivecommons.api.domain.DomainEvent;
@@ -13,6 +17,7 @@
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.Base64;
 
 public class JacksonMessageConverter implements MessageConverter {
     private static final String CONTENT_TYPE = "application/json";
@@ -28,7 +33,7 @@ public JacksonMessageConverter(ObjectMapper objectMapper) {
     public <T> AsyncQuery<T> readAsyncQuery(Message message, Class<T> bodyClass) {
         try {
             final AsyncQueryJson asyncQueryJson = readValue(message, AsyncQueryJson.class);
-            final T value = objectMapper.treeToValue(asyncQueryJson.getQueryData(), bodyClass);
+            T value = extractData(bodyClass, asyncQueryJson.getQueryData());
             return new AsyncQuery<>(asyncQueryJson.getResource(), value);
         } catch (IOException e) {
             throw new MessageConversionException("Failed to convert Message content", e);
@@ -39,7 +44,9 @@ public <T> AsyncQuery<T> readAsyncQuery(Message message, Class<T> bodyClass) {
     public <T> DomainEvent<T> readDomainEvent(Message message, Class<T> bodyClass) {
         try {
             final DomainEventJson domainEventJson = readValue(message, DomainEventJson.class);
-            final T value = objectMapper.treeToValue(domainEventJson.getData(), bodyClass);
+
+            T value = extractData(bodyClass, domainEventJson.getData());
+
             return new DomainEvent<>(domainEventJson.getName(), domainEventJson.getEventId(), value);
         } catch (IOException e) {
             throw new MessageConversionException("Failed to convert Message content", e);
@@ -50,7 +57,7 @@ public <T> DomainEvent<T> readDomainEvent(Message message, Class<T> bodyClass) {
     public <T> Command<T> readCommand(Message message, Class<T> bodyClass) {
         try {
             final CommandJson commandJson = readValue(message, CommandJson.class);
-            final T value = objectMapper.treeToValue(commandJson.getData(), bodyClass);
+            T value = extractData(bodyClass, commandJson.getData());
             return new Command<>(commandJson.getName(), commandJson.getCommandId(), value);
         } catch (IOException e) {
             throw new MessageConversionException("Failed to convert Message content", e);
@@ -60,6 +67,14 @@ public <T> Command<T> readCommand(Message message, Class<T> bodyClass) {
     @Override
     public <T> T readValue(Message message, Class<T> valueClass) {
         try {
+            if(valueClass == CloudEvent.class){
+
+                return (T) EventFormatProvider
+                        .getInstance()
+                        .resolveFormat(JsonFormat.CONTENT_TYPE)
+                        .deserialize(objectMapper.readValue(message.getBody(), byte[].class));
+
+            }
             return objectMapper.readValue(message.getBody(), valueClass);
         } catch (IOException e) {
             throw new MessageConversionException("Failed to convert Message content", e);
@@ -87,10 +102,62 @@ public <T> AsyncQuery<T> readAsyncQueryStructure(Message message) {
         return new AsyncQuery<>(asyncQueryJson.getResource(), (T) asyncQueryJson.getQueryData());
     }
 
+
+    public Message commandToMessage(Command<CloudEvent> object) {
+        byte[] data = EventFormatProvider
+                .getInstance()
+                .resolveFormat(JsonFormat.CONTENT_TYPE)
+                .serialize(object.getData());
+
+        return getRabbitMessage(new Command<>(object.getName(), object.getCommandId(), data));
+    }
+
+
+    public Message eventToMessage(DomainEvent<CloudEvent> object) {
+        byte[] data = EventFormatProvider
+                .getInstance()
+                .resolveFormat(JsonFormat.CONTENT_TYPE)
+                .serialize(object.getData());
+
+        return getRabbitMessage(new DomainEvent<>(object.getName(), object.getEventId(), data));
+    }
+
+
+    public Message queryToMessage(AsyncQuery<CloudEvent> object) {
+        byte[] data = EventFormatProvider
+                .getInstance()
+                .resolveFormat(JsonFormat.CONTENT_TYPE)
+                .serialize(object.getQueryData());
+
+        return getRabbitMessage(new AsyncQuery<>(object.getResource(), data));
+    }
     @Override
     public Message toMessage(Object object) {
+        byte[] bytes;
+        if(object instanceof DomainEvent
+                && ((DomainEvent) object).getData() instanceof CloudEvent){
+
+           return eventToMessage((DomainEvent) object);
+
+        }
+        if(object instanceof Command
+                && ((Command) object).getData() instanceof CloudEvent){
+
+            return commandToMessage((Command) object);
+        }
+        if(object instanceof AsyncQuery
+                && ((AsyncQuery) object).getQueryData() instanceof CloudEvent){
+
+            return queryToMessage((AsyncQuery) object);
+        }
+
+        return getRabbitMessage(object);
+    }
+
+    private RabbitMessage getRabbitMessage(Object object) {
         byte[] bytes;
         try {
+
             String jsonString = this.objectMapper.writeValueAsString(object);
             bytes = jsonString.getBytes(StandardCharsets.UTF_8);
         } catch (IOException e) {
@@ -103,6 +170,23 @@ public Message toMessage(Object object) {
         return new RabbitMessage(bytes, props);
     }
 
+    private <T> T extractData(Class<T> bodyClass, JsonNode node) throws JsonProcessingException {
+        T value;
+        if(bodyClass == CloudEvent.class){
+
+            value = (T) EventFormatProvider
+                    .getInstance()
+                    .resolveFormat(JsonFormat.CONTENT_TYPE)
+                    .deserialize(Base64.getDecoder()
+                            .decode(node.asText()));
+
+        }
+        else{
+            value = objectMapper.treeToValue(node, bodyClass);
+        }
+        return value;
+    }
+
     @Data
     private static class AsyncQueryJson {
         private String resource;
diff --git a/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/listeners/ApplicationQueryListener.java b/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/listeners/ApplicationQueryListener.java
@@ -1,6 +1,9 @@
 package org.reactivecommons.async.rabbit.listeners;
 
 import com.rabbitmq.client.AMQP;
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.provider.EventFormatProvider;
+import io.cloudevents.jackson.JsonFormat;
 import lombok.extern.java.Log;
 import org.reactivecommons.async.api.handlers.registered.RegisteredQueryHandler;
 import org.reactivecommons.async.commons.DiscardNotifier;
@@ -149,8 +152,16 @@ protected Function<Mono<Object>, Mono<Object>> enrichPostProcess(Message msg) {
             final String correlationID = msg.getProperties().getHeaders().get(CORRELATION_ID).toString();
             final HashMap<String, Object> headers = new HashMap<>();
             headers.put(CORRELATION_ID, correlationID);
+            Object response = signal.get();
+            if(response instanceof CloudEvent) {
+                byte[] serialized = EventFormatProvider
+                        .getInstance()
+                        .resolveFormat(JsonFormat.CONTENT_TYPE)
+                        .serialize((CloudEvent) response);
+                return sender.sendNoConfirm(serialized, replyExchange, replyID, headers, false);
+            }
 
-            return sender.sendNoConfirm(signal.get(), replyExchange, replyID, headers, false);
+            return sender.sendNoConfirm(response, replyExchange, replyID, headers, false);
         });
     }
 
diff --git a/domain/domain-events/domain-events-api.gradle b/domain/domain-events/domain-events-api.gradle
@@ -5,4 +5,5 @@ ext {
 
 dependencies {
     api 'org.reactivestreams:reactive-streams:1.0.4'
+    api 'io.cloudevents:cloudevents-api:2.5.0'
 }
diff --git a/domain/domain-events/src/main/java/org/reactivecommons/api/domain/DomainEventBus.java b/domain/domain-events/src/main/java/org/reactivecommons/api/domain/DomainEventBus.java
@@ -1,8 +1,11 @@
 package org.reactivecommons.api.domain;
 
+import io.cloudevents.CloudEvent;
 import org.reactivestreams.Publisher;
 
 public interface DomainEventBus {
     <T> Publisher<Void> emit(DomainEvent<T> event);
-    <T> Publisher<Void> emit(CloudEvent<T> event);
+
+    Publisher<Void> emit(CloudEvent event);
+
 }
diff --git a/samples/async/receiver-responder/async-receiver-sample.gradle b/samples/async/receiver-responder/async-receiver-sample.gradle
@@ -4,4 +4,5 @@ dependencies {
     implementation project(":async-commons-rabbit-starter-eda")
     implementation 'org.springframework.boot:spring-boot-starter'
     runtimeOnly 'org.springframework.boot:spring-boot-devtools'
+    implementation 'io.cloudevents:cloudevents-core:2.5.0'
 }
diff --git a/samples/async/receiver-responder/src/main/java/sample/SampleReceiverApp.java b/samples/async/receiver-responder/src/main/java/sample/SampleReceiverApp.java
diff --git a/samples/async/sender-client/async-sender-client.gradle b/samples/async/sender-client/async-sender-client.gradle
diff --git a/samples/async/sender-client/src/main/java/sample/SampleRestController.java b/samples/async/sender-client/src/main/java/sample/SampleRestController.java

Original file line number	Diff line number	Diff line change
`@@ -7,4 +7,6 @@ dependencies {`
`7`	`7`	`api project(':domain-events-api')`
`8`	`8`	`compileOnly 'io.projectreactor:reactor-core'`
`9`	`9`	`testImplementation 'io.projectreactor:reactor-test'`
	`10`	`+ implementation 'io.cloudevents:cloudevents-json-jackson:2.5.0'`
	`11`	`+`
`10`	`12`	`}`
Original file line number	Diff line number	Diff line change
`@@ -14,4 +14,5 @@ dependencies {`
`14`	`14`	`api 'com.rabbitmq:amqp-client'`
`15`	`15`	`api 'com.fasterxml.jackson.core:jackson-databind'`
`16`	`16`	`testImplementation 'io.projectreactor:reactor-test'`
	`17`	`+ implementation 'io.cloudevents:cloudevents-json-jackson:2.5.0'`
`17`	`18`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,9 @@`
`1`	`1`	`package org.reactivecommons.async.rabbit;`
`2`	`2`
	`3`	`+import io.cloudevents.CloudEvent;`
`3`	`4`	`import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;`
`4`	`5`	`import org.reactivecommons.async.commons.config.BrokerConfig;`
	`6`	`+import org.reactivestreams.Publisher;`
`5`	`7`	`import reactor.core.publisher.Mono;`
`6`	`8`	`import org.reactivecommons.api.domain.DomainEvent;`
`7`	`9`	`import org.reactivecommons.api.domain.DomainEventBus;`
`@@ -31,8 +33,8 @@ public <T> Mono<Void> emit(DomainEvent<T> event) {`
`31`	`33`	`}`
`32`	`34`
`33`	`35`	`@Override`
`34`		`- public <T> Mono<Void> emit(CloudEvent<T> event) {`
`35`		`- return emit(new DomainEvent(event.getName(), event.getId(), event))`
	`36`	`+ public Publisher<Void> emit(CloudEvent event) {`
	`37`	`+ return emit(new DomainEvent<>(event.getType(), event.getId(), event));`
`36`	`38`	`}`
`37`	`39`
`38`	`40`	`}`