feat: listConsumers (#161)

Author: s12f

client/build.gradle.kts

@@ -183,6 +183,13 @@ tasks.withType<Javadoc> {
     exclude("io/hstream/impl/**", "io/hstream/util/**")
 }
 
+val clientVersion = version
+tasks.withType<Jar> {
+    manifest {
+        attributes["Implementation-Version"] = clientVersion
+    }
+}
+
 spotless {
     java {
         googleJavaFormat()

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

@@ -0,0 +1,55 @@
+package io.hstream;
+
+public class ConsumerInformation {
+  String name;
+  String uri;
+
+  String userAgent;
+
+  public String getName() {
+    return name;
+  }
+
+  public String getUri() {
+    return uri;
+  }
+
+  public String getUserAgent() {
+    return userAgent;
+  }
+
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  public static final class Builder {
+    private String name;
+    private String uri;
+    private String userAgent;
+
+    private Builder() {}
+
+    public Builder name(String name) {
+      this.name = name;
+      return this;
+    }
+
+    public Builder uri(String uri) {
+      this.uri = uri;
+      return this;
+    }
+
+    public Builder userAgent(String userAgent) {
+      this.userAgent = userAgent;
+      return this;
+    }
+
+    public ConsumerInformation build() {
+      ConsumerInformation consumerInformation = new ConsumerInformation();
+      consumerInformation.uri = this.uri;
+      consumerInformation.name = this.name;
+      consumerInformation.userAgent = this.userAgent;
+      return consumerInformation;
+    }
+  }
+}

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

@@ -142,4 +142,6 @@ static HStreamClientBuilder builder() {
   View getView(String name);
 
   void deleteView(String name);
+
+  List<ConsumerInformation> listConsumers(String subscriptionId);
 }

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

@@ -14,6 +14,8 @@ public class ChannelProvider implements Closeable {
   private ChannelCredentials credentials;
 
   private final ConcurrentHashMap<String, ManagedChannel> provider;
+  String userAgent =
+      "hstreamdb-java/" + ChannelProvider.class.getPackage().getImplementationVersion();
 
   public ChannelProvider(int size) {
     provider = new ConcurrentHashMap<>(size);
@@ -35,13 +37,15 @@ public ManagedChannel get(String serverUrl) {
           url ->
               ManagedChannelBuilder.forTarget(url)
                   .usePlaintext()
+                  .userAgent(userAgent)
                   .executor(MoreExecutors.directExecutor())
                   .build());
     }
     return provider.computeIfAbsent(
         serverUrl,
         url ->
             Grpc.newChannelBuilder(url, credentials)
+                .userAgent(userAgent)
                 .executor(MoreExecutors.directExecutor())
                 .build());
   }

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

@@ -77,9 +77,10 @@ public HStreamClient build() {
   private List<String> parseServerUrls(String url) {
     var prefix = "hstream://";
     String uriStr = url.strip();
-    if (uriStr.startsWith(prefix)) {
-      uriStr = uriStr.substring(prefix.length());
+    if (!uriStr.startsWith(prefix)) {
+      throw new HStreamDBClientException(
+          "incorrect serviceUrl:" + uriStr + " (correct example: hstream://127.0.0.1:6570)");
     }
-    return List.of(uriStr.split(","));
+    return List.of(uriStr.substring(prefix.length()).split(","));
   }
 }

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

@@ -197,4 +197,13 @@ public static TaskStatus taskStatusFromInternal(TaskStatusPB statusPB) {
         throw new IllegalArgumentException("Unknown task status: " + statusPB);
     }
   }
+
+  public static ConsumerInformation consumerInformationFromGrpc(
+      io.hstream.internal.Consumer consumer) {
+    return ConsumerInformation.newBuilder()
+        .name(consumer.getName())
+        .uri(consumer.getUri())
+        .userAgent(consumer.getUserAgent())
+        .build();
+  }
 }

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

@@ -5,6 +5,7 @@ import io.grpc.ChannelCredentials
 import io.hstream.BufferedProducerBuilder
 import io.hstream.Cluster
 import io.hstream.ConsumerBuilder
+import io.hstream.ConsumerInformation
 import io.hstream.HStreamClient
 import io.hstream.ProducerBuilder
 import io.hstream.Query
@@ -23,6 +24,7 @@ import io.hstream.internal.DeleteViewRequest
 import io.hstream.internal.GetQueryRequest
 import io.hstream.internal.GetViewRequest
 import io.hstream.internal.HStreamApiGrpcKt
+import io.hstream.internal.ListConsumersRequest
 import io.hstream.internal.ListQueriesRequest
 import io.hstream.internal.ListShardsRequest
 import io.hstream.internal.ListStreamsRequest
@@ -262,6 +264,15 @@ class HStreamClientKtImpl(bootstrapServerUrls: List<String>, credentials: Channe
         }
     }
 
+    override fun listConsumers(subscriptionId: String?): List<ConsumerInformation> {
+        return runBlocking {
+            val serverUrl = lookupSubscriptionServerUrl(subscriptionId)
+            val stub = HStreamApiGrpcKt.HStreamApiCoroutineStub(channelProvider.get(serverUrl))
+            stub.listConsumers(ListConsumersRequest.newBuilder().setSubscriptionId(subscriptionId).build())
+                .consumersList.stream().map(GrpcUtils::consumerInformationFromGrpc).toList()
+        }
+    }
+
     private final suspend fun lookupSubscriptionServerUrl(subscriptionId: String?): String {
         return unaryCallCoroutine {
             val req: LookupSubscriptionRequest =