feat: add StreamShardReaderBatchReceiver, and add lookup for query/view apis (#209)

Author: s12f

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

@@ -0,0 +1,8 @@
+package io.hstream;
+
+import java.util.List;
+
+public interface StreamShardReaderBatchReceiver {
+
+  void process(List<ReceivedRecord> receivedRecords);
+}

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

@@ -9,6 +9,7 @@ public interface StreamShardReaderBuilder {
   StreamShardReaderBuilder shardOffset(StreamShardOffset shardOffset);
 
   StreamShardReaderBuilder receiver(StreamShardReaderReceiver streamShardReaderReceiver);
+  StreamShardReaderBuilder batchReceiver(StreamShardReaderBatchReceiver streamShardReaderReceiver);
 
   StreamShardReader build();
 }

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

@@ -3,10 +3,7 @@
 import static com.google.common.base.Preconditions.*;
 import static com.google.common.base.Preconditions.checkArgument;
 
-import io.hstream.StreamShardOffset;
-import io.hstream.StreamShardReader;
-import io.hstream.StreamShardReaderBuilder;
-import io.hstream.StreamShardReaderReceiver;
+import io.hstream.*;
 
 public class StreamShardReaderBuilderImpl implements StreamShardReaderBuilder {
 
@@ -16,6 +13,7 @@ public class StreamShardReaderBuilderImpl implements StreamShardReaderBuilder {
   private StreamShardOffset shardOffset;
 
   private StreamShardReaderReceiver receiver;
+  private StreamShardReaderBatchReceiver batchReceiver;
 
   public StreamShardReaderBuilderImpl(HStreamClientKtImpl client) {
     this.client = client;
@@ -45,14 +43,21 @@ public StreamShardReaderBuilder receiver(StreamShardReaderReceiver streamShardRe
     return this;
   }
 
+  @Override
+  public StreamShardReaderBuilder batchReceiver(StreamShardReaderBatchReceiver batchReceiver) {
+    this.batchReceiver = batchReceiver;
+    return this;
+  }
+
   @Override
   public StreamShardReader build() {
     checkNotNull(client);
     checkArgument(streamName != null, "StreamShardReaderBuilder: `streamName` should not be null");
     checkArgument(shardId > 0, "StreamShardReaderBuilder: `shardId` error");
     checkArgument(
         shardOffset != null, "StreamShardReaderBuilder: `shardOffset` should not be null");
-    checkArgument(receiver != null, "StreamShardReaderBuilder: `receiver` should not be null");
-    return new StreamShardReaderKtImpl(client, streamName, shardId, shardOffset, receiver);
+    checkArgument(receiver != null || batchReceiver != null,
+            "StreamShardReaderBuilder: `receiver` should not be both null");
+    return new StreamShardReaderKtImpl(client, streamName, shardId, shardOffset, receiver, batchReceiver);
   }
 }

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

@@ -310,7 +310,7 @@ class HStreamClientKtImpl(
 
     override fun terminateQuery(name: String?) {
         checkNotNull(name)
-        unaryCallBlocked {
+        unaryCallBlockedWithLookup(ResourceType.ResQuery, name) {
             it.terminateQuery(TerminateQueryRequest.newBuilder().setQueryId(name).build())
         }
     }
@@ -325,14 +325,14 @@ class HStreamClientKtImpl(
     }
 
     override fun getView(name: String?): View {
-        return unaryCallBlocked {
+        return unaryCallBlockedWithLookup(ResourceType.ResView, name) {
             val result = it.getView(GetViewRequest.newBuilder().setViewId(name).build())
             GrpcUtils.viewFromInternal(result)
         }
     }
 
     override fun deleteView(name: String?) {
-        unaryCallBlocked {
+        return unaryCallBlockedWithLookup(ResourceType.ResView, name) {
             it.deleteView(DeleteViewRequest.newBuilder().setViewId(name).build())
         }
     }

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

@@ -7,6 +7,7 @@ import io.hstream.ReceivedRecord
 import io.hstream.Record
 import io.hstream.StreamShardOffset
 import io.hstream.StreamShardReader
+import io.hstream.StreamShardReaderBatchReceiver
 import io.hstream.StreamShardReaderReceiver
 import io.hstream.internal.LookupShardReaderRequest
 import io.hstream.internal.ReadShardStreamRequest
@@ -20,13 +21,15 @@ import org.slf4j.LoggerFactory
 import java.time.Instant
 import java.util.UUID
 import java.util.concurrent.Executors
+import kotlin.streams.toList
 
 class StreamShardReaderKtImpl(
     private val client: HStreamClientKtImpl,
     private val streamName: String,
     private val shardId: Long,
     private val shardOffset: StreamShardOffset,
-    private val receiver: StreamShardReaderReceiver,
+    private val receiver: StreamShardReaderReceiver?,
+    private val batchReceiver: StreamShardReaderBatchReceiver?,
 ) : AbstractService(), StreamShardReader {
     private val readerScope = CoroutineScope(Dispatchers.IO)
     private val readerName: String = UUID.randomUUID().toString()
@@ -76,19 +79,34 @@ class StreamShardReaderKtImpl(
             val receivedHStreamRecords = RecordUtils.decompress(receivedRecord)
             val createdTimestamp = receivedRecord.record.publishTime
             val createdTime = Instant.ofEpochSecond(createdTimestamp.seconds, createdTimestamp.nanos.toLong())
-            for (receivedHStreamRecord in receivedHStreamRecords) {
 
+            if (batchReceiver != null) {
                 executorService.submit {
                     if (!isRunning) {
                         return@submit
                     }
 
                     try {
-                        receiver.process(toReceivedRecord(receivedHStreamRecord, createdTime))
+                        val records = receivedHStreamRecords.stream().map { toReceivedRecord(it, createdTime) }.toList()
+                        batchReceiver.process(records)
                     } catch (e: Exception) {
                         notifyFailed(e)
                     }
                 }
+            } else {
+                for (receivedHStreamRecord in receivedHStreamRecords) {
+                    executorService.submit {
+                        if (!isRunning) {
+                            return@submit
+                        }
+
+                        try {
+                            receiver!!.process(toReceivedRecord(receivedHStreamRecord, createdTime))
+                        } catch (e: Exception) {
+                            notifyFailed(e)
+                        }
+                    }
+                }
             }
         }
     }