Turn PayloadMapper and MessageMapper into interfaces

deployment/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/deployment/ReactiveMesssagingNatsJetstreamProcessor.java

@@ -8,8 +8,8 @@
 import io.quarkiverse.reactive.messaging.nats.jetstream.JetStreamRecorder;
 import io.quarkiverse.reactive.messaging.nats.jetstream.client.ConnectionFactory;
 import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.ConsumerMapperImpl;
-import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.MessageMapper;
-import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.PayloadMapper;
+import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.DefaultMessageMapper;
+import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.DefaultPayloadMapper;
 import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.StreamStateMapperImpl;
 import io.quarkiverse.reactive.messaging.nats.jetstream.tracing.JetStreamInstrumenter;
 import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
@@ -54,8 +54,8 @@ void createNatsConnector(BuildProducer<AdditionalBeanBuildItem> buildProducer) {
         buildProducer.produce(AdditionalBeanBuildItem.unremovableOf(JetStreamInstrumenter.class));
         buildProducer.produce(AdditionalBeanBuildItem.unremovableOf(ExecutionHolder.class));
         buildProducer.produce(AdditionalBeanBuildItem.unremovableOf(ConnectionFactory.class));
-        buildProducer.produce(AdditionalBeanBuildItem.unremovableOf(PayloadMapper.class));
-        buildProducer.produce(AdditionalBeanBuildItem.unremovableOf(MessageMapper.class));
+        buildProducer.produce(AdditionalBeanBuildItem.unremovableOf(DefaultPayloadMapper.class));
+        buildProducer.produce(AdditionalBeanBuildItem.unremovableOf(DefaultMessageMapper.class));
         buildProducer.produce(AdditionalBeanBuildItem.unremovableOf(ConsumerMapperImpl.class));
         buildProducer.produce(AdditionalBeanBuildItem.unremovableOf(StreamStateMapperImpl.class));
     }

integration-tests/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/it/Data.java

@@ -1,14 +1,22 @@
 package io.quarkiverse.reactive.messaging.nats.jetstream.it;
 
+import java.time.Instant;
+
+import com.fasterxml.jackson.annotation.JsonView;
+
 import io.quarkus.runtime.annotations.RegisterForReflection;
 
 @RegisterForReflection
 public class Data {
+
     private String data;
     private String resourceId;
 
     private String messageId;
 
+    @JsonView(IncludeTimestamps.class)
+    private Instant creationTime;
+
     public Data(String data, String resourceId) {
         this(data, resourceId, null);
     }
@@ -22,6 +30,13 @@ public Data(String data, String resourceId, String messageId) {
     public Data() {
     }
 
+    public Data(String data, String resourceId, String messageId, Instant creationTime) {
+        this.data = data;
+        this.resourceId = resourceId;
+        this.messageId = messageId;
+        this.creationTime = creationTime;
+    }
+
     public String getData() {
         return data;
     }
@@ -45,4 +60,12 @@ public String getMessageId() {
     public void setMessageId(String messageId) {
         this.messageId = messageId;
     }
+
+    public Instant getCreationTime() {
+        return creationTime;
+    }
+
+    public void setCreationTime(Instant creationTime) {
+        this.creationTime = creationTime;
+    }
 }

integration-tests/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/it/DataConsumingBean.java

@@ -35,7 +35,7 @@ private void handleData(Message<Data> message) {
         message.getMetadata(JetStreamIncomingMessageMetadata.class)
                 .ifPresent(metadata -> lastData = Optional.of(
                         new Data(message.getPayload().getData(), metadata.headers().get("RESOURCE_ID").get(0),
-                                metadata.messageId())));
+                                metadata.messageId(), message.getPayload().getCreationTime())));
         message.ack();
     }
 }

integration-tests/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/it/DataResource.java

@@ -28,6 +28,8 @@
 import org.eclipse.microprofile.reactive.messaging.Message;
 import org.eclipse.microprofile.reactive.messaging.Metadata;
 
+import com.fasterxml.jackson.annotation.JsonView;
+
 import io.quarkiverse.reactive.messaging.nats.jetstream.JetStreamOutgoingMessageMetadata;
 import io.smallrye.mutiny.Uni;
 
@@ -47,6 +49,13 @@ public Data getLast() {
         return bean.getLast().orElseGet(Data::new);
     }
 
+    @GET
+    @Path("/last-with-timestamp")
+    @JsonView(IncludeTimestamps.class)
+    public Data getLastWithTimestamp() {
+        return bean.getLast().orElseGet(Data::new);
+    }
+
     @POST
     @Consumes("application/json")
     @Path("/{messageId}")

integration-tests/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/it/IncludeTimestamps.java

@@ -0,0 +1,4 @@
+package io.quarkiverse.reactive.messaging.nats.jetstream.it;
+
+public final class IncludeTimestamps {
+}

integration-tests/src/test/java/io/quarkiverse/reactive/messaging/nats/jetstream/it/DataResourceCustomMappersTest.java

@@ -0,0 +1,81 @@
+package io.quarkiverse.reactive.messaging.nats.jetstream.it;
+
+import static io.restassured.RestAssured.get;
+import static io.restassured.RestAssured.given;
+import static org.awaitility.Awaitility.await;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.concurrent.TimeUnit;
+
+import jakarta.enterprise.context.ApplicationScoped;
+
+import org.junit.jupiter.api.Test;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import io.quarkiverse.reactive.messaging.nats.jetstream.it.DataResourceCustomMappersTest.Profile;
+import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.DefaultPayloadMapper;
+import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.PayloadMapper;
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.QuarkusTestProfile;
+import io.quarkus.test.junit.TestProfile;
+
+@QuarkusTest
+@TestProfile(Profile.class)
+public class DataResourceCustomMappersTest {
+
+    @Test
+    public void data() {
+        final var messageId = "8cb9fd88-08e9-422d-9f19-a3b4b3cc8cb7";
+        final var data = "N6cXzM";
+        final var resourceId = "56d5cc43-92dd-4df9-b385-1e412fd8fc8a";
+        final var now = Instant.now();
+
+        given()
+                .header("Content-Type", "application/json")
+                .pathParam("messageId", messageId)
+                .body(new Data(data, resourceId, null, now))
+                .post("/data/{messageId}")
+                .then().statusCode(204);
+
+        await().atMost(1, TimeUnit.MINUTES).until(() -> {
+            final var dataValue = get("/data/last-with-timestamp").as(Data.class);
+            return messageId.equals(dataValue.getMessageId()) && data.equals(dataValue.getData())
+                    && resourceId.equals(dataValue.getResourceId()) && dataValue.getCreationTime().equals(now);
+        });
+    }
+
+    public static class Profile implements QuarkusTestProfile {
+
+        @ApplicationScoped
+        public PayloadMapper getPayloadMapper(ObjectMapper objectMapper) {
+            return new DefaultPayloadMapper(objectMapper) {
+                @Override
+                public <T> T of(byte[] data, Class<T> type) {
+                    try {
+                        return objectMapper.readerWithView(IncludeTimestamps.class).readValue(data, type);
+                    } catch (IOException e) {
+                        throw new RuntimeException(e);
+                    }
+                }
+
+                @Override
+                public byte[] of(Object payload) {
+                    try {
+                        if (payload == null) {
+                            return new byte[0];
+                        } else if (payload instanceof byte[]) {
+                            return (byte[]) payload;
+                        } else {
+                            return objectMapper.writerWithView(IncludeTimestamps.class).writeValueAsBytes(payload);
+                        }
+                    } catch (JsonProcessingException e) {
+                        throw new RuntimeException(e);
+                    }
+                }
+            };
+        }
+    }
+}

runtime/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/client/DefaultConnection.java

@@ -1,7 +1,7 @@
 package io.quarkiverse.reactive.messaging.nats.jetstream.client;
 
 import static io.nats.client.Connection.Status.CONNECTED;
-import static io.quarkiverse.reactive.messaging.nats.jetstream.mapper.MessageMapper.MESSAGE_TYPE_HEADER;
+import static io.quarkiverse.reactive.messaging.nats.jetstream.mapper.DefaultMessageMapper.MESSAGE_TYPE_HEADER;
 import static io.smallrye.reactive.messaging.tracing.TracingUtils.traceOutgoing;
 
 import java.io.IOException;

runtime/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/client/ReaderSubscribeConnection.java

@@ -202,13 +202,14 @@ private Optional<io.nats.client.Message> nextMessage() {
         }
     }
 
+    @SuppressWarnings("unchecked")
     private Multi<org.eclipse.microprofile.reactive.messaging.Message<K>> createMulti(io.nats.client.Message message,
             boolean tracingEnabled, Class<?> payloadType, Context context) {
         if (message == null || message.getData() == null) {
             return Multi.createFrom().empty();
         } else {
             return Multi.createFrom()
-                    .item(() -> delegate.messageMapper().of(message, tracingEnabled, payloadType, context,
+                    .item(() -> delegate.messageMapper().of(message, tracingEnabled, (Class<K>) payloadType, context,
                             new ExponentialBackoff(
                                     consumerConfiguration.consumerConfiguration().exponentialBackoff(),
                                     consumerConfiguration.consumerConfiguration().exponentialBackoffMaxDuration()),

runtime/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/client/configuration/DefaultConnectionConfiguration.java

@@ -5,7 +5,7 @@
 
 import io.nats.client.ErrorListener;
 import io.quarkiverse.reactive.messaging.nats.NatsConfiguration;
-import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.PayloadMapper;
+import io.quarkiverse.reactive.messaging.nats.jetstream.mapper.DefaultPayloadMapper;
 
 class DefaultConnectionConfiguration implements ConnectionConfiguration {
     private final NatsConfiguration configuration;
@@ -86,7 +86,7 @@ public Optional<String> getTlsAlgorithm() {
 
     private ErrorListener getInstanceOfErrorListener(String className) {
         try {
-            var clazz = PayloadMapper.loadClass(className);
+            var clazz = DefaultPayloadMapper.loadClass(className);
             var constructor = clazz.getConstructor();
             return (ErrorListener) constructor.newInstance();
         } catch (NoSuchMethodException | InstantiationException | IllegalAccessException

runtime/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/mapper/DefaultMessageMapper.java

@@ -0,0 +1,63 @@
+package io.quarkiverse.reactive.messaging.nats.jetstream.mapper;
+
+import static io.smallrye.reactive.messaging.tracing.TracingUtils.traceIncoming;
+
+import java.time.Duration;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+
+import org.eclipse.microprofile.reactive.messaging.Message;
+
+import io.quarkiverse.reactive.messaging.nats.jetstream.ExponentialBackoff;
+import io.quarkiverse.reactive.messaging.nats.jetstream.JetStreamIncomingMessage;
+import io.quarkiverse.reactive.messaging.nats.jetstream.tracing.JetStreamInstrumenter;
+import io.quarkiverse.reactive.messaging.nats.jetstream.tracing.JetStreamTrace;
+import io.quarkus.arc.DefaultBean;
+import io.vertx.mutiny.core.Context;
+
+@ApplicationScoped
+@DefaultBean
+public class DefaultMessageMapper implements MessageMapper {
+
+    public static final String MESSAGE_TYPE_HEADER = "message.type";
+
+    private final PayloadMapper payloadMapper;
+    private final JetStreamInstrumenter instrumenter;
+
+    @Inject
+    public DefaultMessageMapper(PayloadMapper payloadMapper,
+            JetStreamInstrumenter instrumenter) {
+        this.payloadMapper = payloadMapper;
+        this.instrumenter = instrumenter;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public <T> org.eclipse.microprofile.reactive.messaging.Message<T> of(
+            io.nats.client.Message message,
+            boolean tracingEnabled,
+            Class<T> payloadType,
+            Context context,
+            ExponentialBackoff exponentialBackoff,
+            Duration ackTimeout) {
+        try {
+            final var incomingMessage = payloadType != null
+                    ? new JetStreamIncomingMessage<>(message, payloadMapper.of(message, payloadType), context,
+                            exponentialBackoff, ackTimeout)
+                    : new JetStreamIncomingMessage<>(message,
+                            payloadMapper.of(message).orElseGet(() -> message.getData()),
+                            context,
+                            exponentialBackoff, ackTimeout);
+            if (tracingEnabled) {
+                return (Message<T>) traceIncoming(instrumenter.receiver(), incomingMessage,
+                        JetStreamTrace.trace(incomingMessage));
+            } else {
+                return (Message<T>) incomingMessage;
+            }
+        } catch (ClassCastException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+}

runtime/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/mapper/DefaultPayloadMapper.java

@@ -0,0 +1,83 @@
+package io.quarkiverse.reactive.messaging.nats.jetstream.mapper;
+
+import static io.quarkiverse.reactive.messaging.nats.jetstream.mapper.DefaultMessageMapper.MESSAGE_TYPE_HEADER;
+
+import java.io.IOException;
+import java.util.Optional;
+
+import jakarta.enterprise.context.ApplicationScoped;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import io.nats.client.api.MessageInfo;
+import io.quarkus.arc.DefaultBean;
+
+@ApplicationScoped
+@DefaultBean
+public class DefaultPayloadMapper implements PayloadMapper {
+
+    private final ObjectMapper objectMapper;
+
+    public DefaultPayloadMapper(ObjectMapper objectMapper) {
+        this.objectMapper = objectMapper;
+    }
+
+    /**
+     * Returns a byte array containing the supplied payload.
+     *
+     * @param payload the payload
+     * @return a byte array encapsulation of the payload
+     */
+    @Override
+    public byte[] of(final Object payload) {
+        try {
+            if (payload == null) {
+                return new byte[0];
+            } else if (payload instanceof byte[]) {
+                return (byte[]) payload;
+            } else {
+                return objectMapper.writeValueAsBytes(payload);
+            }
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public <T> T of(byte[] data, Class<T> type) {
+        try {
+            return objectMapper.readValue(data, type);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public <T> Optional<T> of(io.nats.client.Message message) {
+        return Optional.ofNullable(message).flatMap(m -> Optional.ofNullable(m.getHeaders()))
+                .flatMap(headers -> Optional.ofNullable(headers.getFirst(MESSAGE_TYPE_HEADER)))
+                .map(DefaultPayloadMapper::loadClass)
+                .map(type -> (T) of(message.getData(), type));
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public <T> Optional<T> of(MessageInfo message) {
+        return Optional.ofNullable(message).flatMap(m -> Optional.ofNullable(m.getHeaders()))
+                .flatMap(headers -> Optional.ofNullable(headers.getFirst(MESSAGE_TYPE_HEADER)))
+                .map(DefaultPayloadMapper::loadClass)
+                .map(type -> (T) of(message.getData(), type));
+    }
+
+    @SuppressWarnings("unchecked")
+    public static <T> Class<T> loadClass(String type) {
+        try {
+            final ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+            return (Class<T>) classLoader.loadClass(type);
+        } catch (ClassNotFoundException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}