[DEVEX-250] Added Message deserialization

Author: oskardudycz

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

@@ -10,9 +10,15 @@ abstract class AbstractRead implements Publisher<ReadMessage> {
     protected static final StreamsOuterClass.ReadReq.Options.Builder defaultReadOptions;
 
     private final GrpcClient client;
+<<<<<<< HEAD
     private final OptionsBase<?> options;
 
     protected AbstractRead(GrpcClient client, OptionsBase<?> options) {
+=======
+    private final OptionsWithBackPressureAndSerialization<?> options;
+
+    protected AbstractRead(GrpcClient client, OptionsWithBackPressureAndSerialization<?> options) {
+>>>>>>> ce0858b ([DEVEX-250] Added Message deserialization)
         this.client = client;
         this.options = options;
     }
@@ -28,8 +34,22 @@ protected AbstractRead(GrpcClient client, OptionsBase<?> options) {
     @Override
     @SuppressWarnings("unchecked")
     public void subscribe(Subscriber<? super ReadMessage> subscriber) {
+<<<<<<< HEAD
         ReadSubscription readSubscription = new ReadSubscription(subscriber);
         subscriber.onSubscribe(readSubscription);
+=======
+        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;
+            }
+>>>>>>> ce0858b ([DEVEX-250] Added Message deserialization)
 
         CompletableFuture<ReadSubscription> result = new CompletableFuture<>();
         this.client.run(channel -> {

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

@@ -22,9 +22,15 @@ abstract class AbstractRegularSubscription {
     protected SubscriptionListener listener;
     protected Checkpointer checkpointer = null;
     private final GrpcClient client;
+<<<<<<< HEAD
     private final OptionsBase<?> options;
 
     protected AbstractRegularSubscription(GrpcClient client, OptionsBase<?> options) {
+=======
+    private final OptionsWithBackPressureAndSerialization<?> options;
+
+    protected AbstractRegularSubscription(GrpcClient client, OptionsWithBackPressureAndSerialization<?> options) {
+>>>>>>> ce0858b ([DEVEX-250] Added Message deserialization)
         this.client = client;
         this.options = options;
     }
@@ -155,5 +161,28 @@ public void onCompleted() {
 
             return future;
         });
+<<<<<<< HEAD
+=======
+
+        return future;
+    }
+
+    private ReadResponseObserver createObserver(WorkItemArgs args, CompletableFuture<Subscription> future) {
+        StreamConsumer consumer = new SubscriptionStreamConsumer(this.listener, this.checkpointer, future, (subscriptionId, event, action) -> {
+            ClientTelemetry.traceSubscribe(
+                    action,
+                    subscriptionId,
+                    args.getChannel(),
+                    client.getSettings(),
+                    options.getCredentials(),
+                    event);
+        });
+
+        return new ReadResponseObserver(
+                this.options, 
+                consumer, 
+                this.client.getSerializer(options.serializationSettings().orElse(null))
+        );
+>>>>>>> ce0858b ([DEVEX-250] Added Message deserialization)
     }
 }

db-client-java/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(),

db-client-java/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;
 
@@ -18,11 +21,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() {
@@ -72,10 +78,27 @@ public <A> CompletableFuture<A> runWithArgs(Function<WorkItemArgs, CompletableFu
                 return;
             }
 
+<<<<<<< HEAD
             action.apply(args).whenComplete((outcome, error) -> {
                 if (outcome != null) {
                     result.complete(outcome);
                     return;
+=======
+                    if (e instanceof StatusRuntimeException) {
+                        StatusRuntimeException ex = (StatusRuntimeException) e;
+
+                        if (ex.getStatus().getCode().equals(Status.Code.UNAVAILABLE)) {
+                            this.push(new CreateChannel(args.getId()));
+                        }
+                    }
+
+                    logger.debug("RunWorkItem[{}] completed exceptionally: {}", args.getId(), e.toString());
+
+                    if (e instanceof RuntimeException)
+                        throw (RuntimeException) e;
+                    else
+                        throw new RuntimeException(e);
+>>>>>>> ce0858b ([DEVEX-250] Added Message deserialization)
                 }
 
                 try {
@@ -115,4 +138,8 @@ public CompletableFuture<Void> shutdown() {
     public KurrentDBClientSettings getSettings() {
         return this.settings;
     }
+
+    public MessageSerializer getSerializer(OperationSerializationSettings serializationSettings) {
+        return this.serializer.with(serializationSettings);
+    }
 }

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

@@ -3,6 +3,7 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.json.JsonMapper;
 import io.kurrent.dbclient.serialization.MessageSerializationContext;
+import io.kurrent.dbclient.serialization.MessageSerializer;
 import org.reactivestreams.Publisher;
 
 import java.util.*;
@@ -140,11 +141,13 @@ public CompletableFuture<WriteResult> appendToStream(String streamName, AppendTo
         if (options == null)
             options = AppendToStreamOptions.get();
 
-        MessageSerializationContext serializationContext = new MessageSerializationContext(fromStreamName(streamName));
+        MessageSerializationContext serializationContext = 
+                new MessageSerializationContext(fromStreamName(streamName));
 
-        Iterator<MessageData> messageData = options.serializationSettings()
-                .map(serializer::with)
-                .orElse(serializer)
+        MessageSerializer serializer = getGrpcClient()
+                .getSerializer(options.serializationSettings().orElse(null));
+        
+        Iterator<MessageData> messageData = serializer
                 .serialize(messages, serializationContext)
                 .iterator();
 

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

@@ -16,7 +16,6 @@
 public class KurrentDBClientBase {
     final Logger logger = LoggerFactory.getLogger(KurrentDBClientBase.class);
     final private GrpcClient client;
-    final protected MessageSerializer serializer;
 
     KurrentDBClientBase(KurrentDBClientSettings settings) {
         Discovery discovery;
@@ -33,8 +32,6 @@ public class KurrentDBClientBase {
         this.client = service.getHandle();
 
         CompletableFuture.runAsync(service, createConnectionLoopExecutor());
-
-        serializer = MessageSerializerBuilder.get(settings.getSerializationSettings());
     }
     private Executor createConnectionLoopExecutor() {
         return Executors.newSingleThreadExecutor(r -> {

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

@@ -34,7 +34,7 @@ public Message(Object data, Object metadata, UUID messageId) {
      *
      * <p>Example:
      * <pre>
-     * // Create a message with a specific ID
+     * Create a message with a specific ID
      * UserRegistered userRegistered = new UserRegistered("123", "Alice");
      * Message message = Message.from(userRegistered);
      * </pre>
@@ -53,7 +53,7 @@ public static Message from(Object data) {
      *
      * <p>Example:
      * <pre>
-     * // Create a message with a specific ID
+     * Create a message with a specific ID
      * UserRegistered userRegistered = new UserRegistered("123", "Alice");
      * UUID messageId = UUID.randomUUID();
      * Message message = Message.from(userRegistered, messageId);
@@ -74,7 +74,7 @@ public static Message from(Object data, UUID messageId) {
      *
      * <p>Example:
      * <pre>
-     * // Create a message with data and metadata
+     * Create a message with data and metadata
      * OrderPlaced orderPlaced = new OrderPlaced("ORD-123", 99.99);
      * EventMetadata metadata = new EventMetadata(
      *     "user-456", 

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

@@ -8,7 +8,7 @@ class OptionsBase<T> {
     private final OperationKind kind;
     private UserCredentials credentials;
     private boolean requiresLeader;
-    private Map<String, String> headers = new HashMap<>();
+    private final Map<String, String> headers = new HashMap<>();
 
     protected OptionsBase() {
         this(OperationKind.Regular);

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

@@ -0,0 +1,29 @@
+package io.kurrent.dbclient;
+
+import java.util.Optional;
+
+import io.kurrent.dbclient.serialization.OperationSerializationSettings;
+
+class OptionsWithBackPressureAndSerialization<T> extends OptionsWithBackPressure<T> {
+    public OperationSerializationSettings serializationSettings;
+
+    protected OptionsWithBackPressureAndSerialization(OperationKind kind) {
+        super(kind);
+    }
+    
+    /**
+     * Allows to customize or disable the automatic deserialization.
+     */
+    public Optional<OperationSerializationSettings> serializationSettings() {
+        return Optional.ofNullable(serializationSettings);
+    }
+
+    /**
+     * Customize or disable the automatic deserialization.
+     */
+    @SuppressWarnings("unchecked")
+    public T serializationSettings(OperationSerializationSettings serializationSettings) {
+        this.serializationSettings = serializationSettings;
+        return (T)this;
+    }
+}

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

@@ -1,6 +1,6 @@
 package io.kurrent.dbclient;
 
-class OptionsWithPositionAndResolveLinkTosBase<T> extends OptionsWithBackPressure<T> {
+class OptionsWithPositionAndResolveLinkTosBase<T> extends OptionsWithBackPressureAndSerialization<T> {
     private StreamPosition<Position> position;
 
     protected OptionsWithPositionAndResolveLinkTosBase(OperationKind kind) {