diff --git a/.gitignore b/.gitignore index 6e99d4ef..efc5a2ea 100644 --- a/.gitignore +++ b/.gitignore @@ -24,3 +24,4 @@ dependency-reduced-pom.xml logs/ *.xml.releaseBackup +.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 27a76f5e..f8605f8d 100644 --- a/opengemini-client-api/pom.xml +++ b/opengemini-client-api/pom.xml @@ -36,5 +36,27 @@ annotations ${annotations.version} + + + io.grpc + grpc-api + + + io.grpc + grpc-stub + + + io.grpc + grpc-protobuf + + + 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/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..12b3e026 --- /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; + } +} 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..433e55d5 --- /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.Address; +import io.opengemini.client.api.AuthConfig; +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); +} 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..10c882ba --- /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; + } +} 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..d573de3c --- /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; + } +} 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 new file mode 100644 index 00000000..e5569dcc --- /dev/null +++ b/opengemini-client-api/src/main/java/io/opengemini/proto/PingRequest.java @@ -0,0 +1,71 @@ +/* + * 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 java.nio.charset.StandardCharsets; + +/** + * 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(StandardCharsets.UTF_8); + } + + public static PingRequest parseFrom(byte[] data) { + PingRequest request = new PingRequest(); + if (data != null && data.length > 0) { + request.setClientId(new String(data, StandardCharsets.UTF_8)); + } + 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; + } + } +} 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..c23a2e00 --- /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; + } + } +} 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..a6c4b4d2 --- /dev/null +++ b/opengemini-client-api/src/main/java/io/opengemini/proto/Record.java @@ -0,0 +1,182 @@ +/* + * 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 java.nio.charset.StandardCharsets; + +/** + * 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(StandardCharsets.UTF_8); + 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; + } + } +} 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..1ca7917c --- /dev/null +++ b/opengemini-client-api/src/main/java/io/opengemini/proto/WriteRequest.java @@ -0,0 +1,181 @@ +/* + * 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 java.nio.charset.StandardCharsets; +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(StandardCharsets.UTF_8); + 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; + } + } +} 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..440eb97b --- /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; + } + } +} 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..8c5c7e5d --- /dev/null +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClient.java @@ -0,0 +1,34 @@ +/* + * 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.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(); +} 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..68607931 --- /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); + } +} 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..78298fc5 --- /dev/null +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/OpenGeminiGrpcClientImpl.java @@ -0,0 +1,168 @@ +/* + * 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.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.util.List; +import java.util.Random; + +/** + * 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(); + } +} 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 new file mode 100644 index 00000000..b5800c2c --- /dev/null +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/BytesBuilder.java @@ -0,0 +1,245 @@ +/* + * 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.nio.charset.StandardCharsets; +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(StandardCharsets.UTF_8); + 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(StandardCharsets.UTF_8); + 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(StandardCharsets.UTF_8).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; + } +} 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..06b1b6c5 --- /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; + } +} 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..93e17ccf --- /dev/null +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Field.java @@ -0,0 +1,56 @@ +/* + * 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.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; + } +} 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..48789625 --- /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; + } +} 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..64bebe18 --- /dev/null +++ b/opengemini-client/src/main/java/io/opengemini/client/impl/grpc/record/Record.java @@ -0,0 +1,225 @@ +/* + * 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; + +/** + * 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; + } + } +} 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 91216244..0c52d203 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 @@ -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 + ${maven-protobuf-maven-plugin.version} + org.apache.maven.plugins maven-enforcer-plugin @@ -219,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} @@ -238,20 +248,6 @@ - - - vertx-grpc - - compile - compile-custom - - - grpc-vertx - - io.vertx:vertx-grpc-protoc-plugin:${vertx.version}:exe:${os.detected.classifier} - - -