feat: update protocol (#212)

Author: s12f

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

@@ -173,4 +173,6 @@ static HStreamClientBuilder builder() {
   String getConnectorLogs(String name, int begin, int count);
 
   void deleteConnector(String name);
+
+  String getTailRecordId(String streamName, long shardId);
 }

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

@@ -6,7 +6,9 @@ public interface StreamShardReaderBuilder {
 
   StreamShardReaderBuilder shardId(long shardId);
 
-  StreamShardReaderBuilder shardOffset(StreamShardOffset shardOffset);
+  StreamShardReaderBuilder from(StreamShardOffset shardOffset);
+  StreamShardReaderBuilder maxReadBatches(long maxReadBatches);
+  StreamShardReaderBuilder until(StreamShardOffset until);
 
   StreamShardReaderBuilder receiver(StreamShardReaderReceiver streamShardReaderReceiver);
   StreamShardReaderBuilder batchReceiver(StreamShardReaderBatchReceiver streamShardReaderReceiver);

client/src/main/java/io/hstream/impl/StreamShardReaderBuilderImpl.java

@@ -10,7 +10,9 @@ public class StreamShardReaderBuilderImpl implements StreamShardReaderBuilder {
   private final HStreamClientKtImpl client;
   private String streamName;
   private long shardId;
-  private StreamShardOffset shardOffset;
+  private StreamShardOffset from;
+  long maxReadBatches;
+  StreamShardOffset until;
 
   private StreamShardReaderReceiver receiver;
   private StreamShardReaderBatchReceiver batchReceiver;
@@ -32,8 +34,20 @@ public StreamShardReaderBuilder shardId(long shardId) {
   }
 
   @Override
-  public StreamShardReaderBuilder shardOffset(StreamShardOffset shardOffset) {
-    this.shardOffset = shardOffset;
+  public StreamShardReaderBuilder from(StreamShardOffset shardOffset) {
+    this.from = shardOffset;
+    return this;
+  }
+
+  @Override
+  public StreamShardReaderBuilder maxReadBatches(long maxReadBatches) {
+    this.maxReadBatches = maxReadBatches;
+    return this;
+  }
+
+  @Override
+  public StreamShardReaderBuilder until(StreamShardOffset until) {
+    this.until = until;
     return this;
   }
 
@@ -55,9 +69,9 @@ public StreamShardReader build() {
     checkArgument(streamName != null, "StreamShardReaderBuilder: `streamName` should not be null");
     checkArgument(shardId > 0, "StreamShardReaderBuilder: `shardId` error");
     checkArgument(
-        shardOffset != null, "StreamShardReaderBuilder: `shardOffset` should not be null");
+        from != null, "StreamShardReaderBuilder: `from` should not be null");
     checkArgument(receiver != null || batchReceiver != null,
             "StreamShardReaderBuilder: `receiver` should not be both null");
-    return new StreamShardReaderKtImpl(client, streamName, shardId, shardOffset, receiver, batchReceiver);
+    return new StreamShardReaderKtImpl(client, streamName, shardId, from, maxReadBatches, until, receiver, batchReceiver);
   }
 }

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

@@ -37,6 +37,7 @@ import io.hstream.internal.GetConnectorSpecRequest
 import io.hstream.internal.GetQueryRequest
 import io.hstream.internal.GetStreamRequest
 import io.hstream.internal.GetSubscriptionRequest
+import io.hstream.internal.GetTailRecordIdRequest
 import io.hstream.internal.GetViewRequest
 import io.hstream.internal.HStreamApiGrpcKt
 import io.hstream.internal.ListConnectorsRequest
@@ -414,6 +415,17 @@ class HStreamClientKtImpl(
         }
     }
 
+    override fun getTailRecordId(streamName: String?, shardId: Long): String {
+        checkNotNull(streamName)
+        return unaryCallBlocked {
+            val recordId = it.getTailRecordId(GetTailRecordIdRequest.newBuilder()
+                .setStreamName(streamName)
+                .setShardId(shardId)
+                .build()).tailRecordId
+            GrpcUtils.recordIdFromGrpc(recordId)
+        }
+    }
+
     private suspend fun lookupSubscriptionServerUrl(subscriptionId: String?): String {
         return unaryCallCoroutine {
             val req: LookupSubscriptionRequest =

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

@@ -27,7 +27,9 @@ class StreamShardReaderKtImpl(
     private val client: HStreamClientKtImpl,
     private val streamName: String,
     private val shardId: Long,
-    private val shardOffset: StreamShardOffset,
+    private val from: StreamShardOffset,
+    private val maxReadBatches: Long,
+    private val until: StreamShardOffset?,
     private val receiver: StreamShardReaderReceiver?,
     private val batchReceiver: StreamShardReaderBatchReceiver?,
 ) : AbstractService(), StreamShardReader {
@@ -44,10 +46,15 @@ class StreamShardReaderKtImpl(
                     .setReaderId(readerName).build()
                 val lookupShardReaderResp = client.unaryCallBlocked { it.lookupShardReader(lookupShardReaderRequest) }
                 val serverUrl = lookupShardReaderResp.serverNode.host + ":" + lookupShardReaderResp.serverNode.port
-                val respFlow = client.getCoroutineStub(serverUrl).readShardStream(
-                    ReadShardStreamRequest.newBuilder().setReaderId(readerName).setShardId(shardId)
-                        .setShardOffset(GrpcUtils.streamShardOffsetToGrpc(shardOffset)).build()
-                )
+                val readerBuilder = ReadShardStreamRequest.newBuilder()
+                    .setReaderId(readerName)
+                    .setShardId(shardId)
+                    .setFrom(GrpcUtils.streamShardOffsetToGrpc(from))
+                    .setMaxReadBatches(maxReadBatches)
+                if (until != null) {
+                    readerBuilder.until = GrpcUtils.streamShardOffsetToGrpc(until)
+                }
+                val respFlow = client.getCoroutineStub(serverUrl).readShardStream(readerBuilder.build())
                 notifyStarted()
                 readerScope.launch {
                     respFlow.collect {