Skip to content

Commit 9e6efb3

Browse files
chenhuansomechenhuansome
committed
feat: add the relevant configuration of interceptors to opengemini-client-java
Signed-off-by: chenhuan <[email protected]>
1 parent fce3458 commit 9e6efb3

File tree

11 files changed

+393
-25
lines changed

11 files changed

+393
-25
lines changed

‎opengemini-client-api/pom.xml‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,4 +37,16 @@
3737
<version>${annotations.version}</version>
3838
</dependency>
3939
</dependencies>
40+
<build>
41+
<plugins>
42+
<plugin>
43+
<groupId>org.apache.maven.plugins</groupId>
44+
<artifactId>maven-compiler-plugin</artifactId>
45+
<configuration>
46+
<source>11</source>
47+
<target>11</target>
48+
</configuration>
49+
</plugin>
50+
</plugins>
51+
</build>
4052
</project>

‎opengemini-client-api/src/main/java/io/opengemini/client/api/Point.java‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import lombok.Setter;
2121

2222
import java.math.BigDecimal;
23+
import java.util.HashMap;
2324
import java.util.Map;
2425

2526
@Getter
@@ -40,6 +41,36 @@ public class Point {
4041
private Map<String, String> tags;
4142
private Map<String, Object> fields;
4243

44+
public Point() {
45+
this.fields = new HashMap<>();
46+
this.tags = new HashMap<>();
47+
}
48+
49+
public Point measurement(String measurement) {
50+
this.measurement = measurement;
51+
return this;
52+
}
53+
54+
public Point addTag(String key, String value) {
55+
if (key != null && value != null) {
56+
this.tags.put(key, value);
57+
}
58+
return this;
59+
}
60+
61+
public Point addField(String key, Object value) {
62+
if (key != null && value != null) {
63+
this.fields.put(key, value);
64+
}
65+
return this;
66+
}
67+
68+
public Point time(long time, Precision precision) {
69+
this.time = time;
70+
this.precision = precision;
71+
return this;
72+
}
73+
4374
/**
4475
* Calculate the line protocol string for this point
4576
*

‎opengemini-client-api/src/main/java/io/opengemini/client/api/Query.java‎

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,14 @@
1616

1717
package io.opengemini.client.api;
1818

19+
import io.opentelemetry.api.trace.Span;
1920
import lombok.AllArgsConstructor;
2021
import lombok.Getter;
2122
import lombok.Setter;
2223

24+
import java.util.HashMap;
25+
import java.util.Map;
26+
2327
@AllArgsConstructor
2428
@Getter
2529
@Setter
@@ -44,6 +48,8 @@ public class Query {
4448
*/
4549
private Precision precision;
4650

51+
private Map<String, Object> attributes = new HashMap<>();
52+
4753
public Query(String command) {
4854
this.command = command;
4955
}
@@ -53,4 +59,19 @@ public Query(String command, String database, String retentionPolicy) {
5359
this.database = database;
5460
this.retentionPolicy = retentionPolicy;
5561
}
56-
}
62+
63+
public Query(String command, String database, String retentionPolicy, Precision precision) {
64+
this.command = command;
65+
this.database = database;
66+
this.retentionPolicy = retentionPolicy;
67+
this.precision = precision;
68+
}
69+
70+
public void setAttribute(String key, Object value) {
71+
attributes.put(key, value);
72+
}
73+
74+
public Object getAttribute(String key) {
75+
return attributes.get(key);
76+
}
77+
}

‎opengemini-client-api/src/main/java/io/opengemini/client/api/Write.java‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
/*
2+
* Copyright 2024 openGemini Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.opengemini.client.api;
18+
19+
import lombok.Getter;
20+
import lombok.Setter;
21+
22+
import java.util.HashMap;
23+
import java.util.Map;
24+
25+
@Getter
26+
@Setter
27+
public class Write {
28+
private String database;
29+
private String retentionPolicy;
30+
private String measurement;
31+
private String lineProtocol;
32+
private String precision;
33+
34+
private Map<String, Object> attributes = new HashMap<>();
35+
36+
public Write(String database, String retentionPolicy, String measurement,
37+
String lineProtocol, String precision) {
38+
this.database = database;
39+
this.retentionPolicy = retentionPolicy;
40+
this.measurement = measurement;
41+
this.lineProtocol = lineProtocol;
42+
this.precision = precision;
43+
}
44+
45+
public void setAttribute(String key, Object value) {
46+
attributes.put(key, value);
47+
}
48+
49+
public Object getAttribute(String key) {
50+
return attributes.get(key);
51+
}
52+
}

‎opengemini-client-common/src/main/java/io/opengemini/client/common/BaseClient.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -162,4 +162,4 @@ protected String getQueryUrl(Query query) {
162162
public void close() throws IOException {
163163
scheduler.ifPresent(ExecutorService::shutdown);
164164
}
165-
}
165+
}

‎opengemini-client/pom.xml‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,5 +72,17 @@
7272
<version>${vertx.version}</version>
7373
</dependency>
7474
</dependencies>
75+
<build>
76+
<plugins>
77+
<plugin>
78+
<groupId>org.apache.maven.plugins</groupId>
79+
<artifactId>maven-compiler-plugin</artifactId>
80+
<configuration>
81+
<source>11</source>
82+
<target>11</target>
83+
</configuration>
84+
</plugin>
85+
</plugins>
86+
</build>
7587

7688
</project>

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

Lines changed: 83 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -21,33 +21,33 @@
2121
import io.github.openfacade.http.HttpClientConfig;
2222
import io.github.openfacade.http.HttpClientFactory;
2323
import io.github.openfacade.http.HttpResponse;
24-
import io.opengemini.client.api.AuthConfig;
25-
import io.opengemini.client.api.AuthType;
26-
import io.opengemini.client.api.Configuration;
27-
import io.opengemini.client.api.OpenGeminiAsyncClient;
28-
import io.opengemini.client.api.OpenGeminiException;
29-
import io.opengemini.client.api.Point;
30-
import io.opengemini.client.api.Pong;
31-
import io.opengemini.client.api.Query;
32-
import io.opengemini.client.api.QueryResult;
33-
import io.opengemini.client.api.RetentionPolicy;
34-
import io.opengemini.client.api.RpConfig;
24+
import io.opengemini.client.api.*;
3525
import io.opengemini.client.common.BaseClient;
3626
import io.opengemini.client.common.CommandFactory;
3727
import io.opengemini.client.common.HeaderConst;
3828
import io.opengemini.client.common.JacksonService;
3929
import io.opengemini.client.common.ResultMapper;
30+
import io.opengemini.client.interceptor.Interceptor;
4031
import org.apache.commons.lang3.StringUtils;
4132
import org.jetbrains.annotations.NotNull;
4233

4334
import java.io.IOException;
4435
import java.nio.charset.StandardCharsets;
45-
import java.util.List;
46-
import java.util.Optional;
47-
import java.util.StringJoiner;
36+
import java.util.*;
4837
import java.util.concurrent.CompletableFuture;
4938

5039
public class OpenGeminiClient extends BaseClient implements OpenGeminiAsyncClient {
40+
private final List<Interceptor> interceptors = new ArrayList<>();
41+
42+
/**
43+
* Add interceptors to the client.
44+
*/
45+
public void addInterceptors(Interceptor... interceptors) {
46+
Collections.addAll(this.interceptors, interceptors);
47+
}
48+
49+
50+
5151
protected final Configuration conf;
5252

5353
private final HttpClient client;
@@ -57,6 +57,10 @@ public OpenGeminiClient(@NotNull Configuration conf) {
5757
this.conf = conf;
5858
AuthConfig authConfig = conf.getAuthConfig();
5959
HttpClientConfig httpConfig = conf.getHttpConfig();
60+
if (httpConfig == null) {
61+
httpConfig = new HttpClientConfig.Builder().build();
62+
conf.setHttpConfig(httpConfig);
63+
}
6064
if (authConfig != null && authConfig.getAuthType().equals(AuthType.PASSWORD)) {
6165
httpConfig.addRequestFilter(
6266
new BasicAuthRequestFilter(authConfig.getUsername(), String.valueOf(authConfig.getPassword())));
@@ -159,7 +163,7 @@ public CompletableFuture<Void> write(String database, String retentionPolicy, Po
159163
if (StringUtils.isEmpty(body)) {
160164
return CompletableFuture.completedFuture(null);
161165
}
162-
return executeWrite(database, retentionPolicy, body);
166+
return executeWrite(database, retentionPolicy, body );
163167
}
164168

165169
@Override
@@ -195,9 +199,27 @@ public CompletableFuture<Pong> ping() {
195199
*
196200
* @param query the query to execute.
197201
*/
198-
protected CompletableFuture<QueryResult> executeQuery(Query query) {
202+
public CompletableFuture<QueryResult> executeQuery(Query query) {
199203
String queryUrl = getQueryUrl(query);
200-
return get(queryUrl).thenCompose(response -> convertResponse(response, QueryResult.class));
204+
205+
// 执行所有queryBefore拦截器
206+
CompletableFuture<Void> beforeFutures = CompletableFuture.allOf(
207+
interceptors.stream()
208+
.map(interceptor -> interceptor.queryBefore(query))
209+
.toArray(CompletableFuture[]::new)
210+
);
211+
212+
return beforeFutures.thenCompose(voidResult -> {
213+
return executeHttpQuery(query).thenCompose(response -> {
214+
// 执行所有queryAfter拦截器
215+
CompletableFuture<Void> afterFutures = CompletableFuture.allOf(
216+
interceptors.stream()
217+
.map(interceptor -> interceptor.queryAfter(query, response))
218+
.toArray(CompletableFuture[]::new)
219+
);
220+
return afterFutures.thenCompose(voidResult2 -> convertResponse(response, QueryResult.class));
221+
});
222+
});
201223
}
202224

203225
/**
@@ -217,9 +239,33 @@ protected CompletableFuture<QueryResult> executePostQuery(Query query) {
217239
* @param retentionPolicy the name of the retention policy.
218240
* @param lineProtocol the line protocol string to write.
219241
*/
220-
protected CompletableFuture<Void> executeWrite(String database, String retentionPolicy, String lineProtocol) {
242+
public CompletableFuture<Void> executeWrite(String database, String retentionPolicy, String lineProtocol) {
221243
String writeUrl = getWriteUrl(database, retentionPolicy);
222-
return post(writeUrl, lineProtocol).thenCompose(response -> convertResponse(response, Void.class));
244+
Write write = new Write(
245+
database,
246+
retentionPolicy,
247+
"default_measurement", // Default measurement name
248+
lineProtocol,
249+
"ns" // Default precision
250+
);
251+
252+
// Execute all writeBefore interceptors
253+
CompletableFuture<Void> beforeFutures = CompletableFuture.allOf(
254+
interceptors.stream()
255+
.map(interceptor -> interceptor.writeBefore(write))
256+
.toArray(CompletableFuture[]::new)
257+
);
258+
259+
return beforeFutures.thenCompose(voidResult -> {
260+
return executeHttpWrite(write).thenCompose(response -> { // response 是 io.github.openfacade.http.HttpResponse
261+
CompletableFuture<Void> afterFutures = CompletableFuture.allOf(
262+
interceptors.stream()
263+
.map(interceptor -> interceptor.writeAfter(write, response)) // 传递正确的类型
264+
.toArray(CompletableFuture[]::new)
265+
);
266+
return afterFutures.thenCompose(voidResult2 -> convertResponse(response, Void.class));
267+
});
268+
});
223269
}
224270

225271
/**
@@ -258,7 +304,7 @@ private CompletableFuture<HttpResponse> get(String url) {
258304

259305
private CompletableFuture<HttpResponse> post(String url, String body) {
260306
return client.post(buildUriWithPrefix(url), body == null ? new byte[0] : body.getBytes(StandardCharsets.UTF_8),
261-
headers);
307+
headers);
262308
}
263309

264310
@Override
@@ -270,4 +316,20 @@ public void close() throws IOException {
270316
public String toString() {
271317
return "OpenGeminiClient{" + "httpEngine=" + conf.getHttpConfig().engine() + '}';
272318
}
273-
}
319+
320+
/**
321+
* 执行 HTTP 查询请求
322+
*/
323+
private CompletableFuture<HttpResponse> executeHttpQuery(Query query) {
324+
String queryUrl = getQueryUrl(query);
325+
return get(queryUrl);
326+
}
327+
328+
/**
329+
* 执行 HTTP 写入请求
330+
*/
331+
private CompletableFuture<HttpResponse> executeHttpWrite(Write write) {
332+
String writeUrl = getWriteUrl(write.getDatabase(), write.getRetentionPolicy());
333+
return post(writeUrl, write.getLineProtocol());
334+
}
335+
}

‎opengemini-client/src/main/java/io/opengemini/client/impl/OpenGeminiClientFactory.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -54,4 +54,4 @@ public static OpenGeminiClient create(@NotNull Configuration configuration) thro
5454
}
5555
return new OpenGeminiClient(configuration);
5656
}
57-
}
57+
}

0 commit comments

Comments
 (0)