[DEVEX-250] Added the initial implementation of Jackson Serializer

Author: oskardudycz

db-client-java/build.gradle

@@ -46,10 +46,11 @@ dependencies {
     implementation "io.grpc:grpc-stub:${grpcVersion}"
     implementation "io.grpc:grpc-protobuf:${grpcVersion}"
     implementation "com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}"
+    implementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310'
     implementation "org.slf4j:slf4j-api:2.0.17"
     implementation "org.bouncycastle:bcprov-jdk18on:1.80"
     implementation "org.bouncycastle:bcpkix-jdk18on:1.80"
-
+    
     implementation platform("io.opentelemetry:opentelemetry-bom:${openTelemetryVersion}")
     implementation "io.opentelemetry:opentelemetry-api"
     implementation "io.opentelemetry.semconv:opentelemetry-semconv:${openTelemetrySemConvVersion}"
@@ -64,7 +65,6 @@ dependencies {
     testImplementation "org.reactivestreams:reactive-streams-tck:${reactiveStreamsApiVersion}"
     testImplementation "org.testcontainers:testcontainers:${testcontainersVersion}"
     testImplementation platform("com.fasterxml.jackson:jackson-bom:${jacksonVersion}")
-    testImplementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310'
     testImplementation "com.github.javafaker:javafaker:1.0.2"
     testImplementation 'org.slf4j:slf4j-simple:2.0.17'
     testImplementation "io.opentelemetry:opentelemetry-sdk"

db-client-java/src/main/java/io/kurrent/dbclient/MessageData.java

@@ -12,15 +12,15 @@ public final class MessageData {
     private final byte[] messageData;
     private final byte[] messageMetadata;
 
-    MessageData(String messageType, byte[] messageData) {
+    public MessageData(String messageType, byte[] messageData) {
         this(messageType, messageData, null, UUID.randomUUID(), ContentType.JSON);
     }
 
-    MessageData(String messageType, byte[] messageData, byte[] userMetadata) {
+    public MessageData(String messageType, byte[] messageData, byte[] userMetadata) {
         this(messageType, messageData, userMetadata, UUID.randomUUID(), ContentType.JSON);
     }
-    
-    MessageData(String messageType, byte[] messageData, byte[] userMetadata, UUID messageId, String contentType) {
+
+    public MessageData(String messageType, byte[] messageData, byte[] userMetadata, UUID messageId, String contentType) {
         this.messageId = messageId;
         this.messageType = messageType;
         this.contentType = contentType;

db-client-java/src/main/java/io/kurrent/dbclient/serialization/JacksonSerializer.java

@@ -0,0 +1,56 @@
+package io.kurrent.dbclient.serialization;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.*;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Optional;
+
+public class JacksonSerializer implements Serializer {
+    public static final JsonMapper.Builder defaultBuilder = JsonMapper.builder()
+            .addModule(new JavaTimeModule())
+            .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+            .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false)
+            .propertyNamingStrategy(PropertyNamingStrategies.LOWER_CAMEL_CASE)
+            .enable(MapperFeature.ACCEPT_CASE_INSENSITIVE_PROPERTIES)
+            .serializationInclusion(JsonInclude.Include.NON_NULL)
+            .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
+    
+    private static final Logger logger = LoggerFactory.getLogger(JacksonSerializer.class);
+    
+    private final JsonMapper jsonMapper;
+
+    public JacksonSerializer(){
+        this(defaultBuilder);
+    }
+    
+    public JacksonSerializer(JsonMapper.Builder builder){
+        jsonMapper = builder != null ? builder.build() : defaultBuilder.build();
+    }
+    
+    @Override
+    public byte[] serialize(Object value) {
+        try {
+            return jsonMapper.writeValueAsBytes(value);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public <MessageType> Optional<MessageType> deserialize(Class<MessageType> eventClass, byte[] data) {
+        try {
+            MessageType result = jsonMapper.readValue(data, eventClass);
+
+            return Optional.ofNullable(result);
+        } catch (IOException e) {
+            logger.warn("Error deserializing event {}", eventClass.getName(), e);
+            return Optional.empty();
+        }
+    }
+}

db-client-java/src/main/java/io/kurrent/dbclient/serialization/MessageSerializerImpl.java

@@ -7,14 +7,18 @@
 import java.util.stream.Collectors;
 
 class MessageSerializerImpl implements MessageSerializer {
+    Serializer serializer = new JacksonSerializer();
+    
     @Override
     public MessageSerializer with(OperationSerializationSettings serializationSettings) {
-        return null;
+        return this;
     }
 
     @Override
     public MessageData serialize(Message value, MessageSerializationContext context) {
-        return null;
+        return MessageData
+                .builderAsJson(value.getData().getClass().getTypeName(), serializer.serialize(value))
+                .build();
     }
 
     @Override

db-client-java/src/main/java/io/kurrent/dbclient/serialization/Serializer.java

@@ -0,0 +1,32 @@
+package io.kurrent.dbclient.serialization;
+
+import java.util.Optional;
+
+/// <summary>
+/// Defines the core serialization capabilities required by the KurrentDB client.
+/// Implementations of this interface handle the conversion between Java objects and their
+/// binary representation for storage in and retrieval from the event store.
+/// <br />
+/// The client ships default Jackson implementation, but custom implementations can be provided or other formats.
+/// </summary>
+public interface Serializer {
+    /// <summary>
+    /// Converts a Java object to its binary representation for storage in the event store.
+    /// </summary>
+    /// <param name="value">The object to serialize. This could be an event, command, or metadata object.</param>
+    /// <returns>
+    /// A binary representation of the object that can be stored in KurrentDB.
+    /// </returns>
+    byte[] serialize(Object value);
+
+    /// <summary>
+    /// Reconstructs a Java object from its binary representation retrieved from the event store.
+    /// </summary>
+    /// <param name="data">The binary data to deserialize, typically retrieved from a KurrentDB event.</param>
+    /// <param name="type">The target Java type to deserialize the data into, determined from message type mappings.</param>
+    /// <returns>
+    /// The deserialized object cast to the specified type, or null if the data cannot be deserialized.
+    /// The returned object will be an instance of the specified type or a compatible subtype.
+    /// </returns>
+    <MessageType> Optional<MessageType> deserialize(Class<MessageType> eventClass, byte[] data);
+}

db-client-java/src/test/java/io/kurrent/dbclient/streams/serialization/SerializationTests.java

@@ -16,25 +16,19 @@ default void testPlainJavaObjectsAreSerializedAndDeserializedUsingAutoSerializat
 
         // Given
         final String streamName = generateName();
-        final List<UserRegistered> expected = generateMessages(2);
+        final Object[] expected = generateMessages(2).toArray();
 
-
         // When
         AppendToStreamOptions appendOptions = AppendToStreamOptions.get()
                 .expectedRevision(ExpectedRevision.noStream());
 
-        WriteResult appendResult = client.appendToStream(streamName, appendOptions, expected.iterator())
+        WriteResult appendResult = client.appendToStream(streamName, appendOptions, expected)
                 .get();
 
-        Assertions.assertEquals(ExpectedRevision.expectedRevision(0), appendResult.getNextExpectedRevision());
-
-        ReadStreamOptions readStreamOptions = ReadStreamOptions.get()
-                .fromEnd()
-                .backwards()
-                .maxCount(1);
-
+        Assertions.assertEquals(ExpectedRevision.expectedRevision(1), appendResult.getNextExpectedRevision());
+        
         // Ensure appended event is readable
-        ReadResult result = client.readStream(streamName, readStreamOptions)
+        ReadResult result = client.readStream(streamName, ReadStreamOptions.get())
                 .get();
 
         Assertions.assertEquals(2, result.getEvents().size());