feat: add getSubscription (#162)

Author: s12f

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

@@ -0,0 +1,42 @@
+package io.hstream;
+
+import java.util.List;
+
+public class GetSubscriptionResponse {
+  Subscription subscription;
+  List<SubscriptionOffset> offsets;
+
+  public Subscription getSubscription() {
+    return subscription;
+  }
+
+  public List<SubscriptionOffset> getOffsets() {
+    return offsets;
+  }
+
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  public static final class Builder {
+    private Subscription subscription;
+    private List<SubscriptionOffset> offsets;
+
+    public Builder subscription(Subscription subscription) {
+      this.subscription = subscription;
+      return this;
+    }
+
+    public Builder offsets(List<SubscriptionOffset> offsets) {
+      this.offsets = offsets;
+      return this;
+    }
+
+    public GetSubscriptionResponse build() {
+      GetSubscriptionResponse getSubscriptionResponse = new GetSubscriptionResponse();
+      getSubscriptionResponse.offsets = this.offsets;
+      getSubscriptionResponse.subscription = this.subscription;
+      return getSubscriptionResponse;
+    }
+  }
+}

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

@@ -110,6 +110,8 @@ static HStreamClientBuilder builder() {
    */
   List<Subscription> listSubscriptions();
 
+  GetSubscriptionResponse getSubscription(String subscriptionId);
+
   /**
    * Delete the specified subscription with subscriptionId.
    *

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

@@ -0,0 +1,40 @@
+package io.hstream;
+
+public class SubscriptionOffset {
+  long shardId;
+  long batchId;
+
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  public long getShardId() {
+    return shardId;
+  }
+
+  public long getBatchId() {
+    return batchId;
+  }
+
+  public static final class Builder {
+    private long shardId;
+    private long batchId;
+
+    public Builder withShardId(long shardId) {
+      this.shardId = shardId;
+      return this;
+    }
+
+    public Builder withBatchId(long batchId) {
+      this.batchId = batchId;
+      return this;
+    }
+
+    public SubscriptionOffset build() {
+      SubscriptionOffset subscriptionOffset = new SubscriptionOffset();
+      subscriptionOffset.shardId = this.shardId;
+      subscriptionOffset.batchId = this.batchId;
+      return subscriptionOffset;
+    }
+  }
+}

client/src/main/java/io/hstream/util/GrpcUtils.java

@@ -7,6 +7,7 @@
 import io.hstream.internal.SpecialOffset;
 import io.hstream.internal.TaskStatusPB;
 import java.time.Instant;
+import java.util.stream.Collectors;
 
 /**
  * A class of utility functions to convert between the GRPC generated classes and the custom classes
@@ -206,4 +207,23 @@ public static ConsumerInformation consumerInformationFromGrpc(
         .userAgent(consumer.getUserAgent())
         .build();
   }
+
+  public static GetSubscriptionResponse GetSubscriptionResponseFromGrpc(
+      io.hstream.internal.GetSubscriptionResponse response) {
+    return GetSubscriptionResponse.newBuilder()
+        .subscription(subscriptionFromGrpc(response.getSubscription()))
+        .offsets(
+            response.getOffsetsList().stream()
+                .map(GrpcUtils::subscriptionOffsetFromGrpc)
+                .collect(Collectors.toList()))
+        .build();
+  }
+
+  public static SubscriptionOffset subscriptionOffsetFromGrpc(
+      io.hstream.internal.SubscriptionOffset offset) {
+    return SubscriptionOffset.newBuilder()
+        .withShardId(offset.getShardId())
+        .withBatchId(offset.getBatchId())
+        .build();
+  }
 }

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

@@ -6,6 +6,7 @@ import io.hstream.BufferedProducerBuilder
 import io.hstream.Cluster
 import io.hstream.ConsumerBuilder
 import io.hstream.ConsumerInformation
+import io.hstream.GetSubscriptionResponse
 import io.hstream.HStreamClient
 import io.hstream.ProducerBuilder
 import io.hstream.Query
@@ -22,6 +23,7 @@ import io.hstream.internal.DeleteStreamRequest
 import io.hstream.internal.DeleteSubscriptionRequest
 import io.hstream.internal.DeleteViewRequest
 import io.hstream.internal.GetQueryRequest
+import io.hstream.internal.GetSubscriptionRequest
 import io.hstream.internal.GetViewRequest
 import io.hstream.internal.HStreamApiGrpcKt
 import io.hstream.internal.ListConsumersRequest
@@ -187,6 +189,14 @@ class HStreamClientKtImpl(bootstrapServerUrls: List<String>, credentials: Channe
             )
         }
     }
+
+    override fun getSubscription(subscriptionId: String?): GetSubscriptionResponse {
+        return unaryCallBlocked {
+            val response = it.getSubscription(GetSubscriptionRequest.newBuilder().setId(subscriptionId).build())
+            GrpcUtils.GetSubscriptionResponseFromGrpc(response)
+        }
+    }
+
     override fun deleteSubscription(subscriptionId: String?) {
         deleteSubscription(subscriptionId, false)
     }