feat: support RPC protocol write

opengemini-client-api/src/main/java/io/opengemini/client/api/Configuration.java

@@ -40,4 +40,6 @@ public class Configuration {
     boolean gzipEnabled;
 
     HttpClientConfig httpConfig;
+
+    GrpcConfig rpcConfig;
 }

opengemini-client-api/src/main/java/io/opengemini/client/api/GrpcConfig.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2024 openGemini Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.opengemini.client.api;
+
+import lombok.Builder;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+
+@Getter
+@Setter
+@ToString
+@Builder
+public class GrpcConfig {
+    private String host;
+    private Integer port;
+    private String username;
+    private String password;
+    private boolean useSSL = false;
+    private boolean waitForReady = true;
+    private String caCertPath;
+    private String clientCertPath;
+    private String clientKeyPath;
+}

opengemini-client-api/src/main/java/io/opengemini/client/api/OpenGeminiAsyncClient.java

@@ -99,6 +99,14 @@ public interface OpenGeminiAsyncClient extends AutoCloseable {
      */
     CompletableFuture<Void> write(String database, String retentionPolicy, Point point);
 
+    /**
+     * Writing via GRPC points to the database.
+     *
+     * @param database the name of the database.
+     * @param points   the points to write.
+     */
+    CompletableFuture<Void> writeByGrpc(String database, List<Point> points);
+
     /**
      * Write points to the database.
      *

opengemini-client-common/src/main/java/io/opengemini/client/common/BaseAsyncClient.java

@@ -139,6 +139,22 @@ public CompletableFuture<Void> write(String database, String retentionPolicy, Li
         if (points.isEmpty()) {
             return CompletableFuture.completedFuture(null);
         }
+        String body = toLineProtocol(points);
+        if (StringUtils.isEmpty(body)) {
+            return CompletableFuture.completedFuture(null);
+        }
+        return executeWrite(database, retentionPolicy, body);
+    }
+
+    @Override
+    public CompletableFuture<Void> writeByGrpc(String database, List<Point> points) {
+        if (points.isEmpty()) {
+            return CompletableFuture.completedFuture(null);
+        }
+        return executeWriteByGrpc(database, points);
+    }
+
+    private static String toLineProtocol(List<Point> points) {
         StringJoiner sj = new StringJoiner("\n");
         for (Point point : points) {
             String lineProtocol = point.lineProtocol();
@@ -147,11 +163,7 @@ public CompletableFuture<Void> write(String database, String retentionPolicy, Li
             }
             sj.add(lineProtocol);
         }
-        String body = sj.toString();
-        if (StringUtils.isEmpty(body)) {
-            return CompletableFuture.completedFuture(null);
-        }
-        return executeWrite(database, retentionPolicy, body);
+        return sj.toString();
     }
 
     /**
@@ -187,6 +199,15 @@ protected abstract CompletableFuture<Void> executeWrite(String database,
                                                             String retentionPolicy,
                                                             String lineProtocol);
 
+    /**
+     * The implementation class needs to implement this method to execute a write operation via an RPC call.
+     *
+     * @param database     the name of the database.
+     * @param points       the points to write.
+     */
+    protected abstract CompletableFuture<Void> executeWriteByGrpc(String database,
+                                                                  List<Point> points);
+
     /**
      * The implementation class needs to implement this method to execute a ping call.
      */

opengemini-client/pom.xml

@@ -54,6 +54,41 @@
             <version>${okhttp.version}</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-netty</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-stub</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-protobuf</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-grpc</artifactId>
+            <version>${vertx-grpc.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-testing</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-junit5</artifactId>
+            <version>${vertx-grpc.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <version>${junit.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

opengemini-client/src/main/java/io/opengemini/client/impl/OpenGeminiClient.java

@@ -21,20 +21,16 @@
 import io.github.openfacade.http.HttpClientConfig;
 import io.github.openfacade.http.HttpClientFactory;
 import io.github.openfacade.http.HttpResponse;
-import io.opengemini.client.api.AuthConfig;
-import io.opengemini.client.api.AuthType;
-import io.opengemini.client.api.Configuration;
-import io.opengemini.client.api.OpenGeminiException;
-import io.opengemini.client.api.Pong;
-import io.opengemini.client.api.Query;
-import io.opengemini.client.api.QueryResult;
+import io.opengemini.client.api.*;
 import io.opengemini.client.common.BaseAsyncClient;
 import io.opengemini.client.common.HeaderConst;
 import io.opengemini.client.common.JacksonService;
+import io.opengemini.client.impl.grpc.GrpcClient;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
@@ -43,6 +39,8 @@ public class OpenGeminiClient extends BaseAsyncClient {
 
     private final HttpClient client;
 
+    private final GrpcClient grpcClient;
+
     public OpenGeminiClient(@NotNull Configuration conf) {
         super(conf);
         this.conf = conf;
@@ -53,6 +51,12 @@ public OpenGeminiClient(@NotNull Configuration conf) {
                     new BasicAuthRequestFilter(authConfig.getUsername(), String.valueOf(authConfig.getPassword())));
         }
         this.client = HttpClientFactory.createHttpClient(httpConfig);
+
+        if (conf.getRpcConfig() != null) {
+            grpcClient = GrpcClient.create(conf.getRpcConfig());
+        } else {
+            grpcClient = null;
+        }
     }
 
     /**
@@ -90,6 +94,20 @@ protected CompletableFuture<Void> executeWrite(String database, String retention
         return post(writeUrl, lineProtocol).thenCompose(response -> convertResponse(response, Void.class));
     }
 
+    /**
+     * Execute a write call with RPC client
+     *
+     * @param database the name of the database.
+     * @param points   the points to write.
+     */
+    @Override
+    protected CompletableFuture<Void> executeWriteByGrpc(String database, List<Point> points) {
+        if (grpcClient == null) {
+            throw new IllegalStateException("RPC client not initialized");
+        }
+        return grpcClient.getWriteClient().writeRows(database, points);
+    }
+
     /**
      * Execute a ping call with java HttpClient.
      */
@@ -127,12 +145,15 @@ public CompletableFuture<HttpResponse> get(String url) {
 
     public CompletableFuture<HttpResponse> post(String url, String body) {
         return client.post(buildUriWithPrefix(url), body == null ? new byte[0] : body.getBytes(StandardCharsets.UTF_8),
-                           headers);
+                headers);
     }
 
     @Override
     public void close() throws IOException {
         this.client.close();
+        if (this.grpcClient != null) {
+            this.grpcClient.close();
+        }
     }
 
     @Override

opengemini-client/src/main/java/io/opengemini/client/impl/grpc/GrpcClient.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2024 openGemini Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.opengemini.client.impl.grpc;
+
+import io.opengemini.client.api.GrpcConfig;
+import io.opengemini.client.impl.grpc.service.WriteService;
+import io.opengemini.client.impl.grpc.support.RpcClientSupplier;
+
+public class GrpcClient {
+    private final GrpcConfig config;
+    private final RpcClientConnectionManager connectionManager;
+
+    private final RpcClientSupplier<WriteService> writeClient;
+
+    public static GrpcClient create(final GrpcConfig config) {
+        return new GrpcClient(config);
+    }
+
+    private GrpcClient(GrpcConfig config) {
+        this.config = config;
+        this.connectionManager = new RpcClientConnectionManager(config);
+        this.writeClient = new RpcClientSupplier<>(() -> {
+            try {
+                return new WriteService(this.connectionManager);
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        });
+    }
+
+
+    public WriteService getWriteClient() {
+        return writeClient.get();
+    }
+
+    public synchronized void close() {
+        writeClient.close();
+        connectionManager.close();
+    }
+}