feat(rawhandler): Allow raw message handler to listening events

juancgalvis · juancgalvis · commit 8fceb15157dc · 2025-02-24T15:17:55.000-05:00
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
@@ -4,12 +4,14 @@
 import lombok.AccessLevel;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
+import org.reactivecommons.api.domain.RawMessage;
 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.RawEventHandler;
 import org.reactivecommons.async.api.handlers.registered.RegisteredCommandHandler;
 import org.reactivecommons.async.api.handlers.registered.RegisteredEventListener;
 import org.reactivecommons.async.api.handlers.registered.RegisteredQueryHandler;
@@ -37,7 +39,8 @@ public static HandlerRegistry register() {
         return instance;
     }
 
-    public <T> HandlerRegistry listenDomainEvent(String domain, String eventName, DomainEventHandler<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));
         return this;
@@ -49,6 +52,12 @@ public HandlerRegistry listenDomainCloudEvent(String domain, String eventName, C
         return this;
     }
 
+    public HandlerRegistry listenDomainRawEvent(String domain, String eventName, RawEventHandler handler) {
+        domainEventListeners.computeIfAbsent(domain, ignored -> new CopyOnWriteArrayList<>())
+                .add(new RegisteredEventListener<>(eventName, handler, RawMessage.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));
@@ -61,7 +70,8 @@ public HandlerRegistry listenCloudEvent(String eventName, CloudEventHandler hand
         return this;
     }
 
-    public <T> HandlerRegistry listenNotificationEvent(String eventName, DomainEventHandler<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;
     }
@@ -71,7 +81,8 @@ public HandlerRegistry listenNotificationCloudEvent(String eventName, CloudEvent
         return this;
     }
 
-    public <T> HandlerRegistry handleDynamicEvents(String eventNamePattern, DomainEventHandler<T> handler, Class<T> eventClass) {
+    public <T> HandlerRegistry handleDynamicEvents(String eventNamePattern, DomainEventHandler<T> handler,
+                                                   Class<T> eventClass) {
         dynamicEventHandlers.add(new RegisteredEventListener<>(eventNamePattern, handler, eventClass));
         return this;
     }
@@ -102,7 +113,8 @@ public <R> HandlerRegistry serveQuery(String resource, QueryHandlerDelegate<Void
     }
 
     public <R> HandlerRegistry serveCloudEventQuery(String resource, QueryHandler<R, CloudEvent> handler) {
-        handlers.add(new RegisteredQueryHandler<>(resource, (ignored, message) -> handler.handle(message), CloudEvent.class));
+        handlers.add(new RegisteredQueryHandler<>(resource, (ignored, message) -> handler.handle(message),
+                CloudEvent.class));
         return this;
     }
 
diff --git a/async/async-commons-api/src/main/java/org/reactivecommons/async/api/handlers/RawEventHandler.java b/async/async-commons-api/src/main/java/org/reactivecommons/async/api/handlers/RawEventHandler.java
@@ -0,0 +1,6 @@
+package org.reactivecommons.async.api.handlers;
+
+import org.reactivecommons.api.domain.RawMessage;
+
+public interface RawEventHandler<T extends RawMessage> extends EventHandler<T> {
+}
diff --git a/async/async-commons/src/main/java/org/reactivecommons/async/commons/communications/Message.java b/async/async-commons/src/main/java/org/reactivecommons/async/commons/communications/Message.java
@@ -1,13 +1,15 @@
 package org.reactivecommons.async.commons.communications;
 
+import org.reactivecommons.api.domain.RawMessage;
+
 import java.util.Map;
 
 /**
  * Simple Internal Message representation
  *
  * @author Daniel Bustamante Ospina
  */
-public interface Message {
+public interface Message extends RawMessage {
 
     byte[] getBody();
 
diff --git a/async/async-kafka/src/main/java/org/reactivecommons/async/kafka/KafkaDomainEventBus.java b/async/async-kafka/src/main/java/org/reactivecommons/async/kafka/KafkaDomainEventBus.java
@@ -4,6 +4,7 @@
 import lombok.AllArgsConstructor;
 import org.reactivecommons.api.domain.DomainEvent;
 import org.reactivecommons.api.domain.DomainEventBus;
+import org.reactivecommons.api.domain.RawMessage;
 import org.reactivecommons.async.kafka.communications.ReactiveMessageSender;
 import org.reactivestreams.Publisher;
 
@@ -31,4 +32,14 @@ public Publisher<Void> emit(CloudEvent event) {
     public Publisher<Void> emit(String domain, CloudEvent event) {
         throw new UnsupportedOperationException(NOT_IMPLEMENTED_YET);
     }
+
+    @Override
+    public Publisher<Void> emit(RawMessage event) {
+        return sender.send(event);
+    }
+
+    @Override
+    public Publisher<Void> emit(String domain, RawMessage event) {
+        throw new UnsupportedOperationException(NOT_IMPLEMENTED_YET);
+    }
 }
diff --git a/async/async-kafka/src/main/java/org/reactivecommons/async/kafka/KafkaMessage.java b/async/async-kafka/src/main/java/org/reactivecommons/async/kafka/KafkaMessage.java
@@ -15,6 +15,7 @@
 public class KafkaMessage implements Message {
     private final byte[] body;
     private final Properties properties;
+    private final String type;
 
     @Data
     public static class KafkaMessageProperties implements Properties {
@@ -30,7 +31,11 @@ public String getContentType() {
     }
 
     public static KafkaMessage fromDelivery(ReceiverRecord<String, byte[]> record) {
-        return new KafkaMessage(record.value(), createMessageProps(record));
+        return fromDelivery(record, null);
+    }
+
+    public static KafkaMessage fromDelivery(ReceiverRecord<String, byte[]> record, String type) {
+        return new KafkaMessage(record.value(), createMessageProps(record), type);
     }
 
     private static Properties createMessageProps(ReceiverRecord<String, byte[]> record) {
diff --git a/async/async-kafka/src/main/java/org/reactivecommons/async/kafka/converters/json/KafkaJacksonMessageConverter.java b/async/async-kafka/src/main/java/org/reactivecommons/async/kafka/converters/json/KafkaJacksonMessageConverter.java
@@ -22,6 +22,9 @@ public KafkaJacksonMessageConverter(ObjectMapper objectMapper) {
 
     @Override
     public Message toMessage(Object object) {
+        if (object instanceof KafkaMessage) {
+            return (KafkaMessage) object;
+        }
         byte[] bytes;
         try {
             String jsonString = this.objectMapper.writeValueAsString(object);
@@ -30,7 +33,7 @@ public Message toMessage(Object object) {
             throw new MessageConversionException(FAILED_TO_CONVERT_MESSAGE_CONTENT, e);
         }
         KafkaMessageProperties props = buildProperties(object);
-        return new KafkaMessage(bytes, props);
+        return new KafkaMessage(bytes, props, null);
     }
 
     private KafkaMessageProperties buildProperties(Object message) {
diff --git a/async/async-kafka/src/main/java/org/reactivecommons/async/kafka/listeners/GenericMessageListener.java b/async/async-kafka/src/main/java/org/reactivecommons/async/kafka/listeners/GenericMessageListener.java
@@ -112,7 +112,7 @@ protected Mono<ReceiverRecord<String, byte[]>> handle(ReceiverRecord<String, byt
         try {
             final String executorPath = getExecutorPath(msj);
             final Function<Message, Mono<Object>> handler = getExecutor(executorPath);
-            final Message message = KafkaMessage.fromDelivery(msj);
+            final Message message = KafkaMessage.fromDelivery(msj, executorPath);
 
             Mono<Object> flow = Mono.defer(() -> handler.apply(message))
                     .transform(enrichPostProcess(message));
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
@@ -3,6 +3,7 @@
 import io.cloudevents.CloudEvent;
 import org.reactivecommons.api.domain.DomainEvent;
 import org.reactivecommons.api.domain.DomainEventBus;
+import org.reactivecommons.api.domain.RawMessage;
 import org.reactivecommons.async.commons.config.BrokerConfig;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
 import org.reactivestreams.Publisher;
@@ -49,4 +50,15 @@ public Publisher<Void> emit(String domain, CloudEvent event) {
         throw new UnsupportedOperationException("Not implemented yet");
     }
 
+    @Override
+    public Publisher<Void> emit(RawMessage rawEvent) {
+        return sender.sendWithConfirm(rawEvent, exchange, rawEvent.getType(),
+                        Collections.emptyMap(), persistentEvents)
+                .onErrorMap(err -> new RuntimeException("Event send failure: " + rawEvent.getType(), err));
+    }
+
+    @Override
+    public Publisher<Void> emit(String domain, RawMessage event) {
+        throw new UnsupportedOperationException("Not implemented yet");
+    }
 }
diff --git a/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/RabbitMessage.java b/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/RabbitMessage.java
@@ -11,6 +11,7 @@
 public class RabbitMessage implements Message {
     private final byte[] body;
     private final Properties properties;
+    private final String type;
 
     @Data
     public static class RabbitMessageProperties implements Properties {
@@ -22,7 +23,11 @@ public static class RabbitMessageProperties implements Properties {
     }
 
     public static RabbitMessage fromDelivery(Delivery delivery) {
-        return new RabbitMessage(delivery.getBody(), createMessageProps(delivery));
+        return fromDelivery(delivery, null);
+    }
+
+    public static RabbitMessage fromDelivery(Delivery delivery, String executorPath) {
+        return new RabbitMessage(delivery.getBody(), createMessageProps(delivery), executorPath);
     }
 
     private static Message.Properties createMessageProps(Delivery msj) {
diff --git a/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/converters/json/RabbitJacksonMessageConverter.java b/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/converters/json/RabbitJacksonMessageConverter.java
@@ -18,6 +18,9 @@ public RabbitJacksonMessageConverter(ObjectMapper objectMapper) {
 
     @Override
     public Message toMessage(Object object) {
+        if (object instanceof RabbitMessage) {
+            return (RabbitMessage) object;
+        }
         byte[] bytes;
         try {
             String jsonString = this.objectMapper.writeValueAsString(object);
@@ -29,10 +32,10 @@ public Message toMessage(Object object) {
         if (object instanceof CloudEvent) {
             props.setContentType(APPLICATION_CLOUD_EVENT_JSON);
         } else {
-            props.setContentType(CONTENT_TYPE);
+            props.setContentType(APPLICATION_JSON);
         }
         props.setContentEncoding(StandardCharsets.UTF_8.name());
         props.setContentLength(bytes.length);
-        return new RabbitMessage(bytes, props);
+        return new RabbitMessage(bytes, props, null);
     }
 }
diff --git a/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/listeners/ApplicationEventListener.java b/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/listeners/ApplicationEventListener.java
@@ -4,6 +4,7 @@
 import lombok.extern.java.Log;
 import org.reactivecommons.async.api.handlers.CloudEventHandler;
 import org.reactivecommons.async.api.handlers.DomainEventHandler;
+import org.reactivecommons.async.api.handlers.RawEventHandler;
 import org.reactivecommons.async.api.handlers.registered.RegisteredEventListener;
 import org.reactivecommons.async.commons.DiscardNotifier;
 import org.reactivecommons.async.commons.EventExecutor;
@@ -109,6 +110,9 @@ private <T, D> Function<Message, Object> resolveConverter(RegisteredEventListene
         if (registeredEventListener.getHandler() instanceof CloudEventHandler) {
             return messageConverter::readCloudEvent;
         }
+        if (registeredEventListener.getHandler() instanceof RawEventHandler) {
+            return message -> message;
+        }
         throw new RuntimeException("Unknown handler type");
     }
 }
diff --git a/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/listeners/GenericMessageListener.java b/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/listeners/GenericMessageListener.java
@@ -116,7 +116,7 @@ protected Mono<AcknowledgableDelivery> handle(AcknowledgableDelivery msj, Instan
         try {
             final String executorPath = getExecutorPath(msj);
             final Function<Message, Mono<Object>> handler = getExecutor(executorPath);
-            final Message message = RabbitMessage.fromDelivery(msj);
+            final Message message = RabbitMessage.fromDelivery(msj, executorPath);
 
             Mono<Object> flow = defer(() -> handler.apply(message))
                     .transform(enrichPostProcess(message));
diff --git a/async/async-rabbit/src/test/java/org/reactivecommons/async/helpers/TestStubs.java b/async/async-rabbit/src/test/java/org/reactivecommons/async/helpers/TestStubs.java
@@ -15,6 +15,6 @@ public static Message mockMessage() {
         properties.getHeaders().put(CORRELATION_ID, "correlation");
         properties.getHeaders().put(SERVED_QUERY_ID, "my-query");
         return new RabbitMessage("{\"id\":\"id\",\"name\":\"name\",\"date\":\"2020-10-22T17:03:26.062Z\"}".getBytes()
-                , properties);
+                , properties, null);
     }
 }
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
@@ -9,4 +9,7 @@ public interface DomainEventBus {
 
     Publisher<Void> emit(CloudEvent event);
     Publisher<Void> emit(String domain, CloudEvent event);
+
+    Publisher<Void> emit(RawMessage event);
+    Publisher<Void> emit(String domain, RawMessage event);
 }
diff --git a/domain/domain-events/src/main/java/org/reactivecommons/api/domain/RawMessage.java b/domain/domain-events/src/main/java/org/reactivecommons/api/domain/RawMessage.java
@@ -0,0 +1,5 @@
+package org.reactivecommons.api.domain;
+
+public interface RawMessage {
+    String getType();
+}
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/senders/GenericDomainEventBus.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/senders/GenericDomainEventBus.java
@@ -4,6 +4,7 @@
 import lombok.RequiredArgsConstructor;
 import org.reactivecommons.api.domain.DomainEvent;
 import org.reactivecommons.api.domain.DomainEventBus;
+import org.reactivecommons.api.domain.RawMessage;
 import org.reactivecommons.async.starter.exceptions.InvalidConfigurationException;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
@@ -44,4 +45,18 @@ public Publisher<Void> emit(String domain, CloudEvent event) {
         }
         return domainEventBus.emit(event);
     }
+
+    @Override
+    public Publisher<Void> emit(RawMessage event) {
+        return emit(DEFAULT_DOMAIN, event);
+    }
+
+    @Override
+    public Publisher<Void> emit(String domain, RawMessage event) {
+        DomainEventBus domainEventBus = domainEventBuses.get(domain);
+        if (domainEventBus == null) {
+            return Mono.error(() -> new InvalidConfigurationException("Domain not found: " + domain));
+        }
+        return domainEventBus.emit(event);
+    }
 }
