feat: introduce grpc support and code generation and define writeByGrpc interface

.github/workflows/pr-title-lint.yml

@@ -0,0 +1,29 @@
+# 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.
+
+name: pr title lint
+
+on:
+  pull_request:
+    types: [opened, edited, synchronize, reopened]
+    branches:
+      - main
+
+jobs:
+  runner:
+    name: pr title lint
+    runs-on: ubuntu-latest
+    steps:
+      - name: PR Title Lint
+        uses: openGemini/pr-title-checker@v1.0.2

.typos.toml

@@ -0,0 +1,17 @@
+# 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.
+
+[default.extend-words]
+# abbr
+"vertx" = "vertx"

ci/spotbugs/exclude.xml

@@ -41,5 +41,8 @@
     <Match>
         <Bug pattern="URF_UNREAD_PUBLIC_OR_PROTECTED_FIELD"/>
     </Match>
-
+    <!-- Exclude gRPC auto-generated classes -->
+    <Match>
+        <Package name="io.opengemini.client.proto"/>
+    </Match>
 </FindBugsFilter>

opengemini-client-api/pom.xml

@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.opengemini</groupId>
         <artifactId>opengemini-client-parent</artifactId>
-        <version>0.3.1</version>
+        <version>0.3.2</version>
     </parent>
 
     <artifactId>opengemini-client-api</artifactId>

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

@@ -108,6 +108,14 @@ public interface OpenGeminiAsyncClient extends AutoCloseable {
      */
     CompletableFuture<Void> write(String database, String retentionPolicy, List<Point> points);
 
+    /**
+     * 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);
+
     /**
      * Ping the OpenGemini server
      */

opengemini-client-common/pom.xml

@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.opengemini</groupId>
         <artifactId>opengemini-client-parent</artifactId>
-        <version>0.3.1</version>
+        <version>0.3.2</version>
     </parent>
 
     <artifactId>opengemini-client-common</artifactId>

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

@@ -154,6 +154,14 @@ public CompletableFuture<Void> write(String database, String retentionPolicy, Li
         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);
+    }
+
     /**
      * {@inheritDoc}
      */
@@ -187,6 +195,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-common/src/main/java/io/opengemini/client/common/ResultMapper.java

@@ -18,22 +18,25 @@
 
 import io.opengemini.client.api.QueryResult;
 import io.opengemini.client.api.RetentionPolicy;
+import io.opengemini.client.api.Series;
+import io.opengemini.client.api.SeriesResult;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 public class ResultMapper {
 
     public static List<String> toDatabases(QueryResult result) {
-        return result.getResults()
-                .get(0)
-                .getSeries()
-                .get(0)
-                .getValues()
-                .stream()
-                .map(x -> String.valueOf(x.get(0)))
-                .collect(Collectors.toList());
+        return Optional.ofNullable(result.getResults())
+                .map(list -> list.get(0))
+                .map(SeriesResult::getSeries)
+                .map(list -> list.get(0))
+                .map(Series::getValues)
+                .map(list -> list.stream().map(x -> String.valueOf(x.get(0))).collect(Collectors.toList()))
+                .orElse(Collections.emptyList());
     }
 
     public static List<RetentionPolicy> toRetentionPolicies(QueryResult result) {

opengemini-client-reactor/pom.xml

@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.opengemini</groupId>
         <artifactId>opengemini-client-parent</artifactId>
-        <version>0.3.1</version>
+        <version>0.3.2</version>
     </parent>
 
     <artifactId>opengemini-client-reactor</artifactId>

opengemini-client/pom.xml

@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.opengemini</groupId>
         <artifactId>opengemini-client-parent</artifactId>
-        <version>0.3.1</version>
+        <version>0.3.2</version>
     </parent>
 
     <artifactId>opengemini-client</artifactId>
@@ -54,6 +54,23 @@
             <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.version}</version>
+        </dependency>
     </dependencies>
 
 </project>

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

@@ -25,6 +25,7 @@
 import io.opengemini.client.api.AuthType;
 import io.opengemini.client.api.Configuration;
 import io.opengemini.client.api.OpenGeminiException;
+import io.opengemini.client.api.Point;
 import io.opengemini.client.api.Pong;
 import io.opengemini.client.api.Query;
 import io.opengemini.client.api.QueryResult;
@@ -35,6 +36,7 @@
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
@@ -90,6 +92,11 @@ protected CompletableFuture<Void> executeWrite(String database, String retention
         return post(writeUrl, lineProtocol).thenCompose(response -> convertResponse(response, Void.class));
     }
 
+    @Override
+    protected CompletableFuture<Void> executeWriteByGrpc(String database, List<Point> points) {
+        return null;
+    }
+
     /**
      * Execute a ping call with java HttpClient.
      */

opengemini-client/src/main/proto/write.proto

@@ -0,0 +1,61 @@
+syntax = "proto3";
+package proto;
+option java_multiple_files = true;
+option java_package = "io.opengemini.client.proto";
+option java_outer_classname = "WriteProto";
+
+// WriteService represents a openGemini RPC write service.
+service WriteService {
+  // Write writes the given records to the specified database and retention policy.
+  rpc Write (WriteRequest) returns (WriteResponse) {}
+  // Ping is used to check if the server is alive
+  rpc Ping(PingRequest) returns (PingResponse) {}
+}
+
+message WriteRequest {
+  uint32 version = 1;
+  string database = 2;
+  string retention_policy = 3;
+  string username = 4;
+  string password = 5;
+  repeated Record records = 6;
+}
+
+message WriteResponse {
+  ResponseCode code = 1;
+}
+
+message Record {
+  string measurement = 1;
+  int64 min_time = 2;
+  int64 max_time = 3;
+  CompressMethod compress_method = 4;
+  bytes block = 5;
+}
+
+enum CompressMethod {
+  UNCOMPRESSED = 0;
+  LZ4_FAST = 1;
+  ZSTD_FAST = 2;
+  SNAPPY = 3;
+}
+
+enum ResponseCode {
+  Success = 0;
+  Partial = 1;
+  Failed = 2;
+}
+
+message PingRequest {
+  string client_id = 1;
+}
+
+enum ServerStatus {
+  Up = 0;
+  Down = 1;
+  Unknown = 99;
+}
+
+message PingResponse {
+  ServerStatus status = 1;
+}