WIP: [DEVEX-250] Add built-in auto-serialization

CODE-OF-CONDUCT.md

@@ -1,6 +1,6 @@
 # Contributor Covenant Code of Conduct
 
-`KurrentDB-Client-Java` follows the widely-adopted Contributor Covenant Code of Conduct.
+`Kurrent-Client-Java` follows the widely-adopted Contributor Covenant Code of Conduct.
 
 ## Our Pledge
 

build.gradle

@@ -61,6 +61,7 @@ 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"
@@ -79,7 +80,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"

src/main/java/io/kurrent/dbclient/AbstractRead.java

@@ -10,9 +10,9 @@ abstract class AbstractRead implements Publisher<ReadMessage> {
     protected static final StreamsOuterClass.ReadReq.Options.Builder defaultReadOptions;
 
     private final GrpcClient client;
-    private final OptionsWithBackPressure<?> options;
+    private final OptionsWithBackPressureAndSerialization<?> options;
 
-    protected AbstractRead(GrpcClient client, OptionsWithBackPressure<?> options) {
+    protected AbstractRead(GrpcClient client, OptionsWithBackPressureAndSerialization<?> options) {
         this.client = client;
         this.options = options;
     }
@@ -27,13 +27,17 @@ protected AbstractRead(GrpcClient client, OptionsWithBackPressure<?> options) {
 
     @Override
     public void subscribe(Subscriber<? super ReadMessage> subscriber) {
-        ReadResponseObserver observer = new ReadResponseObserver(options, new ReadStreamConsumer(subscriber));
+        ReadResponseObserver observer = new ReadResponseObserver(
+                options, 
+                new ReadStreamConsumer(subscriber), 
+                this.client.getSerializer(options.serializationSettings().orElse(null))
+        );
 
         this.client.getWorkItemArgs().whenComplete((args, error) -> {
-           if (error != null) {
-               observer.onError(error);
-               return;
-           }
+            if (error != null) {
+                observer.onError(error);
+                return;
+            }
 
             StreamsOuterClass.ReadReq request = StreamsOuterClass.ReadReq.newBuilder()
                     .setOptions(createOptions())

src/main/java/io/kurrent/dbclient/AbstractRegularSubscription.java

@@ -3,13 +3,6 @@
 import io.kurrent.dbclient.proto.shared.Shared;
 import io.kurrent.dbclient.proto.streams.StreamsGrpc;
 import io.kurrent.dbclient.proto.streams.StreamsOuterClass;
-import io.grpc.Metadata;
-import io.grpc.Status;
-import io.grpc.StatusRuntimeException;
-import io.grpc.stub.ClientCallStreamObserver;
-import io.grpc.stub.ClientResponseObserver;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import io.grpc.ManagedChannel;
 
 import java.util.concurrent.CompletableFuture;
@@ -21,9 +14,9 @@ abstract class AbstractRegularSubscription {
     protected SubscriptionListener listener;
     protected Checkpointer checkpointer = null;
     private final GrpcClient client;
-    private final OptionsWithBackPressure<?> options;
+    private final OptionsWithBackPressureAndSerialization<?> options;
 
-    protected AbstractRegularSubscription(GrpcClient client, OptionsWithBackPressure<?> options) {
+    protected AbstractRegularSubscription(GrpcClient client, OptionsWithBackPressureAndSerialization<?> options) {
         this.client = client;
         this.options = options;
     }
@@ -77,6 +70,10 @@ private ReadResponseObserver createObserver(ManagedChannel channel, CompletableF
                 }
         );
 
-        return new ReadResponseObserver(this.options, consumer);
+        return new ReadResponseObserver(
+                this.options, 
+                consumer, 
+                this.client.getSerializer(options.serializationSettings().orElse(null))
+        );
     }
 }

src/main/java/io/kurrent/dbclient/AbstractSubscribePersistentSubscription.java

@@ -1,14 +1,15 @@
 package io.kurrent.dbclient;
 
-import io.kurrent.dbclient.proto.persistentsubscriptions.Persistent;
-import io.kurrent.dbclient.proto.persistentsubscriptions.PersistentSubscriptionsGrpc;
-import io.kurrent.dbclient.proto.shared.Shared;
 import io.grpc.Metadata;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.grpc.stub.ClientCallStreamObserver;
 import io.grpc.stub.ClientResponseObserver;
 import io.grpc.stub.StreamObserver;
+import io.kurrent.dbclient.proto.persistentsubscriptions.Persistent;
+import io.kurrent.dbclient.proto.persistentsubscriptions.PersistentSubscriptionsGrpc;
+import io.kurrent.dbclient.proto.shared.Shared;
+import io.kurrent.dbclient.serialization.MessageSerializer;
 
 import java.util.concurrent.CompletableFuture;
 
@@ -18,20 +19,26 @@ abstract class AbstractSubscribePersistentSubscription {
     private final String group;
     private final PersistentSubscriptionListener listener;
     private final SubscribePersistentSubscriptionOptions options;
+    private final MessageSerializer messageSerializer;
 
     static {
         defaultReadOptions = Persistent.ReadReq.Options.newBuilder()
                 .setUuidOption(Persistent.ReadReq.Options.UUIDOption.newBuilder()
                         .setStructured(Shared.Empty.getDefaultInstance()));
     }
 
-    public AbstractSubscribePersistentSubscription(GrpcClient client, String group,
-                                                   SubscribePersistentSubscriptionOptions options,
-                                                   PersistentSubscriptionListener listener) {
+    public AbstractSubscribePersistentSubscription(
+            GrpcClient client,
+            String group,
+            SubscribePersistentSubscriptionOptions options,
+            PersistentSubscriptionListener listener,
+            MessageSerializer messageSerializer
+    ) {
         this.client = client;
         this.group = group;
         this.options = options;
         this.listener = listener;
+        this.messageSerializer = messageSerializer;
     }
 
     protected abstract Persistent.ReadReq.Options.Builder createOptions();
@@ -91,7 +98,7 @@ public void onNext(Persistent.ReadResp readResp) {
                         int retryCount = readResp.getEvent().hasNoRetryCount() ? 0 : readResp.getEvent().getRetryCount();
 
                         try {
-                            ResolvedEvent resolvedEvent = ResolvedEvent.fromWire(readResp.getEvent());
+                            ResolvedEvent resolvedEvent = ResolvedEvent.fromWire(readResp.getEvent(), messageSerializer);
                             ClientTelemetry.traceSubscribe(
                                     () -> listener.onEvent(this._subscription, retryCount, resolvedEvent),
                                     _subscription.getSubscriptionId(),

src/main/java/io/kurrent/dbclient/AppendToStream.java

@@ -17,12 +17,20 @@
 class AppendToStream {
     private final GrpcClient client;
     private final String streamName;
-    private final List<EventData> events;
+    private final StreamState streamState;
+    private final List<MessageData> events;
     private final AppendToStreamOptions options;
 
-    public AppendToStream(GrpcClient client, String streamName, Iterator<EventData> events, AppendToStreamOptions options) {
+    public AppendToStream(
+            GrpcClient client, 
+            String streamName, 
+            StreamState streamState, 
+            Iterator<MessageData> events, 
+            AppendToStreamOptions options
+    ) {
         this.client = client;
         this.streamName = streamName;
+        this.streamState = streamState;
         this.events = new ArrayList<>();
         while (events.hasNext()) {
             this.events.add(events.next());
@@ -40,9 +48,9 @@ public CompletableFuture<WriteResult> execute() {
                 this.options.getCredentials()));
     }
 
-    private CompletableFuture<WriteResult> append(ManagedChannel channel, List<EventData> events) {
+    private CompletableFuture<WriteResult> append(ManagedChannel channel, List<MessageData> events) {
         CompletableFuture<WriteResult> result = new CompletableFuture<>();
-        StreamsOuterClass.AppendReq.Options.Builder options = this.options.getStreamState().applyOnWire(StreamsOuterClass.AppendReq.Options.newBuilder()
+        StreamsOuterClass.AppendReq.Options.Builder options = this.streamState.applyOnWire(StreamsOuterClass.AppendReq.Options.newBuilder()
                 .setStreamIdentifier(Shared.StreamIdentifier.newBuilder()
                         .setStreamName(ByteString.copyFromUtf8(streamName))
                         .build()));
@@ -93,18 +101,18 @@ private CompletableFuture<WriteResult> append(ManagedChannel channel, List<Event
         try {
             requestStream.onNext(StreamsOuterClass.AppendReq.newBuilder().setOptions(options).build());
 
-            for (EventData e : events) {
+            for (MessageData e : events) {
                 StreamsOuterClass.AppendReq.ProposedMessage.Builder msgBuilder = StreamsOuterClass.AppendReq.ProposedMessage.newBuilder()
                         .setId(Shared.UUID.newBuilder()
                                 .setStructured(Shared.UUID.Structured.newBuilder()
-                                        .setMostSignificantBits(e.getEventId().getMostSignificantBits())
-                                        .setLeastSignificantBits(e.getEventId().getLeastSignificantBits())))
-                        .setData(ByteString.copyFrom(e.getEventData()))
+                                        .setMostSignificantBits(e.getMessageId().getMostSignificantBits())
+                                        .setLeastSignificantBits(e.getMessageId().getLeastSignificantBits())))
+                        .setData(ByteString.copyFrom(e.getMessageData()))
                         .putMetadata(SystemMetadataKeys.CONTENT_TYPE, e.getContentType())
-                        .putMetadata(SystemMetadataKeys.TYPE, e.getEventType());
+                        .putMetadata(SystemMetadataKeys.TYPE, e.getMessageType());
 
-                if (e.getUserMetadata() != null) {
-                    msgBuilder.setCustomMetadata(ByteString.copyFrom(e.getUserMetadata()));
+                if (e.getMessageMetadata() != null) {
+                    msgBuilder.setCustomMetadata(ByteString.copyFrom(e.getMessageMetadata()));
                 }
 
                 requestStream.onNext(StreamsOuterClass.AppendReq.newBuilder()
@@ -117,7 +125,7 @@ private CompletableFuture<WriteResult> append(ManagedChannel channel, List<Event
             String leaderPort = e.getTrailers().get(Metadata.Key.of("leader-endpoint-port", Metadata.ASCII_STRING_MARSHALLER));
 
             if (leaderHost != null && leaderPort != null) {
-                NotLeaderException reason = new NotLeaderException(leaderHost, Integer.valueOf(leaderPort));
+                NotLeaderException reason = new NotLeaderException(leaderHost, Integer.parseInt(leaderPort));
                 result.completeExceptionally(reason);
             } else {
                 result.completeExceptionally(e);

src/main/java/io/kurrent/dbclient/AppendToStreamOptions.java

@@ -1,12 +1,36 @@
 package io.kurrent.dbclient;
 
+import io.kurrent.dbclient.serialization.OperationSerializationSettings;
+
+import java.util.Optional;
+
 /**
  * Options of the append stream request.
  */
 public class AppendToStreamOptions extends OptionsWithStreamStateBase<AppendToStreamOptions> {
+    private OperationSerializationSettings serializationSettings = null;
+
     private AppendToStreamOptions() {
     }
 
+    /**
+     * Returns optional serialization settings
+     */
+    public Optional<OperationSerializationSettings> serializationSettings() {
+        return Optional.ofNullable(this.serializationSettings);
+    }
+
+    /**
+     * Allows to customize or disable the automatic deserialization
+     *
+     * @param serializationSettings - expected revision.
+     * @return updated options.
+     */
+    public AppendToStreamOptions serializationSettings(OperationSerializationSettings serializationSettings) {
+        this.serializationSettings = serializationSettings;
+        return this;
+    }
+
     /**
      * Returns options with default values.
      */

src/main/java/io/kurrent/dbclient/ClientTelemetry.java

@@ -28,15 +28,15 @@ private static Tracer getTracer() {
                 ClientTelemetry.class.getPackage().getImplementationVersion());
     }
 
-    private static List<EventData> tryInjectTracingContext(Span span, List<EventData> events) {
-        List<EventData> injectedEvents = new ArrayList<>();
-        for (EventData event : events) {
+    private static List<MessageData> tryInjectTracingContext(Span span, List<MessageData> events) {
+        List<MessageData> injectedEvents = new ArrayList<>();
+        for (MessageData event : events) {
             boolean isJsonEvent = Objects.equals(event.getContentType(), ContentType.JSON);
 
-            injectedEvents.add(EventDataBuilder
-                    .binary(event.getEventId(), event.getEventType(), event.getEventData(), isJsonEvent)
-                    .metadataAsBytes(tryInjectTracingContext(span, event.getUserMetadata()))
-                    .build());
+            injectedEvents.add(
+                    MessageDataBuilder
+                            .with(event.getMessageType(), event.getMessageData(), tryInjectTracingContext(span, event.getMessageMetadata()), event.getMessageId(), isJsonEvent)
+                            .build());
         }
         return injectedEvents;
     }
@@ -85,9 +85,9 @@ private static SpanContext tryExtractTracingContext(byte[] userMetadataBytes) {
     }
 
     static CompletableFuture<WriteResult> traceAppend(
-            BiFunction<ManagedChannel, List<EventData>, CompletableFuture<WriteResult>> appendOperation,
+            BiFunction<ManagedChannel, List<MessageData>, CompletableFuture<WriteResult>> appendOperation,
             ManagedChannel channel,
-            List<EventData> events, String streamId, KurrentDBClientSettings settings,
+            List<MessageData> events, String streamId, KurrentDBClientSettings settings,
             UserCredentials optionalCallCredentials) {
         Span span = createSpan(
                 ClientTelemetryConstants.Operations.APPEND,

src/main/java/io/kurrent/dbclient/ConnectionSettingsBuilder.java

@@ -2,6 +2,7 @@
 
 
 import io.grpc.ClientInterceptor;
+import io.kurrent.dbclient.serialization.KurrentDBClientSerializationSettings;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -36,6 +37,7 @@ public class ConnectionSettingsBuilder {
     private List<ClientInterceptor> _interceptors = new ArrayList<>();
     private String _tlsCaFile = null;
     private Set<String> _features = new HashSet<>();
+    private KurrentDBClientSerializationSettings _serializationSettings;
 
     ConnectionSettingsBuilder() {}
 
@@ -60,7 +62,9 @@ public KurrentDBClientSettings buildConnectionSettings() {
                 _defaultDeadline,
                 _interceptors,
                 _tlsCaFile,
-                _features);
+                _features,
+                _serializationSettings
+        );
     }
 
     /**
@@ -241,6 +245,15 @@ public ConnectionSettingsBuilder feature(String feature) {
         return this;
     }
 
+    /**
+     * Provides configuration options for messages serialization and deserialization in the KurrentDB client.
+     * If null, default settings are used.
+     */
+    public ConnectionSettingsBuilder serialization(KurrentDBClientSerializationSettings serializationSettings) {
+        this._serializationSettings = serializationSettings;
+        return this;
+    }
+
     void parseGossipSeed(String host) {
         String[] hostParts = host.split(":");
 

src/main/java/io/kurrent/dbclient/EventData.java

@@ -96,5 +96,9 @@ public static EventDataBuilder builderAsBinary(String eventType, byte[] eventDat
     public static EventDataBuilder builderAsBinary(UUID eventId, String eventType, byte[] eventData) {
         return EventDataBuilder.binary(eventId, eventType, eventData);
     }
+
+    public MessageData toMessageData() {
+        return new MessageData(eventType, eventData, userMetadata, eventId, contentType);
+    }
 }
 

src/main/java/io/kurrent/dbclient/GrpcClient.java

@@ -3,6 +3,9 @@
 import io.grpc.ManagedChannel;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
+import io.kurrent.dbclient.serialization.MessageSerializer;
+import io.kurrent.dbclient.serialization.MessageSerializerBuilder;
+import io.kurrent.dbclient.serialization.OperationSerializationSettings;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -19,11 +22,14 @@ class GrpcClient {
     private final AtomicBoolean closed;
     private final LinkedBlockingQueue<Msg> queue;
     private final KurrentDBClientSettings settings;
+    private final MessageSerializer serializer;
 
     GrpcClient(KurrentDBClientSettings settings, AtomicBoolean closed, LinkedBlockingQueue<Msg> queue) {
         this.settings = settings;
         this.closed = closed;
         this.queue = queue;
+
+        this.serializer = MessageSerializerBuilder.get(settings.getSerializationSettings());
     }
 
     public boolean isShutdown() {
@@ -101,7 +107,7 @@ public <A> CompletableFuture<A> runWithArgs(Function<WorkItemArgs, CompletableFu
                     logger.debug("RunWorkItem[{}] completed exceptionally: {}", args.getId(), e.toString());
 
                     if (e instanceof RuntimeException)
-                        throw (RuntimeException)e;
+                        throw (RuntimeException) e;
                     else
                         throw new RuntimeException(e);
                 }
@@ -120,4 +126,8 @@ public CompletableFuture<Void> shutdown() {
     public KurrentDBClientSettings getSettings() {
         return this.settings;
     }
+
+    public MessageSerializer getSerializer(OperationSerializationSettings serializationSettings) {
+        return this.serializer.with(serializationSettings);
+    }
 }