From 1e0caa3a5375b76b3bf3f15245b449abdd63d11b Mon Sep 17 00:00:00 2001 From: weiping-code <54316849+weiping-code@users.noreply.github.com> Date: Wed, 22 Apr 2026 18:13:48 +0800 Subject: [PATCH 1/4] feat: add gRPC write support with binary encoding Signed-off-by: weiping-code <54316849+weiping-code@users.noreply.github.com> --- .gitignore | 1 + opengemini-client-api/pom.xml | 17 ++ .../io/opengemini/client/api/TlsConfig.java | 58 ++--- .../client/api/grpc/CompressMethod.java | 43 ++++ .../client/api/grpc/GrpcConfig.java | 61 +++++ .../client/api/grpc/ResponseCode.java | 45 ++++ .../client/api/grpc/ServerStatus.java | 45 ++++ .../java/io/opengemini/proto/PingRequest.java | 69 +++++ .../io/opengemini/proto/PingResponse.java | 68 +++++ .../main/java/io/opengemini/proto/Record.java | 180 +++++++++++++ .../io/opengemini/proto/WriteRequest.java | 183 +++++++++++++ .../io/opengemini/proto/WriteResponse.java | 68 +++++ .../src/main/proto/write.proto | 59 +++++ .../impl/grpc/OpenGeminiGrpcClient.java | 35 +++ .../grpc/OpenGeminiGrpcClientFactory.java | 29 +++ .../impl/grpc/OpenGeminiGrpcClientImpl.java | 171 +++++++++++++ .../client/impl/grpc/record/BytesBuilder.java | 242 ++++++++++++++++++ .../client/impl/grpc/record/ColVal.java | 178 +++++++++++++ .../client/impl/grpc/record/Field.java | 57 +++++ .../client/impl/grpc/record/FieldType.java | 75 ++++++ .../client/impl/grpc/record/Record.java | 227 ++++++++++++++++ pom.xml | 10 + ...ot.autoconfigure.AutoConfiguration.imports | 1 + 23 files changed, 1893 insertions(+), 29 deletions(-) create mode 100644 opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/CompressMethod.java create mode 100644 opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/GrpcConfig.java create mode 100644 opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ResponseCode.java create mode 100644 opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ServerStatus.java create mode 100644 opengemini-client-api/src/main/java/io/opengemini/proto/PingRequest.java create mode 100644 opengemini-client-api/src/main/java/io/opengemini/proto/PingResponse.java create mode 100644 opengemini-client-api/src/main/java/io/opengemini/proto/Record.java create mode 100644 opengemini-client-api/src/main/java/io/opengemini/proto/WriteRequest.java create mode 100644 opengemini-client-api/src/main/java/io/opengemini/proto/WriteResponse.java create mode 100755 opengemini-client-api/src/main/proto/write.proto create mode 100644 opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClient.java create mode 100644 opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientFactory.java create mode 100644 opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientImpl.java create mode 100644 opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/BytesBuilder.java create mode 100644 opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/ColVal.java create mode 100644 opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Field.java create mode 100644 opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/FieldType.java create mode 100644 opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Record.java diff --git a/.gitignore b/.gitignore index 6e99d4ef..8b38d3e7 100644 --- a/.gitignore +++ b/.gitignore @@ -24,3 +24,4 @@ dependency-reduced-pom.xml logs/ *.xml.releaseBackup +.go \ No newline at end of file diff --git a/opengemini-client-api/pom.xml b/opengemini-client-api/pom.xml index 27a76f5e..1e4989ff 100644 --- a/opengemini-client-api/pom.xml +++ b/opengemini-client-api/pom.xml @@ -36,5 +36,22 @@ annotations ${annotations.version} + + + io.grpc + grpc-api + + + io.grpc + grpc-stub + + + io.grpc + grpc-protobuf + + + com.google.protobuf + protobuf-java + diff --git a/opengemini-client-api/src/main/java/io/opengemini/client/api/TlsConfig.java b/opengemini-client-api/src/main/java/io/opengemini/client/api/TlsConfig.java index 94fd6fe3..e70dbfcd 100644 --- a/opengemini-client-api/src/main/java/io/opengemini/client/api/TlsConfig.java +++ b/opengemini-client-api/src/main/java/io/opengemini/client/api/TlsConfig.java @@ -14,32 +14,32 @@ * limitations under the License. */ -//package io.opengemini.client.api; -// -//import lombok.AllArgsConstructor; -//import lombok.NoArgsConstructor; -//import lombok.Setter; -//import lombok.ToString; -// -//@Setter -//@NoArgsConstructor -//@AllArgsConstructor -//public class TlsConfig { -// public String keyStorePath; -// -// @ToString.Exclude -// public char[] keyStorePassword; -// -// public String trustStorePath; -// -// @ToString.Exclude -// public char[] trustStorePassword; -// -// public boolean verifyDisabled; -// -// public boolean hostnameVerifyDisabled; -// -// public String[] versions; -// -// public String[] cipherSuites; -//} +package io.opengemini.client.api; + +import lombok.AllArgsConstructor; +import lombok.NoArgsConstructor; +import lombok.Setter; +import lombok.ToString; + +@Setter +@NoArgsConstructor +@AllArgsConstructor +public class TlsConfig { + public String keyStorePath; + + @ToString.Exclude + public char[] keyStorePassword; + + public String trustStorePath; + + @ToString.Exclude + public char[] trustStorePassword; + + public boolean verifyDisabled; + + public boolean hostnameVerifyDisabled; + + public String[] versions; + + public String[] cipherSuites; +} diff --git a/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/CompressMethod.java b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/CompressMethod.java new file mode 100644 index 00000000..8d4a9e20 --- /dev/null +++ b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/CompressMethod.java @@ -0,0 +1,43 @@ +/* + * 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.grpc; + +public enum CompressMethod { + UNCOMPRESSED(0), + LZ4_FAST(1), + ZSTD_FAST(2), + SNAPPY(3); + + private final int value; + + CompressMethod(int value) { + this.value = value; + } + + public int getValue() { + return value; + } + + public static CompressMethod forNumber(int value) { + for (CompressMethod method : values()) { + if (method.value == value) { + return method; + } + } + return UNCOMPRESSED; + } +} \ No newline at end of file diff --git a/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/GrpcConfig.java b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/GrpcConfig.java new file mode 100644 index 00000000..0478e579 --- /dev/null +++ b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/GrpcConfig.java @@ -0,0 +1,61 @@ +/* + * 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.grpc; + +import io.opengemini.client.api.AuthConfig; +import io.opengemini.client.api.Address; +import io.opengemini.client.api.TlsConfig; +import lombok.Getter; +import lombok.Setter; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; + +/** + * GrpcConfig represents the configuration information for write service by gRPC. + */ +@Getter +@Setter +public class GrpcConfig { + + /** + * Addresses Configure the service endpoints for the openGemini grpc write service. + * This parameter is required. + */ + private List
addresses = new ArrayList<>(); + + /** + * AuthConfig configuration information for authentication. + */ + private AuthConfig authConfig; + + /** + * TlsConfig configuration information for tls. + */ + private TlsConfig tlsConfig; + + /** + * CompressMethod determines the compress method used for data transmission. + */ + private CompressMethod compressMethod = CompressMethod.UNCOMPRESSED; + + /** + * Timeout default 30s + */ + private Duration timeout = Duration.ofSeconds(30); +} \ No newline at end of file diff --git a/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ResponseCode.java b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ResponseCode.java new file mode 100644 index 00000000..c9743d61 --- /dev/null +++ b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ResponseCode.java @@ -0,0 +1,45 @@ +/* + * 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.grpc; + +/** + * ResponseCode represents the response code from gRPC write service. + */ +public enum ResponseCode { + Success(0), + Partial(1), + Failed(2); + + private final int value; + + ResponseCode(int value) { + this.value = value; + } + + public int getValue() { + return value; + } + + public static ResponseCode forNumber(int value) { + for (ResponseCode code : values()) { + if (code.value == value) { + return code; + } + } + return Success; + } +} \ No newline at end of file diff --git a/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ServerStatus.java b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ServerStatus.java new file mode 100644 index 00000000..fc0bf74e --- /dev/null +++ b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ServerStatus.java @@ -0,0 +1,45 @@ +/* + * 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.grpc; + +/** + * ServerStatus represents the server status from gRPC ping response. + */ +public enum ServerStatus { + Up(0), + Down(1), + Unknown(99); + + private final int value; + + ServerStatus(int value) { + this.value = value; + } + + public int getValue() { + return value; + } + + public static ServerStatus forNumber(int value) { + for (ServerStatus status : values()) { + if (status.value == value) { + return status; + } + } + return Unknown; + } +} \ No newline at end of file diff --git a/opengemini-client-api/src/main/java/io/opengemini/proto/PingRequest.java b/opengemini-client-api/src/main/java/io/opengemini/proto/PingRequest.java new file mode 100644 index 00000000..af862bcc --- /dev/null +++ b/opengemini-client-api/src/main/java/io/opengemini/proto/PingRequest.java @@ -0,0 +1,69 @@ +/* + * 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.proto; + +/** + * Manual proto implementation for PingRequest. + */ +public class PingRequest { + + private String clientId; + + public PingRequest() { + } + + public String getClientId() { + return clientId; + } + + public PingRequest setClientId(String clientId) { + this.clientId = clientId; + return this; + } + + public byte[] toByteArray() { + if (clientId == null || clientId.isEmpty()) { + return new byte[0]; + } + return clientId.getBytes(); + } + + public static PingRequest parseFrom(byte[] data) { + PingRequest request = new PingRequest(); + if (data != null && data.length > 0) { + request.setClientId(new String(data)); + } + return request; + } + + public static Builder newBuilder() { + return new Builder(); + } + + public static class Builder { + private final PingRequest request = new PingRequest(); + + public Builder setClientId(String clientId) { + request.clientId = clientId; + return this; + } + + public PingRequest build() { + return request; + } + } +} \ No newline at end of file diff --git a/opengemini-client-api/src/main/java/io/opengemini/proto/PingResponse.java b/opengemini-client-api/src/main/java/io/opengemini/proto/PingResponse.java new file mode 100644 index 00000000..63fd8071 --- /dev/null +++ b/opengemini-client-api/src/main/java/io/opengemini/proto/PingResponse.java @@ -0,0 +1,68 @@ +/* + * 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.proto; + +import io.opengemini.client.api.grpc.ServerStatus; + +/** + * Manual proto implementation for PingResponse. + */ +public class PingResponse { + + private ServerStatus status = ServerStatus.Up; + + public PingResponse() { + } + + public ServerStatus getStatus() { + return status; + } + + public PingResponse setStatus(ServerStatus status) { + this.status = status; + return this; + } + + public byte[] toByteArray() { + return new byte[]{(byte) status.getValue()}; + } + + public static PingResponse parseFrom(byte[] data) { + PingResponse response = new PingResponse(); + if (data != null && data.length > 0) { + response.setStatus(ServerStatus.forNumber(data[0])); + } + return response; + } + + public static Builder newBuilder() { + return new Builder(); + } + + public static class Builder { + private final PingResponse response = new PingResponse(); + + public Builder setStatus(ServerStatus status) { + response.status = status; + return this; + } + + public PingResponse build() { + return response; + } + } +} \ No newline at end of file diff --git a/opengemini-client-api/src/main/java/io/opengemini/proto/Record.java b/opengemini-client-api/src/main/java/io/opengemini/proto/Record.java new file mode 100644 index 00000000..172d7cc7 --- /dev/null +++ b/opengemini-client-api/src/main/java/io/opengemini/proto/Record.java @@ -0,0 +1,180 @@ +/* + * 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.proto; + +import io.opengemini.client.api.grpc.CompressMethod; + +/** + * Manual proto implementation for Record. + */ +public class Record { + + private String measurement; + private long minTime; + private long maxTime; + private CompressMethod compressMethod; + private byte[] block; + + public Record() { + } + + public String getMeasurement() { + return measurement; + } + + public Record setMeasurement(String measurement) { + this.measurement = measurement; + return this; + } + + public long getMinTime() { + return minTime; + } + + public Record setMinTime(long minTime) { + this.minTime = minTime; + return this; + } + + public long getMaxTime() { + return maxTime; + } + + public Record setMaxTime(long maxTime) { + this.maxTime = maxTime; + return this; + } + + public CompressMethod getCompressMethod() { + return compressMethod; + } + + public Record setCompressMethod(CompressMethod compressMethod) { + this.compressMethod = compressMethod; + return this; + } + + public byte[] getBlock() { + return block; + } + + public Record setBlock(byte[] block) { + this.block = block; + return this; + } + + public byte[] toByteArray(byte[] buf) { + buf = writeString(buf, measurement); + buf = writeInt64(buf, minTime); + buf = writeInt64(buf, maxTime); + buf = writeUint32(buf, compressMethod != null ? compressMethod.getValue() : 0); + buf = writeBytes(buf, block); + return buf; + } + + private byte[] writeString(byte[] b, String s) { + if (s == null) { + return writeUint16(b, 0); + } + byte[] strBytes = s.getBytes(); + b = writeUint16(b, strBytes.length); + byte[] result = new byte[b.length + strBytes.length]; + System.arraycopy(b, 0, result, 0, b.length); + System.arraycopy(strBytes, 0, result, b.length, strBytes.length); + return result; + } + + private byte[] writeUint16(byte[] b, int value) { + byte[] result = new byte[b.length + 2]; + System.arraycopy(b, 0, result, 0, b.length); + result[b.length] = (byte) (value >> 8); + result[b.length + 1] = (byte) value; + return result; + } + + private byte[] writeUint32(byte[] b, int value) { + byte[] result = new byte[b.length + 4]; + System.arraycopy(b, 0, result, 0, b.length); + result[b.length] = (byte) (value >> 24); + result[b.length + 1] = (byte) (value >> 16); + result[b.length + 2] = (byte) (value >> 8); + result[b.length + 3] = (byte) value; + return result; + } + + private byte[] writeInt64(byte[] b, long value) { + byte[] result = new byte[b.length + 8]; + System.arraycopy(b, 0, result, 0, b.length); + result[b.length] = (byte) (value >> 56); + result[b.length + 1] = (byte) (value >> 48); + result[b.length + 2] = (byte) (value >> 40); + result[b.length + 3] = (byte) (value >> 32); + result[b.length + 4] = (byte) (value >> 24); + result[b.length + 5] = (byte) (value >> 16); + result[b.length + 6] = (byte) (value >> 8); + result[b.length + 7] = (byte) value; + return result; + } + + private byte[] writeBytes(byte[] b, byte[] data) { + if (data == null || data.length == 0) { + return writeUint32(b, 0); + } + b = writeUint32(b, data.length); + byte[] result = new byte[b.length + data.length]; + System.arraycopy(b, 0, result, 0, b.length); + System.arraycopy(data, 0, result, b.length, data.length); + return result; + } + + public static Builder newBuilder() { + return new Builder(); + } + + public static class Builder { + private final Record record = new Record(); + + public Builder setMeasurement(String measurement) { + record.measurement = measurement; + return this; + } + + public Builder setMinTime(long minTime) { + record.minTime = minTime; + return this; + } + + public Builder setMaxTime(long maxTime) { + record.maxTime = maxTime; + return this; + } + + public Builder setCompressMethod(CompressMethod compressMethod) { + record.compressMethod = compressMethod; + return this; + } + + public Builder setBlock(byte[] block) { + record.block = block; + return this; + } + + public Record build() { + return record; + } + } +} \ No newline at end of file diff --git a/opengemini-client-api/src/main/java/io/opengemini/proto/WriteRequest.java b/opengemini-client-api/src/main/java/io/opengemini/proto/WriteRequest.java new file mode 100644 index 00000000..ddf583e9 --- /dev/null +++ b/opengemini-client-api/src/main/java/io/opengemini/proto/WriteRequest.java @@ -0,0 +1,183 @@ +/* + * 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.proto; + +import io.opengemini.client.api.grpc.CompressMethod; +import io.opengemini.client.api.grpc.ResponseCode; +import io.opengemini.client.api.grpc.ServerStatus; + +import java.util.ArrayList; +import java.util.List; + +/** + * Manual proto implementation for WriteRequest. + */ +public class WriteRequest { + + private int version; + private String database; + private String retentionPolicy; + private String username; + private String password; + private List records = new ArrayList<>(); + + public WriteRequest() { + } + + public int getVersion() { + return version; + } + + public WriteRequest setVersion(int version) { + this.version = version; + return this; + } + + public String getDatabase() { + return database; + } + + public WriteRequest setDatabase(String database) { + this.database = database; + return this; + } + + public String getRetentionPolicy() { + return retentionPolicy; + } + + public WriteRequest setRetentionPolicy(String retentionPolicy) { + this.retentionPolicy = retentionPolicy; + return this; + } + + public String getUsername() { + return username; + } + + public WriteRequest setUsername(String username) { + this.username = username; + return this; + } + + public String getPassword() { + return password; + } + + public WriteRequest setPassword(String password) { + this.password = password; + return this; + } + + public List getRecordsList() { + return records; + } + + public WriteRequest addRecords(Record record) { + this.records.add(record); + return this; + } + + public byte[] toByteArray() { + return toByteArrayInternal(new byte[0]); + } + + public byte[] toByteArrayInternal(byte[] buf) { + buf = writeUint32(buf, version); + buf = writeString(buf, database); + buf = writeString(buf, retentionPolicy); + buf = writeString(buf, username); + buf = writeString(buf, password); + buf = writeUint32(buf, records.size()); + for (Record record : records) { + buf = record.toByteArray(buf); + } + return buf; + } + + private byte[] writeUint32(byte[] b, int value) { + byte[] result = new byte[b.length + 4]; + System.arraycopy(b, 0, result, 0, b.length); + result[b.length] = (byte) (value >> 24); + result[b.length + 1] = (byte) (value >> 16); + result[b.length + 2] = (byte) (value >> 8); + result[b.length + 3] = (byte) value; + return result; + } + + private byte[] writeString(byte[] b, String s) { + if (s == null) { + return writeUint16(b, 0); + } + byte[] strBytes = s.getBytes(); + b = writeUint16(b, strBytes.length); + byte[] result = new byte[b.length + strBytes.length]; + System.arraycopy(b, 0, result, 0, b.length); + System.arraycopy(strBytes, 0, result, b.length, strBytes.length); + return result; + } + + private byte[] writeUint16(byte[] b, int value) { + byte[] result = new byte[b.length + 2]; + System.arraycopy(b, 0, result, 0, b.length); + result[b.length] = (byte) (value >> 8); + result[b.length + 1] = (byte) value; + return result; + } + + public static Builder newBuilder() { + return new Builder(); + } + + public static class Builder { + private final WriteRequest request = new WriteRequest(); + + public Builder setVersion(int version) { + request.version = version; + return this; + } + + public Builder setDatabase(String database) { + request.database = database; + return this; + } + + public Builder setRetentionPolicy(String retentionPolicy) { + request.retentionPolicy = retentionPolicy; + return this; + } + + public Builder setUsername(String username) { + request.username = username; + return this; + } + + public Builder setPassword(String password) { + request.password = password; + return this; + } + + public Builder addRecords(Record record) { + request.records.add(record); + return this; + } + + public WriteRequest build() { + return request; + } + } +} \ No newline at end of file diff --git a/opengemini-client-api/src/main/java/io/opengemini/proto/WriteResponse.java b/opengemini-client-api/src/main/java/io/opengemini/proto/WriteResponse.java new file mode 100644 index 00000000..475eaf1e --- /dev/null +++ b/opengemini-client-api/src/main/java/io/opengemini/proto/WriteResponse.java @@ -0,0 +1,68 @@ +/* + * 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.proto; + +import io.opengemini.client.api.grpc.ResponseCode; + +/** + * Manual proto implementation for WriteResponse. + */ +public class WriteResponse { + + private ResponseCode code = ResponseCode.Success; + + public WriteResponse() { + } + + public ResponseCode getCode() { + return code; + } + + public WriteResponse setCode(ResponseCode code) { + this.code = code; + return this; + } + + public byte[] toByteArray() { + return new byte[]{(byte) code.getValue()}; + } + + public static WriteResponse parseFrom(byte[] data) { + WriteResponse response = new WriteResponse(); + if (data != null && data.length > 0) { + response.setCode(ResponseCode.forNumber(data[0])); + } + return response; + } + + public static Builder newBuilder() { + return new Builder(); + } + + public static class Builder { + private final WriteResponse response = new WriteResponse(); + + public Builder setCode(ResponseCode code) { + response.code = code; + return this; + } + + public WriteResponse build() { + return response; + } + } +} \ No newline at end of file diff --git a/opengemini-client-api/src/main/proto/write.proto b/opengemini-client-api/src/main/proto/write.proto new file mode 100755 index 00000000..ec89a78c --- /dev/null +++ b/opengemini-client-api/src/main/proto/write.proto @@ -0,0 +1,59 @@ +syntax = "proto3"; +package proto; +option go_package = "./proto"; + +// 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; +} diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClient.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClient.java new file mode 100644 index 00000000..e3e370d9 --- /dev/null +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClient.java @@ -0,0 +1,35 @@ +/* + * 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.grpc.GrpcConfig; +import io.opengemini.proto.PingRequest; +import io.opengemini.proto.PingResponse; +import io.opengemini.proto.WriteRequest; +import io.opengemini.proto.WriteResponse; + +/** + * OpenGeminiGrpcClient interface for gRPC write operations. + */ +public interface OpenGeminiGrpcClient { + + WriteResponse write(WriteRequest request) throws Exception; + + PingResponse ping(PingRequest request); + + void close(); +} \ No newline at end of file diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientFactory.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientFactory.java new file mode 100644 index 00000000..dfb7ed84 --- /dev/null +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientFactory.java @@ -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. + */ + +package io.opengemini.client.impl.grpc; + +import io.opengemini.client.api.grpc.GrpcConfig; + +/** + * Factory class for creating OpenGeminiGrpcClient instances. + */ +public class OpenGeminiGrpcClientFactory { + + public static OpenGeminiGrpcClient create(GrpcConfig config) { + return new OpenGeminiGrpcClientImpl(config); + } +} \ No newline at end of file diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientImpl.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientImpl.java new file mode 100644 index 00000000..525bf053 --- /dev/null +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientImpl.java @@ -0,0 +1,171 @@ +/* + * 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.Address; +import io.opengemini.client.api.OpenGeminiException; +import io.opengemini.client.api.grpc.CompressMethod; +import io.opengemini.client.api.grpc.GrpcConfig; +import io.opengemini.client.api.grpc.ResponseCode; +import io.opengemini.client.api.grpc.ServerStatus; +import io.opengemini.proto.PingRequest; +import io.opengemini.proto.PingResponse; +import io.opengemini.proto.WriteRequest; +import io.opengemini.proto.WriteResponse; + +import java.io.InputStream; +import java.net.HttpURLConnection; +import java.net.URL; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Random; +import java.util.concurrent.TimeUnit; + +/** + * Implementation of OpenGeminiGrpcClient. + */ +public class OpenGeminiGrpcClientImpl implements OpenGeminiGrpcClient { + + private final GrpcConfig config; + private final List
addresses; + private final Random random = new Random(); + + public OpenGeminiGrpcClientImpl(GrpcConfig config) { + this.config = config; + this.addresses = config.getAddresses(); + } + + @Override + public WriteResponse write(WriteRequest request) throws Exception { + Address address = getAddress(); + String urlStr = buildUrl(address); + + try { + URL url = new URL(urlStr); + HttpURLConnection connection = (HttpURLConnection) url.openConnection(); + connection.setRequestMethod("POST"); + connection.setDoOutput(true); + connection.setConnectTimeout((int) config.getTimeout().toMillis()); + connection.setReadTimeout((int) config.getTimeout().toMillis()); + connection.setRequestProperty("Content-Type", "application/protobuf"); + + byte[] data = request.toByteArray(); + connection.setRequestProperty("Content-Length", String.valueOf(data.length)); + + connection.getOutputStream().write(data); + connection.getOutputStream().flush(); + connection.getOutputStream().close(); + + int responseCode = connection.getResponseCode(); + if (responseCode == 200 || responseCode == 201) { + InputStream is = connection.getInputStream(); + byte[] responseData = readAllBytes(is); + is.close(); + + if (responseData.length > 0) { + WriteResponse response = WriteResponse.parseFrom(responseData); + if (response.getCode() != ResponseCode.Success) { + throw new OpenGeminiException("Write failed: " + response.getCode()); + } + return response; + } + return WriteResponse.newBuilder().setCode(ResponseCode.Success).build(); + } else { + throw new OpenGeminiException("HTTP error: " + responseCode); + } + } catch (OpenGeminiException e) { + throw new RuntimeException(e); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + @Override + public PingResponse ping(PingRequest request) { + Address address = getAddress(); + String urlStr = buildPingUrl(address); + + try { + URL url = new URL(urlStr); + HttpURLConnection connection = (HttpURLConnection) url.openConnection(); + connection.setRequestMethod("POST"); + connection.setConnectTimeout((int) config.getTimeout().toMillis()); + connection.setReadTimeout((int) config.getTimeout().toMillis()); + connection.setRequestProperty("Content-Type", "application/protobuf"); + + int responseCode = connection.getResponseCode(); + if (responseCode == 200 || responseCode == 201) { + return PingResponse.newBuilder().setStatus(ServerStatus.Up).build(); + } else { + return PingResponse.newBuilder().setStatus(ServerStatus.Down).build(); + } + } catch (Exception e) { + return PingResponse.newBuilder().setStatus(ServerStatus.Down).build(); + } + } + + @Override + public void close() { + } + + private Address getAddress() { + if (addresses == null || addresses.isEmpty()) { + throw new IllegalStateException("No addresses configured"); + } + int index = random.nextInt(addresses.size()); + return addresses.get(index); + } + + private String buildUrl(Address address) { + return "http://" + address.getHost() + ":" + address.getPort() + "/api/v1/write"; + } + + private String buildPingUrl(Address address) { + return "http://" + address.getHost() + ":" + address.getPort() + "/api/v1/ping"; + } + + private byte[] readAllBytes(InputStream is) throws Exception { + java.io.ByteArrayOutputStream bos = new java.io.ByteArrayOutputStream(); + byte[] buffer = new byte[4096]; + int len; + while ((len = is.read(buffer)) != -1) { + bos.write(buffer, 0, len); + } + return bos.toByteArray(); + } + + private byte[] toByteArray(WriteRequest request) { + io.opengemini.proto.WriteRequest.Builder builder = io.opengemini.proto.WriteRequest.newBuilder(); + builder.setVersion(request.getVersion()); + if (request.getDatabase() != null) { + builder.setDatabase(request.getDatabase()); + } + if (request.getRetentionPolicy() != null) { + builder.setRetentionPolicy(request.getRetentionPolicy()); + } + if (request.getUsername() != null) { + builder.setUsername(request.getUsername()); + } + if (request.getPassword() != null) { + builder.setPassword(request.getPassword()); + } + for (io.opengemini.proto.Record record : request.getRecordsList()) { + builder.addRecords(record); + } + return builder.build().toByteArray(); + } +} \ No newline at end of file diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/BytesBuilder.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/BytesBuilder.java new file mode 100644 index 00000000..0329197e --- /dev/null +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/BytesBuilder.java @@ -0,0 +1,242 @@ +/* + * 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.record; + +import java.util.Arrays; + +/** + * BytesBuilder is a utility class for building byte arrays with various encoding methods. + * This is a Java port of the Go utils.go functions. + */ +public class BytesBuilder { + + private static final int SIZE_OF_INT = 8; + private static final int SIZE_OF_UINT16 = 2; + private static final int SIZE_OF_UINT32 = 4; + + private byte[] buffer; + private int position; + + public BytesBuilder() { + this.buffer = new byte[64]; + this.position = 0; + } + + public BytesBuilder(byte[] initial) { + this.buffer = initial; + this.position = initial.length; + } + + public byte[] toBytes() { + return Arrays.copyOf(buffer, position); + } + + public int length() { + return position; + } + + public BytesBuilder append(byte b) { + ensureCapacity(1); + buffer[position++] = b; + return this; + } + + public BytesBuilder append(byte[] b) { + if (b == null) return this; + ensureCapacity(b.length); + System.arraycopy(b, 0, buffer, position, b.length); + position += b.length; + return this; + } + + private void ensureCapacity(int needed) { + if (position + needed > buffer.length) { + int newSize = Math.max(buffer.length * 2, position + needed); + buffer = Arrays.copyOf(buffer, newSize); + } + } + + public static byte[] appendUint16(byte[] b, int value) { + byte[] result = new byte[b.length + SIZE_OF_UINT16]; + System.arraycopy(b, 0, result, 0, b.length); + result[b.length] = (byte) (value >> 8); + result[b.length + 1] = (byte) value; + return result; + } + + public BytesBuilder appendUint16(int value) { + ensureCapacity(SIZE_OF_UINT16); + buffer[position++] = (byte) (value >> 8); + buffer[position++] = (byte) value; + return this; + } + + public static byte[] appendUint32(byte[] b, long value) { + byte[] result = new byte[b.length + SIZE_OF_UINT32]; + System.arraycopy(b, 0, result, 0, b.length); + result[b.length] = (byte) (value >> 24); + result[b.length + 1] = (byte) (value >> 16); + result[b.length + 2] = (byte) (value >> 8); + result[b.length + 3] = (byte) value; + return result; + } + + public BytesBuilder appendUint32(long value) { + ensureCapacity(SIZE_OF_UINT32); + buffer[position++] = (byte) (value >> 24); + buffer[position++] = (byte) (value >> 16); + buffer[position++] = (byte) (value >> 8); + buffer[position++] = (byte) value; + return this; + } + + public static byte[] appendInt64(byte[] b, long value) { + long u = (value << 1) ^ (value >> 63); + return appendUint64(b, u); + } + + public static byte[] appendDouble(byte[] b, double value) { + return appendInt64(b, Double.doubleToLongBits(value)); + } + + public static byte[] appendBoolean(byte[] b, boolean value) { + byte[] result = new byte[b.length + 1]; + System.arraycopy(b, 0, result, 0, b.length); + result[b.length] = value ? (byte) 1 : (byte) 0; + return result; + } + + public BytesBuilder appendInt64(long value) { + long u = (value << 1) ^ (value >> 63); + return appendUint64(u); + } + + private static byte[] appendUint64(byte[] b, long value) { + byte[] result = new byte[b.length + 8]; + System.arraycopy(b, 0, result, 0, b.length); + result[b.length] = (byte) (value >> 56); + result[b.length + 1] = (byte) (value >> 48); + result[b.length + 2] = (byte) (value >> 40); + result[b.length + 3] = (byte) (value >> 32); + result[b.length + 4] = (byte) (value >> 24); + result[b.length + 5] = (byte) (value >> 16); + result[b.length + 6] = (byte) (value >> 8); + result[b.length + 7] = (byte) value; + return result; + } + + private BytesBuilder appendUint64(long value) { + ensureCapacity(8); + buffer[position++] = (byte) (value >> 56); + buffer[position++] = (byte) (value >> 48); + buffer[position++] = (byte) (value >> 40); + buffer[position++] = (byte) (value >> 32); + buffer[position++] = (byte) (value >> 24); + buffer[position++] = (byte) (value >> 16); + buffer[position++] = (byte) (value >> 8); + buffer[position++] = (byte) value; + return this; + } + + public static byte[] appendInt(byte[] b, int value) { + return appendInt64(b, value); + } + + public BytesBuilder appendInt(int value) { + return appendInt64(value); + } + + public static byte[] appendString(byte[] b, String s) { + if (s == null) { + return appendUint16(b, 0); + } + byte[] strBytes = s.getBytes(); + b = appendUint16(b, strBytes.length); + byte[] result = new byte[b.length + strBytes.length]; + System.arraycopy(b, 0, result, 0, b.length); + System.arraycopy(strBytes, 0, result, b.length, strBytes.length); + return result; + } + + public BytesBuilder appendString(String s) { + if (s == null) { + appendUint16(0); + return this; + } + byte[] strBytes = s.getBytes(); + appendUint16(strBytes.length); + ensureCapacity(strBytes.length); + System.arraycopy(strBytes, 0, buffer, position, strBytes.length); + position += strBytes.length; + return this; + } + + public static byte[] appendBytes(byte[] b, byte[] buf) { + if (buf == null || buf.length == 0) { + return appendUint32(b, 0); + } + b = appendUint32(b, buf.length); + byte[] result = new byte[b.length + buf.length]; + System.arraycopy(b, 0, result, 0, b.length); + System.arraycopy(buf, 0, result, b.length, buf.length); + return result; + } + + public BytesBuilder appendBytes(byte[] buf) { + if (buf == null || buf.length == 0) { + appendUint32(0); + return this; + } + appendUint32(buf.length); + ensureCapacity(buf.length); + System.arraycopy(buf, 0, buffer, position, buf.length); + position += buf.length; + return this; + } + + public BytesBuilder appendDouble(double value) { + return appendInt64(Double.doubleToLongBits(value)); + } + + public BytesBuilder appendBoolean(boolean value) { + ensureCapacity(1); + buffer[position++] = value ? (byte) 1 : (byte) 0; + return this; + } + + public static int sizeOfString(String s) { + if (s == null) { + return SIZE_OF_UINT16; + } + return SIZE_OF_UINT16 + s.getBytes().length; + } + + public static int sizeOfInt() { + return SIZE_OF_INT; + } + + public static int sizeOfUint32() { + return SIZE_OF_UINT32; + } + + public static int sizeOfByteSlice(byte[] s) { + if (s == null) { + return SIZE_OF_UINT32; + } + return SIZE_OF_UINT32 + s.length; + } +} \ No newline at end of file diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/ColVal.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/ColVal.java new file mode 100644 index 00000000..bd5c8d1e --- /dev/null +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/ColVal.java @@ -0,0 +1,178 @@ +/* + * 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.record; + +import java.util.ArrayList; +import java.util.List; + +/** + * ColVal represents column values in a record. + */ +public class ColVal { + + private List stringVals = new ArrayList<>(); + private List integerVals = new ArrayList<>(); + private List floatVals = new ArrayList<>(); + private List booleanVals = new ArrayList<>(); + private int nilCount = 0; + private int len = 0; + + public ColVal() { + } + + public void init() { + stringVals.clear(); + integerVals.clear(); + floatVals.clear(); + booleanVals.clear(); + nilCount = 0; + len = 0; + } + + public void appendString(String val) { + stringVals.add(val); + len++; + } + + public void appendStringNull() { + stringVals.add(null); + nilCount++; + len++; + } + + public void appendStringNulls(int count) { + for (int i = 0; i < count; i++) { + stringVals.add(null); + } + nilCount += count; + len += count; + } + + public void appendInteger(long val) { + integerVals.add(val); + len++; + } + + public void appendIntegerNull() { + integerVals.add(null); + nilCount++; + len++; + } + + public void appendIntegerNulls(int count) { + for (int i = 0; i < count; i++) { + integerVals.add(null); + } + nilCount += count; + len += count; + } + + public void appendFloat(double val) { + floatVals.add(val); + len++; + } + + public void appendFloatNull() { + floatVals.add(null); + nilCount++; + len++; + } + + public void appendFloatNulls(int count) { + for (int i = 0; i < count; i++) { + floatVals.add(null); + } + nilCount += count; + len += count; + } + + public void appendBoolean(boolean val) { + booleanVals.add(val); + len++; + } + + public void appendBooleanNull() { + booleanVals.add(null); + nilCount++; + len++; + } + + public void appendBooleanNulls(int count) { + for (int i = 0; i < count; i++) { + booleanVals.add(null); + } + nilCount += count; + len += count; + } + + public List stringValues() { + return stringVals; + } + + public List integerValues() { + return integerVals; + } + + public List floatValues() { + return floatVals; + } + + public List booleanValues() { + return booleanVals; + } + + public int getNilCount() { + return nilCount; + } + + public int len() { + return len; + } + + public int size() { + int size = 0; + for (String val : stringVals) { + if (val != null) { + size += val.length() + 2; + } + } + size += integerVals.size() * 8; + size += floatVals.size() * 8; + size += booleanVals.size() * 1; + return size; + } + + public byte[] marshal(byte[] buf) { + int sz = size(); + buf = BytesBuilder.appendUint32(buf, sz); + + for (String val : stringVals) { + buf = BytesBuilder.appendString(buf, val); + } + for (Long val : integerVals) { + buf = BytesBuilder.appendInt64(buf, val); + } + for (Double val : floatVals) { + buf = BytesBuilder.appendDouble(buf, val); + } + for (Boolean val : booleanVals) { + buf = BytesBuilder.appendBoolean(buf, val); + } + + return buf; + } +} \ No newline at end of file diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Field.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Field.java new file mode 100644 index 00000000..064c931b --- /dev/null +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Field.java @@ -0,0 +1,57 @@ +/* + * 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.record; + +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; + +/** + * Field represents a field schema in a record. + */ +@Getter +@Setter +@NoArgsConstructor +public class Field { + + private int type; + private String name; + + public Field(int type, String name) { + this.type = type; + this.name = name; + } + + public int getSize() { + int size = 0; + size += BytesBuilder.sizeOfString(name); + size += BytesBuilder.sizeOfInt(); + return size; + } + + @Override + public String toString() { + return name + FieldType.getTypeName(type); + } + + public byte[] marshal(byte[] buf) { + buf = BytesBuilder.appendString(buf, name); + buf = BytesBuilder.appendInt(buf, type); + return buf; + } +} \ No newline at end of file diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/FieldType.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/FieldType.java new file mode 100644 index 00000000..54969ae3 --- /dev/null +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/FieldType.java @@ -0,0 +1,75 @@ +/* + * 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.record; + +/** + * FieldType represents the type of a field in a record. + */ +public class FieldType { + + public static final int UNKNOWN = 0; + public static final int INT = 1; + public static final int UINT = 2; + public static final int FLOAT = 3; + public static final int STRING = 4; + public static final int BOOLEAN = 5; + public static final int TAG = 6; + public static final int LAST = 7; + + public static final String[] FIELD_TYPE_NAME = { + "Unknown", + "Integer", + "Unsigned", + "Float", + "String", + "Boolean", + "Tag", + "Unknown" + }; + + public static String getTypeName(int type) { + if (type >= 0 && type < FIELD_TYPE_NAME.length) { + return FIELD_TYPE_NAME[type]; + } + return FIELD_TYPE_NAME[UNKNOWN]; + } + + public static int getTypeSize(int type) { + switch (type) { + case INT: + case UINT: + return 8; + case FLOAT: + return 8; + case BOOLEAN: + return 1; + case STRING: + case TAG: + return -1; + default: + return 0; + } + } + + public static boolean isStringType(int type) { + return type == STRING || type == TAG; + } + + public static boolean isIntegerType(int type) { + return type == INT || type == UINT; + } +} \ No newline at end of file diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Record.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Record.java new file mode 100644 index 00000000..45b87148 --- /dev/null +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Record.java @@ -0,0 +1,227 @@ +/* + * 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.record; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; +import java.util.List; + +/** + * Record represents a collection of fields and their values. + */ +public class Record { + + public static final String TIME_FIELD = "time"; + + private List schema = new ArrayList<>(); + private List colVals = new ArrayList<>(); + + public Record() { + } + + public void reset() { + schema.clear(); + colVals.clear(); + } + + public void reserveColVal(int size) { + int colLen = colVals.size(); + int colCap = colVals.size(); + int remain = colCap - colLen; + if (size - remain > 0) { + for (int i = 0; i < size - remain; i++) { + colVals.add(new ColVal()); + } + } + for (int i = colLen; i < colLen + size; i++) { + if (i < colVals.size()) { + colVals.get(i).init(); + } + } + } + + public void initColVal(int start, int end) { + for (int i = start; i < end; i++) { + if (i < colVals.size()) { + colVals.get(i).init(); + } + } + } + + public void addColumn(Field field, ColVal colVal) { + schema.add(field); + colVals.add(colVal); + } + + public Field getColumn(int index) { + if (index >= 0 && index < schema.size()) { + return schema.get(index); + } + return null; + } + + public ColVal getColVal(int index) { + if (index >= 0 && index < colVals.size()) { + return colVals.get(index); + } + return null; + } + + public int getRowNums() { + if (colVals.isEmpty()) { + return 0; + } + return colVals.get(colVals.size() - 1).len(); + } + + public long[] times() { + if (colVals.isEmpty()) { + return null; + } + ColVal timeCol = colVals.get(colVals.size() - 1); + List times = timeCol.integerValues(); + long[] result = new long[times.size()]; + for (int i = 0; i < times.size(); i++) { + result[i] = times.get(i); + } + return result; + } + + public void appendTime(long time) { + if (!colVals.isEmpty()) { + colVals.get(colVals.size() - 1).appendInteger(time); + } + } + + public int len() { + return schema.size(); + } + + public List getSchema() { + return schema; + } + + public List getColVals() { + return colVals; + } + + public void sort() { + for (int i = 0; i < schema.size() - 1; i++) { + for (int j = i + 1; j < schema.size(); j++) { + Field fi = schema.get(i); + Field fj = schema.get(j); + + if (TIME_FIELD.equals(fi.getName())) { + continue; + } + if (TIME_FIELD.equals(fj.getName())) { + swap(i, j); + continue; + } + if (fi.getName().compareTo(fj.getName()) > 0) { + swap(i, j); + } + } + } + } + + private void swap(int i, int j) { + Field tmpField = schema.get(i); + schema.set(i, schema.get(j)); + schema.set(j, tmpField); + + ColVal tmpColVal = colVals.get(i); + colVals.set(i, colVals.get(j)); + colVals.set(j, tmpColVal); + } + + public byte[] marshal(byte[] buf) { + buf = BytesBuilder.appendUint32(buf, schema.size()); + for (Field field : schema) { + buf = field.marshal(buf); + } + + buf = BytesBuilder.appendUint32(buf, colVals.size()); + for (ColVal colVal : colVals) { + buf = colVal.marshal(buf); + } + + return buf; + } + + public static Record checkRecord(Record rec) throws Exception { + int colN = rec.schema.size(); + if (colN <= 1 || !TIME_FIELD.equals(rec.schema.get(colN - 1).getName())) { + throw new Exception("invalid schema: " + rec.schema); + } + + ColVal lastCol = rec.colVals.get(colN - 1); + if (lastCol.getNilCount() != 0) { + throw new Exception("invalid colvals"); + } + + for (int i = 1; i < colN; i++) { + if (rec.schema.get(i).getName().equals(rec.schema.get(i - 1).getName())) { + throw new Exception("same schema: " + rec.schema.get(i).getName()); + } + } + + boolean isOrderSchema = true; + for (int i = 0; i < colN - 1; i++) { + Field f = rec.schema.get(i); + ColVal col1 = rec.colVals.get(i); + ColVal col2 = rec.colVals.get(i + 1); + + if (col1.len() != col2.len()) { + throw new Exception("invalid colvals length"); + } + + if (isOrderSchema && i > 0 && rec.schema.get(i - 1).getName().compareTo(rec.schema.get(i).getName()) >= 0) { + isOrderSchema = false; + } + + if (!FieldType.isStringType(f.getType())) { + int typeSize = FieldType.getTypeSize(f.getType()); + int expLen = typeSize * (col1.len() - col1.getNilCount()); + if (expLen != getValLength(col1, f.getType())) { + throw new Exception("invalid colvals val length"); + } + } + } + + if (!isOrderSchema) { + rec.sort(); + } + + return rec; + } + + private static int getValLength(ColVal colVal, int type) { + switch (type) { + case FieldType.INT: + case FieldType.UINT: + return colVal.integerValues().size() * 8; + case FieldType.FLOAT: + return colVal.floatValues().size() * 8; + case FieldType.BOOLEAN: + return colVal.booleanValues().size() * 1; + default: + return 0; + } + } +} \ No newline at end of file diff --git a/pom.xml b/pom.xml index 91216244..f6cd3790 100644 --- a/pom.xml +++ b/pom.xml @@ -116,6 +116,11 @@ pom import + + com.google.protobuf + protobuf-java + ${protobuf.version} + @@ -196,6 +201,11 @@ ${src.dir} + + org.xolstice.maven.plugins + protobuf-maven-plugin + ${protobuf-maven-plugin.version} + org.apache.maven.plugins maven-enforcer-plugin diff --git a/spring/opengemini-spring-boot-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/spring/opengemini-spring-boot-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports index 911226b2..da11d06d 100644 --- a/spring/opengemini-spring-boot-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ b/spring/opengemini-spring-boot-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -1 +1,2 @@ io.opengemini.client.spring.data.config.OpenGeminiAutoConfiguration +io.opengemini.client.spring.data.config.OpenGeminiGrpcAutoConfiguration From dd7ae5c90b7da5c5042b67ef22803707936250da Mon Sep 17 00:00:00 2001 From: weiping-code <54316849+weiping-code@users.noreply.github.com> Date: Wed, 22 Apr 2026 18:43:53 +0800 Subject: [PATCH 2/4] fix: correct protobuf-maven-plugin version property Signed-off-by: weiping-code <54316849+weiping-code@users.noreply.github.com> --- pom.xml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index f6cd3790..5f82ba4f 100644 --- a/pom.xml +++ b/pom.xml @@ -78,7 +78,7 @@ 3.3.1 3.5.0 1.7.1 - 0.6.1 + 0.6.1 1.6.13 4.9.2 4.9.1.0 @@ -204,7 +204,7 @@ org.xolstice.maven.plugins protobuf-maven-plugin - ${protobuf-maven-plugin.version} + ${maven-protobuf-maven-plugin.version} org.apache.maven.plugins From f038220862f99ad45c4c667c7531c4b3a4b9d325 Mon Sep 17 00:00:00 2001 From: weiping-code <54316849+weiping-code@users.noreply.github.com> Date: Wed, 22 Apr 2026 19:02:03 +0800 Subject: [PATCH 3/4] fix: remove missing gRPC auto-configuration from imports Signed-off-by: weiping-code <54316849+weiping-code@users.noreply.github.com> --- ...pringframework.boot.autoconfigure.AutoConfiguration.imports | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/spring/opengemini-spring-boot-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/spring/opengemini-spring-boot-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports index da11d06d..3d432f55 100644 --- a/spring/opengemini-spring-boot-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ b/spring/opengemini-spring-boot-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -1,2 +1 @@ -io.opengemini.client.spring.data.config.OpenGeminiAutoConfiguration -io.opengemini.client.spring.data.config.OpenGeminiGrpcAutoConfiguration +io.opengemini.client.spring.data.config.OpenGeminiAutoConfiguration \ No newline at end of file From 45b7a12159e6069b566e3106790bcad5cc049e22 Mon Sep 17 00:00:00 2001 From: weiping-code <54316849+weiping-code@users.noreply.github.com> Date: Wed, 22 Apr 2026 19:36:08 +0800 Subject: [PATCH 4/4] fix: add package-info, fix imports, add newline, fix spotbugs Signed-off-by: weiping-code <54316849+weiping-code@users.noreply.github.com> --- .gitignore | 2 +- ci/spotbugs/exclude.xml | 45 +++++++++++++++++++ opengemini-client-api/pom.xml | 5 +++ .../client/api/grpc/CompressMethod.java | 2 +- .../client/api/grpc/GrpcConfig.java | 4 +- .../client/api/grpc/ResponseCode.java | 2 +- .../client/api/grpc/ServerStatus.java | 2 +- .../client/api/grpc/package-info.java | 17 +++++++ .../java/io/opengemini/proto/PingRequest.java | 8 ++-- .../io/opengemini/proto/PingResponse.java | 2 +- .../main/java/io/opengemini/proto/Record.java | 6 ++- .../io/opengemini/proto/WriteRequest.java | 8 ++-- .../io/opengemini/proto/WriteResponse.java | 2 +- .../impl/grpc/OpenGeminiGrpcClient.java | 3 +- .../grpc/OpenGeminiGrpcClientFactory.java | 2 +- .../impl/grpc/OpenGeminiGrpcClientImpl.java | 5 +-- .../client/impl/grpc/package-info.java | 17 +++++++ .../client/impl/grpc/record/BytesBuilder.java | 13 +++--- .../client/impl/grpc/record/ColVal.java | 2 +- .../client/impl/grpc/record/Field.java | 3 +- .../client/impl/grpc/record/FieldType.java | 2 +- .../client/impl/grpc/record/Record.java | 4 +- .../client/impl/grpc/record/package-info.java | 17 +++++++ pom.xml | 16 +------ ...ot.autoconfigure.AutoConfiguration.imports | 2 +- 25 files changed, 138 insertions(+), 53 deletions(-) create mode 100644 opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/package-info.java create mode 100644 opengemini-client/src/main/java/io/opengemini/client/impl/grpc/package-info.java create mode 100644 opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/package-info.java diff --git a/.gitignore b/.gitignore index 8b38d3e7..efc5a2ea 100644 --- a/.gitignore +++ b/.gitignore @@ -24,4 +24,4 @@ dependency-reduced-pom.xml logs/ *.xml.releaseBackup -.go \ No newline at end of file +.go diff --git a/ci/spotbugs/exclude.xml b/ci/spotbugs/exclude.xml index 563c8699..4fc237ce 100644 --- a/ci/spotbugs/exclude.xml +++ b/ci/spotbugs/exclude.xml @@ -22,6 +22,34 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + @@ -46,4 +74,21 @@ + + + + + + + + + + + + + + + + + diff --git a/opengemini-client-api/pom.xml b/opengemini-client-api/pom.xml index 1e4989ff..f8605f8d 100644 --- a/opengemini-client-api/pom.xml +++ b/opengemini-client-api/pom.xml @@ -53,5 +53,10 @@ com.google.protobuf protobuf-java + + javax.annotation + javax.annotation-api + 1.3.2 + diff --git a/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/CompressMethod.java b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/CompressMethod.java index 8d4a9e20..12b3e026 100644 --- a/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/CompressMethod.java +++ b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/CompressMethod.java @@ -40,4 +40,4 @@ public static CompressMethod forNumber(int value) { } return UNCOMPRESSED; } -} \ No newline at end of file +} diff --git a/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/GrpcConfig.java b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/GrpcConfig.java index 0478e579..433e55d5 100644 --- a/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/GrpcConfig.java +++ b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/GrpcConfig.java @@ -16,8 +16,8 @@ package io.opengemini.client.api.grpc; -import io.opengemini.client.api.AuthConfig; import io.opengemini.client.api.Address; +import io.opengemini.client.api.AuthConfig; import io.opengemini.client.api.TlsConfig; import lombok.Getter; import lombok.Setter; @@ -58,4 +58,4 @@ public class GrpcConfig { * Timeout default 30s */ private Duration timeout = Duration.ofSeconds(30); -} \ No newline at end of file +} diff --git a/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ResponseCode.java b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ResponseCode.java index c9743d61..10c882ba 100644 --- a/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ResponseCode.java +++ b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ResponseCode.java @@ -42,4 +42,4 @@ public static ResponseCode forNumber(int value) { } return Success; } -} \ No newline at end of file +} diff --git a/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ServerStatus.java b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ServerStatus.java index fc0bf74e..d573de3c 100644 --- a/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ServerStatus.java +++ b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/ServerStatus.java @@ -42,4 +42,4 @@ public static ServerStatus forNumber(int value) { } return Unknown; } -} \ No newline at end of file +} diff --git a/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/package-info.java b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/package-info.java new file mode 100644 index 00000000..17332f2c --- /dev/null +++ b/opengemini-client-api/src/main/java/io/opengemini/client/api/grpc/package-info.java @@ -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. + */ + +package io.opengemini.client.api.grpc; diff --git a/opengemini-client-api/src/main/java/io/opengemini/proto/PingRequest.java b/opengemini-client-api/src/main/java/io/opengemini/proto/PingRequest.java index af862bcc..e5569dcc 100644 --- a/opengemini-client-api/src/main/java/io/opengemini/proto/PingRequest.java +++ b/opengemini-client-api/src/main/java/io/opengemini/proto/PingRequest.java @@ -16,6 +16,8 @@ package io.opengemini.proto; +import java.nio.charset.StandardCharsets; + /** * Manual proto implementation for PingRequest. */ @@ -39,13 +41,13 @@ public byte[] toByteArray() { if (clientId == null || clientId.isEmpty()) { return new byte[0]; } - return clientId.getBytes(); + return clientId.getBytes(StandardCharsets.UTF_8); } public static PingRequest parseFrom(byte[] data) { PingRequest request = new PingRequest(); if (data != null && data.length > 0) { - request.setClientId(new String(data)); + request.setClientId(new String(data, StandardCharsets.UTF_8)); } return request; } @@ -66,4 +68,4 @@ public PingRequest build() { return request; } } -} \ No newline at end of file +} diff --git a/opengemini-client-api/src/main/java/io/opengemini/proto/PingResponse.java b/opengemini-client-api/src/main/java/io/opengemini/proto/PingResponse.java index 63fd8071..c23a2e00 100644 --- a/opengemini-client-api/src/main/java/io/opengemini/proto/PingResponse.java +++ b/opengemini-client-api/src/main/java/io/opengemini/proto/PingResponse.java @@ -65,4 +65,4 @@ public PingResponse build() { return response; } } -} \ No newline at end of file +} diff --git a/opengemini-client-api/src/main/java/io/opengemini/proto/Record.java b/opengemini-client-api/src/main/java/io/opengemini/proto/Record.java index 172d7cc7..a6c4b4d2 100644 --- a/opengemini-client-api/src/main/java/io/opengemini/proto/Record.java +++ b/opengemini-client-api/src/main/java/io/opengemini/proto/Record.java @@ -18,6 +18,8 @@ import io.opengemini.client.api.grpc.CompressMethod; +import java.nio.charset.StandardCharsets; + /** * Manual proto implementation for Record. */ @@ -90,7 +92,7 @@ private byte[] writeString(byte[] b, String s) { if (s == null) { return writeUint16(b, 0); } - byte[] strBytes = s.getBytes(); + byte[] strBytes = s.getBytes(StandardCharsets.UTF_8); b = writeUint16(b, strBytes.length); byte[] result = new byte[b.length + strBytes.length]; System.arraycopy(b, 0, result, 0, b.length); @@ -177,4 +179,4 @@ public Record build() { return record; } } -} \ No newline at end of file +} diff --git a/opengemini-client-api/src/main/java/io/opengemini/proto/WriteRequest.java b/opengemini-client-api/src/main/java/io/opengemini/proto/WriteRequest.java index ddf583e9..1ca7917c 100644 --- a/opengemini-client-api/src/main/java/io/opengemini/proto/WriteRequest.java +++ b/opengemini-client-api/src/main/java/io/opengemini/proto/WriteRequest.java @@ -16,10 +16,8 @@ package io.opengemini.proto; -import io.opengemini.client.api.grpc.CompressMethod; -import io.opengemini.client.api.grpc.ResponseCode; -import io.opengemini.client.api.grpc.ServerStatus; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.List; @@ -123,7 +121,7 @@ private byte[] writeString(byte[] b, String s) { if (s == null) { return writeUint16(b, 0); } - byte[] strBytes = s.getBytes(); + byte[] strBytes = s.getBytes(StandardCharsets.UTF_8); b = writeUint16(b, strBytes.length); byte[] result = new byte[b.length + strBytes.length]; System.arraycopy(b, 0, result, 0, b.length); @@ -180,4 +178,4 @@ public WriteRequest build() { return request; } } -} \ No newline at end of file +} diff --git a/opengemini-client-api/src/main/java/io/opengemini/proto/WriteResponse.java b/opengemini-client-api/src/main/java/io/opengemini/proto/WriteResponse.java index 475eaf1e..440eb97b 100644 --- a/opengemini-client-api/src/main/java/io/opengemini/proto/WriteResponse.java +++ b/opengemini-client-api/src/main/java/io/opengemini/proto/WriteResponse.java @@ -65,4 +65,4 @@ public WriteResponse build() { return response; } } -} \ No newline at end of file +} diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClient.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClient.java index e3e370d9..8c5c7e5d 100644 --- a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClient.java +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClient.java @@ -16,7 +16,6 @@ package io.opengemini.client.impl.grpc; -import io.opengemini.client.api.grpc.GrpcConfig; import io.opengemini.proto.PingRequest; import io.opengemini.proto.PingResponse; import io.opengemini.proto.WriteRequest; @@ -32,4 +31,4 @@ public interface OpenGeminiGrpcClient { PingResponse ping(PingRequest request); void close(); -} \ No newline at end of file +} diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientFactory.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientFactory.java index dfb7ed84..68607931 100644 --- a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientFactory.java +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientFactory.java @@ -26,4 +26,4 @@ public class OpenGeminiGrpcClientFactory { public static OpenGeminiGrpcClient create(GrpcConfig config) { return new OpenGeminiGrpcClientImpl(config); } -} \ No newline at end of file +} diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientImpl.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientImpl.java index 525bf053..78298fc5 100644 --- a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientImpl.java +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientImpl.java @@ -18,7 +18,6 @@ import io.opengemini.client.api.Address; import io.opengemini.client.api.OpenGeminiException; -import io.opengemini.client.api.grpc.CompressMethod; import io.opengemini.client.api.grpc.GrpcConfig; import io.opengemini.client.api.grpc.ResponseCode; import io.opengemini.client.api.grpc.ServerStatus; @@ -30,10 +29,8 @@ import java.io.InputStream; import java.net.HttpURLConnection; import java.net.URL; -import java.nio.ByteBuffer; import java.util.List; import java.util.Random; -import java.util.concurrent.TimeUnit; /** * Implementation of OpenGeminiGrpcClient. @@ -168,4 +165,4 @@ private byte[] toByteArray(WriteRequest request) { } return builder.build().toByteArray(); } -} \ No newline at end of file +} diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/package-info.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/package-info.java new file mode 100644 index 00000000..360dcbd5 --- /dev/null +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/package-info.java @@ -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. + */ + +package io.opengemini.client.impl.grpc; diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/BytesBuilder.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/BytesBuilder.java index 0329197e..b5800c2c 100644 --- a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/BytesBuilder.java +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/BytesBuilder.java @@ -16,6 +16,7 @@ package io.opengemini.client.impl.grpc.record; +import java.nio.charset.StandardCharsets; import java.util.Arrays; /** @@ -56,7 +57,9 @@ public BytesBuilder append(byte b) { } public BytesBuilder append(byte[] b) { - if (b == null) return this; + if (b == null) { + return this; + } ensureCapacity(b.length); System.arraycopy(b, 0, buffer, position, b.length); position += b.length; @@ -164,7 +167,7 @@ public static byte[] appendString(byte[] b, String s) { if (s == null) { return appendUint16(b, 0); } - byte[] strBytes = s.getBytes(); + byte[] strBytes = s.getBytes(StandardCharsets.UTF_8); b = appendUint16(b, strBytes.length); byte[] result = new byte[b.length + strBytes.length]; System.arraycopy(b, 0, result, 0, b.length); @@ -177,7 +180,7 @@ public BytesBuilder appendString(String s) { appendUint16(0); return this; } - byte[] strBytes = s.getBytes(); + byte[] strBytes = s.getBytes(StandardCharsets.UTF_8); appendUint16(strBytes.length); ensureCapacity(strBytes.length); System.arraycopy(strBytes, 0, buffer, position, strBytes.length); @@ -222,7 +225,7 @@ public static int sizeOfString(String s) { if (s == null) { return SIZE_OF_UINT16; } - return SIZE_OF_UINT16 + s.getBytes().length; + return SIZE_OF_UINT16 + s.getBytes(StandardCharsets.UTF_8).length; } public static int sizeOfInt() { @@ -239,4 +242,4 @@ public static int sizeOfByteSlice(byte[] s) { } return SIZE_OF_UINT32 + s.length; } -} \ No newline at end of file +} diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/ColVal.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/ColVal.java index bd5c8d1e..06b1b6c5 100644 --- a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/ColVal.java +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/ColVal.java @@ -175,4 +175,4 @@ public byte[] marshal(byte[] buf) { return buf; } -} \ No newline at end of file +} diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Field.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Field.java index 064c931b..93e17ccf 100644 --- a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Field.java +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Field.java @@ -16,7 +16,6 @@ package io.opengemini.client.impl.grpc.record; -import lombok.AllArgsConstructor; import lombok.Getter; import lombok.NoArgsConstructor; import lombok.Setter; @@ -54,4 +53,4 @@ public byte[] marshal(byte[] buf) { buf = BytesBuilder.appendInt(buf, type); return buf; } -} \ No newline at end of file +} diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/FieldType.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/FieldType.java index 54969ae3..48789625 100644 --- a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/FieldType.java +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/FieldType.java @@ -72,4 +72,4 @@ public static boolean isStringType(int type) { public static boolean isIntegerType(int type) { return type == INT || type == UINT; } -} \ No newline at end of file +} diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Record.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Record.java index 45b87148..64bebe18 100644 --- a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Record.java +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Record.java @@ -17,8 +17,6 @@ package io.opengemini.client.impl.grpc.record; import java.util.ArrayList; -import java.util.Arrays; -import java.util.Comparator; import java.util.List; /** @@ -224,4 +222,4 @@ private static int getValLength(ColVal colVal, int type) { return 0; } } -} \ No newline at end of file +} diff --git a/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/package-info.java b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/package-info.java new file mode 100644 index 00000000..63c2966f --- /dev/null +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/package-info.java @@ -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. + */ + +package io.opengemini.client.impl.grpc.record; diff --git a/pom.xml b/pom.xml index 5f82ba4f..0c52d203 100644 --- a/pom.xml +++ b/pom.xml @@ -229,7 +229,7 @@ org.xolstice.maven.plugins protobuf-maven-plugin - ${maven-protobuf-maven-plugin} + ${maven-protobuf-maven-plugin.version} com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier} @@ -248,20 +248,6 @@ - - - vertx-grpc - - compile - compile-custom - - - grpc-vertx - - io.vertx:vertx-grpc-protoc-plugin:${vertx.version}:exe:${os.detected.classifier} - - - diff --git a/spring/opengemini-spring-boot-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/spring/opengemini-spring-boot-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports index 3d432f55..911226b2 100644 --- a/spring/opengemini-spring-boot-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ b/spring/opengemini-spring-boot-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -1 +1 @@ -io.opengemini.client.spring.data.config.OpenGeminiAutoConfiguration \ No newline at end of file +io.opengemini.client.spring.data.config.OpenGeminiAutoConfiguration