[DEVEX-250] Reshaped the API to handle regular and wrapped messages, together with explicit Message type

Author: oskardudycz

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

@@ -17,12 +17,20 @@
 class AppendToStream {
     private final GrpcClient client;
     private final String streamName;
+    private final StreamState streamState;
     private final List<MessageData> events;
     private final AppendToStreamOptions options;
 
-    public AppendToStream(GrpcClient client, String streamName, Iterator<MessageData> 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());
@@ -42,7 +50,7 @@ public CompletableFuture<WriteResult> execute() {
 
     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()));
@@ -117,7 +125,7 @@ private CompletableFuture<WriteResult> append(ManagedChannel channel, List<Messa
             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);

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

@@ -9,7 +9,6 @@
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
 import static io.kurrent.dbclient.serialization.MessageTypeNamingResolutionContext.fromStreamName;
@@ -33,8 +32,9 @@ public static KurrentDBClient create(KurrentDBClientSettings settings) {
     /**
      * Appends events to a given stream.
      *
+     * @deprecated This method may be removed in future releases. Prefer using appendToStream method with explicit stream state parameter.
      * @param streamName stream's name.
-     * @param events     events to send.
+     * @param events     events to store.
      * @return a write result if successful.
      * @see WriteResult
      */
@@ -45,8 +45,9 @@ public CompletableFuture<WriteResult> appendToStream(String streamName, EventDat
     /**
      * Appends events to a given stream.
      *
+     * @deprecated This method may be removed in future releases. Prefer using appendToStream method with explicit stream state parameter.
      * @param streamName stream's name.
-     * @param events     events to send.
+     * @param events     events to store.
      * @return a write result if successful.
      * @see WriteResult
      */
@@ -57,22 +58,24 @@ public CompletableFuture<WriteResult> appendToStream(String streamName, Iterator
     /**
      * Appends events to a given stream.
      *
+     * @deprecated This method may be removed in future releases. Prefer using appendToStream method with explicit stream state parameter.
      * @param streamName stream's name.
      * @param options    append stream request's options.
-     * @param events     events to send.
+     * @param events     events to store.
      * @return a write result if successful.
      * @see WriteResult
      */
     public CompletableFuture<WriteResult> appendToStream(String streamName, AppendToStreamOptions options, EventData... events) {
         return this.appendToStream(streamName, options, Arrays.stream(events).iterator());
     }
-
+    
     /**
      * Appends events to a given stream.
      *
+     * @deprecated This method may be removed in future releases. Prefer using appendToStream method with explicit stream state parameter.
      * @param streamName stream's name.
      * @param options    append stream request's options.
-     * @param events     events to send.
+     * @param events     events to store.
      * @return a write result if successful.
      * @see WriteResult
      */
@@ -85,73 +88,209 @@ public CompletableFuture<WriteResult> appendToStream(String streamName, AppendTo
                         .map(EventData::toMessageData)
                         .iterator();
 
-        return new AppendToStream(this.getGrpcClient(), streamName, messageData, options).execute();
+        return new AppendToStream(this.getGrpcClient(), streamName, options.getStreamState(), messageData, options).execute();
     }
 
     /**
      * Appends messages to a given stream.
      *
-     * @param streamName stream's name.
-     * @param messages   messages to send.
+     * @param streamName  stream's name.
+     * @param streamState expected stream's state in the database
+     * @param messages    messages to store.
      * @return a write result if successful.
      * @see WriteResult
      */
-    public CompletableFuture<WriteResult> appendToStream(String streamName, List<Message> messages) {
-        // TODO: Sort type erasure issue with Iterator
-        return this.appendToStream(streamName, AppendToStreamOptions.get(), messages);
+    public CompletableFuture<WriteResult> appendToStream(
+            String streamName, 
+            StreamState streamState,
+            Object... messages
+    ) {
+        return this.appendToStream(
+                streamName, 
+                streamState, 
+                Arrays.stream(messages).collect(Collectors.toList()), 
+                AppendToStreamOptions.get()
+        );
     }
 
+
     /**
      * Appends messages to a given stream.
      *
-     * @param streamName stream's name.
-     * @param options    append stream request's options.
-     * @param messages   messages to send.
+     * @param streamName  stream's name.
+     * @param streamState expected stream's state in the database
+     * @param messages    messages to store.
      * @return a write result if successful.
      * @see WriteResult
      */
-    public CompletableFuture<WriteResult> appendToStream(String streamName, AppendToStreamOptions options, Message... messages) {
-        return this.appendToStream(streamName, options, Stream.of(messages).collect(Collectors.toList()));
+    public CompletableFuture<WriteResult> appendToStream(
+            String streamName,
+            StreamState streamState,
+            Message... messages
+    ) {
+        List<Message> toAppend = Arrays.stream(messages).collect(Collectors.toList());
+        return this.appendToStream(
+                streamName,
+                streamState,
+                toAppend,
+                AppendToStreamOptions.get()
+        );
     }
 
     /**
      * Appends messages to a given stream.
      *
-     * @param streamName stream's name.
-     * @param options    append stream request's options.
-     * @param messages   messages to send.
+     * @param streamName  stream's name.
+     * @param streamState expected stream's state in the database
+     * @param messages    messages to store.
      * @return a write result if successful.
      * @see WriteResult
      */
-    public CompletableFuture<WriteResult> appendToStream(String streamName, AppendToStreamOptions options, Object... messages) {
-        return this.appendToStream(streamName, options, Stream.of(messages).map(Message::from).collect(Collectors.toList()));
+    public CompletableFuture<WriteResult> appendToStream(
+            String streamName,
+            StreamState streamState,
+            Iterable<Object> messages
+    ) {
+        return this.appendToStream(
+                streamName,
+                streamState,
+                messages,
+                AppendToStreamOptions.get()
+        );
     }
 
+
     /**
      * Appends messages to a given stream.
      *
-     * @param streamName stream's name.
-     * @param options    append stream request's options.
-     * @param messages   messages to send.
+     * @param streamName  stream's name.
+     * @param streamState expected stream's state in the database
+     * @param messages    messages to store.
+     * @param options     append stream request's options.
+     * @return a write result if successful.
+     * @see WriteResult
+     */
+    public CompletableFuture<WriteResult> appendToStream(
+            String streamName,
+            StreamState streamState,
+            Object[] messages,
+            AppendToStreamOptions options
+    ) {
+        List<Message> toAppend =
+                Arrays.stream(messages)
+                        .map(Message::from)
+                        .collect(Collectors.toList());
+
+        return this.appendToStream(
+                streamName,
+                streamState,
+                toAppend,
+                options
+        );
+    }
+
+    /**
+     * Appends messages to a given stream.
+     *
+     * @param streamName  stream's name.
+     * @param streamState expected stream's state in the database
+     * @param messages    messages to store.
+     * @param options     append stream request's options.
      * @return a write result if successful.
      * @see WriteResult
      */
-    public CompletableFuture<WriteResult> appendToStream(String streamName, AppendToStreamOptions options, List<Message> messages) {
-        // TODO: Sort type erasure issue with Iterator
+    public CompletableFuture<WriteResult> appendToStream(
+            String streamName,
+            StreamState streamState,
+            Iterable<Object> messages,
+            AppendToStreamOptions options
+    ) {
+        List<Message> toAppend =
+                StreamSupport.stream(messages.spliterator(), false)
+                        .map(Message::from)
+                        .collect(Collectors.toList());
+
+        return this.appendToStream(
+                streamName,
+                streamState,
+                toAppend,
+                options
+        );
+    }
+
+    /**
+     * Appends messages to a given stream.
+     *
+     * @param streamName  stream's name.
+     * @param streamState expected stream's state in the database
+     * @param messages    messages to store.
+     * @return a write result if successful.
+     * @see WriteResult
+     */
+    public CompletableFuture<WriteResult> appendToStream(
+            String streamName,
+            StreamState streamState,
+            List<Message> messages
+    ) {
+        return this.appendToStream(streamName, streamState, messages, AppendToStreamOptions.get());
+    }
+
+    /**
+     * Appends messages to a given stream.
+     *
+     * @param streamName  stream's name.
+     * @param streamState expected stream's state in the database.
+     * @param messages    messages to store.
+     * @param options     append stream request's options.
+     * @return a write result if successful.
+     * @see WriteResult
+     */
+    public CompletableFuture<WriteResult> appendToStream(
+            String streamName,
+            StreamState streamState,
+            Message[] messages,
+            AppendToStreamOptions options
+    ) {
+        List<Message> toAppend = Arrays.stream(messages).collect(Collectors.toList());
+
+        return this.appendToStream(
+                streamName,
+                streamState,
+                toAppend,
+                options
+        );
+    }
+
+    /**
+     * Appends messages to a given stream.
+     *
+     * @param streamName  stream's name.
+     * @param streamState expected stream's state in the database.
+     * @param messages    messages to store.
+     * @param options     append stream request's options.
+     * @return a write result if successful.
+     * @see WriteResult
+     */
+    public CompletableFuture<WriteResult> appendToStream(
+            String streamName,
+            StreamState streamState,
+            List<Message> messages,
+            AppendToStreamOptions options
+    ) {
         if (options == null)
             options = AppendToStreamOptions.get();
 
-        MessageSerializationContext serializationContext = 
+        MessageSerializationContext serializationContext =
                 new MessageSerializationContext(fromStreamName(streamName));
 
         MessageSerializer serializer = getGrpcClient()
                 .getSerializer(options.serializationSettings().orElse(null));
-        
+
         Iterator<MessageData> messageData = serializer
                 .serialize(messages, serializationContext)
                 .iterator();
 
-        return new AppendToStream(this.getGrpcClient(), streamName, messageData, options).execute();
+        return new AppendToStream(this.getGrpcClient(), streamName, streamState, messageData, options).execute();
     }
 
     /**

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

@@ -1,23 +1,32 @@
 package io.kurrent.dbclient;
 
+/**
+ * @deprecated This class may be removed in future releases. Prefer using appendToStream method with explicit stream state parameter
+ */
+@Deprecated
 class OptionsWithStreamStateBase<T> extends OptionsBase<T> {
     private StreamState streamState;
 
     protected OptionsWithStreamStateBase() {
         this.streamState = StreamState.any();
     }
 
+    /**
+     * @deprecated This method may be removed in future releases. Prefer using appendToStream method with explicit stream state parameter.
+     */
+    @Deprecated
     StreamState getStreamState() {
         return this.streamState;
     }
 
     /**
      * Asks the server to check that the stream receiving is at the expected state.
-
+     * @deprecated This method may be removed in future releases. Prefer using appendToStream method with explicit stream state parameter.
      * @param state - expected revision.
      * @return updated options.
      */
     @SuppressWarnings("unchecked")
+    @Deprecated
     public T streamState(StreamState state) {
         this.streamState = state;
         return (T) this;
@@ -26,10 +35,11 @@ public T streamState(StreamState state) {
 
     /**
      * Asks the server to check that the stream receiving is at the given expected revision.
-
+     * @deprecated This method may be removed in future releases. Prefer using appendToStream method with explicit stream state parameter.
      * @param revision - expected revision.
      * @return updated options.
      */
+    @Deprecated
     public T streamRevision(long revision) {
         return streamState(StreamState.streamRevision(revision));
     }