feat: add StreamBuilder and Replace Record with ReceivedRecord in Reader (#152)

feat: add StreamBuilder and HStreamClient.createStream(Stream stream)

Author: s12f

client/src/main/java/io/hstream/HStreamClient.java

@@ -59,6 +59,13 @@ static HStreamClientBuilder builder() {
    */
   void createStream(String stream, short replicationFactor, int shardCount, int backlogDuration);
 
+  /**
+   * Create a new stream.
+   *
+   * @param stream Stream Object, you should use {@link Stream.Builder} to build it.
+   */
+  void createStream(Stream stream);
+
   /**
    * List shards in a stream.
    *

client/src/main/java/io/hstream/Reader.java

@@ -7,10 +7,10 @@
 public interface Reader extends AutoCloseable {
 
   /**
-   * Read {@link Record}s from a stream shard.
+   * Read {@link ReceivedRecord}s from a stream shard.
    *
    * @param maxRecords the max number of the returned records
-   * @return the {@link Record}s wrapped in a {@link CompletableFuture}
+   * @return the {@link ReceivedRecord}s wrapped in a {@link CompletableFuture}
    */
-  CompletableFuture<List<Record>> read(int maxRecords);
+  CompletableFuture<List<ReceivedRecord>> read(int maxRecords);
 }

client/src/main/java/io/hstream/ReceivedRecord.java

@@ -0,0 +1,27 @@
+package io.hstream;
+
+public class ReceivedRecord {
+  String recordId;
+  Record record;
+
+  public ReceivedRecord(String recordId, Record record) {
+    this.recordId = recordId;
+    this.record = record;
+  }
+
+  public String getRecordId() {
+    return recordId;
+  }
+
+  public void setRecordId(String recordId) {
+    this.recordId = recordId;
+  }
+
+  public Record getRecord() {
+    return record;
+  }
+
+  public void setRecord(Record record) {
+    this.record = record;
+  }
+}

client/src/main/java/io/hstream/Stream.java

@@ -1,5 +1,8 @@
 package io.hstream;
 
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
 import java.util.Objects;
 
 public class Stream {
@@ -62,4 +65,59 @@ public boolean equals(Object o) {
   public int hashCode() {
     return Objects.hash(streamName, replicationFactor, backlogDuration, shardCount);
   }
+
+  public static final class Builder {
+    private String streamName;
+    private int replicationFactor = 1;
+    private int backlogDuration = 3600 * 24;
+    private int shardCount = 1;
+
+    /**
+     * @param streamName required, the name of the stream
+     * @return Stream.Builder instance
+     */
+    public Builder streamName(String streamName) {
+      this.streamName = streamName;
+      return this;
+    }
+
+    /**
+     * @param replicationFactor optional(default: 1), replication factor of the stream
+     * @return Stream.Builder instance
+     */
+    public Builder replicationFactor(int replicationFactor) {
+      this.replicationFactor = replicationFactor;
+      return this;
+    }
+
+    /**
+     * @param backlogDuration optional(default: 3600 * 24), backlog duration(in seconds) of the
+     *     stream
+     * @return Stream.Builder instance
+     */
+    public Builder backlogDuration(int backlogDuration) {
+      this.backlogDuration = backlogDuration;
+      return this;
+    }
+
+    /**
+     * @param shardCount optional(default: 1), number of shards in the stream
+     * @return Stream.Builder instance
+     */
+    public Builder shardCount(int shardCount) {
+      this.shardCount = shardCount;
+      return this;
+    }
+
+    public Stream build() {
+      checkNotNull(streamName);
+      checkArgument(replicationFactor >= 1 && replicationFactor <= 15);
+      checkArgument(shardCount >= 1);
+      return new Stream(streamName, replicationFactor, backlogDuration, shardCount);
+    }
+  }
+
+  public static Builder newBuilder() {
+    return new Builder();
+  }
 }

client/src/main/kotlin/io/hstream/impl/HStreamClientKtImpl.kt

@@ -121,6 +121,15 @@ class HStreamClientKtImpl(bootstrapServerUrls: List<String>, credentials: Channe
         }
     }
 
+    override fun createStream(stream: Stream?) {
+        checkNotNull(stream)
+        unaryCallBlocked {
+            it.createStream(
+                GrpcUtils.streamToGrpc(stream)
+            )
+        }
+    }
+
     override fun listShards(streamName: String?): List<Shard> {
         checkNotNull(streamName)
         val listShardsRequest = ListShardsRequest.newBuilder().setStreamName(streamName).build()

client/src/main/kotlin/io/hstream/impl/ReaderKtImpl.kt

@@ -2,6 +2,7 @@ package io.hstream.impl
 
 import io.hstream.HStreamDBClientException
 import io.hstream.Reader
+import io.hstream.ReceivedRecord
 import io.hstream.Record
 import io.hstream.StreamShardOffset
 import io.hstream.internal.CreateShardReaderRequest
@@ -43,8 +44,8 @@ class ReaderKtImpl(
         logger.info("created Reader [{}] for stream [{}] shard [{}]", readerId, streamName, shardId)
     }
 
-    override fun read(maxRecords: Int): CompletableFuture<MutableList<Record>> {
-        val readFuture = CompletableFuture<MutableList<Record>>()
+    override fun read(maxRecords: Int): CompletableFuture<MutableList<ReceivedRecord>> {
+        val readFuture = CompletableFuture<MutableList<ReceivedRecord>>()
         readerScope.launch {
             try {
                 val readShardRequest = ReadShardRequest.newBuilder().setReaderId(readerId).setMaxRecords(maxRecords).build()
@@ -54,16 +55,17 @@ class ReaderKtImpl(
                     RecordUtils.decompress(it).map { receivedHStreamRecord ->
                         val hStreamRecord = receivedHStreamRecord.record
                         val header = RecordUtils.parseRecordHeaderFromHStreamRecord(hStreamRecord)
-                        if (RecordUtils.isRawRecord(hStreamRecord)) {
+                        val record = if (RecordUtils.isRawRecord(hStreamRecord)) {
                             val rawRecord = RecordUtils.parseRawRecordFromHStreamRecord(hStreamRecord)
                             Record.newBuilder().rawRecord(rawRecord).partitionKey(header.partitionKey).build()
                         } else {
                             val hRecord = RecordUtils.parseHRecordFromHStreamRecord(hStreamRecord)
                             Record.newBuilder().hRecord(hRecord).partitionKey(header.partitionKey).build()
                         }
+                        ReceivedRecord(GrpcUtils.recordIdFromGrpc(receivedHStreamRecord.recordId), record)
                     }
                 }
-                readFuture.complete(res as MutableList<Record>?)
+                readFuture.complete(res as MutableList<ReceivedRecord>?)
             } catch (e: Throwable) {
                 readFuture.completeExceptionally(HStreamDBClientException(e))
             }