diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/EsClientConfig.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/EsClientConfig.java index 23b53c6..c061fcb 100644 --- a/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/EsClientConfig.java +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/EsClientConfig.java @@ -22,6 +22,8 @@ import java.util.Arrays; * 双集群 ES 客户端:cloud=云集群、local=本地化集群(老系统 Dictionary.LocalEsDate 分界)。 * 低阶 RestClient 也注册为 bean(destroyMethod=close 随容器关闭); * 高阶 ElasticsearchClient 包装同一 RestClient,关闭 client 即关闭底层连接。 + * 注意:生产两个集群均为 ES 7.7.1,高阶客户端(7.17)的 vendor Content-Type 与 + * X-Elastic-Product 校验与 7.7.1 不兼容(406),业务查询一律注入低阶 RestClient bean。 */ @Configuration public class EsClientConfig { diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/es/ElectricityMessageEsGatewayImpl.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/es/ElectricityMessageEsGatewayImpl.java index f33b940..130848b 100644 --- a/xiaoqu-agent-service/src/main/java/me/iiv/agent/es/ElectricityMessageEsGatewayImpl.java +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/es/ElectricityMessageEsGatewayImpl.java @@ -1,16 +1,12 @@ package me.iiv.agent.es; -import co.elastic.clients.elasticsearch.ElasticsearchClient; -import co.elastic.clients.elasticsearch._types.FieldValue; -import co.elastic.clients.elasticsearch._types.SortOrder; -import co.elastic.clients.elasticsearch._types.query_dsl.Query; -import co.elastic.clients.elasticsearch.core.SearchResponse; -import co.elastic.clients.elasticsearch.core.search.Hit; -import co.elastic.clients.json.JsonData; -import jakarta.json.JsonNumber; -import jakarta.json.JsonObject; -import jakarta.json.JsonString; -import jakarta.json.JsonValue; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.elasticsearch.client.Request; +import org.elasticsearch.client.Response; +import org.elasticsearch.client.RestClient; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; @@ -22,8 +18,12 @@ import java.util.ArrayList; import java.util.List; /** - * electricitymessage_yyyyMM ES 查询实现(elasticsearch-java 7.17 强类型 API), - * 只注入 esLocal 客户端(索引只在本地集群,不走 esCloud)。 + * electricitymessage_yyyyMM ES 查询实现。 + * 降级原因:生产 ES 集群为 7.7.1,elasticsearch-java 7.17 高阶客户端(ElasticsearchClient) + * 请求自带 vendor Content-Type(application/vnd.elasticsearch+json; compatible-with=7)被 + * 7.7.1 以 406 Not Acceptable 拒绝,且 7.17 客户端校验响应 X-Elastic-Product 头(7.14 才引入), + * 7.7.1 没有,二次兼容也过不了;故改用低阶 RestClient(只发普通 application/json、无产品头校验)。 + * 只注入 esLocalRestClient(索引只在本地集群,不走 esCloud)。 * 查询口径对齐老 BatteryAnalyseController.batteryAnalyseList 但修掉其坑: * snCode 用 match_phrase(老代码 matchQuery 经 standard analyzer 会把 AB-123 切成 ab/123 误配)、 * messageType 不传不加 term(老代码默认值 0 会把"不传"歧义成"电量")、createTime 闭区间 range、 @@ -37,47 +37,50 @@ public class ElectricityMessageEsGatewayImpl implements ElectricityMessageEsGate private static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").withZone(BUSINESS_ZONE); - private final ElasticsearchClient localClient; + private final RestClient localRestClient; + private final ObjectMapper mapper = new ObjectMapper(); - public ElectricityMessageEsGatewayImpl(@Qualifier("esLocalClient") ElasticsearchClient localClient) { - this.localClient = localClient; + public ElectricityMessageEsGatewayImpl(@Qualifier("esLocalRestClient") RestClient localRestClient) { + this.localRestClient = localRestClient; } @Override public Result search(String index, String snCode, Integer messageType, String startTime, String endTime, int from, int size) { - List filters = new ArrayList<>(); + ArrayNode filters = mapper.createArrayNode(); // 老代码 snCode 字符串走 matchQuery(BaseSearchSupportImpl2 type=0 非数值分支), // standard analyzer 会把 AB-123 切成 ab/123 造成误配;改用 match_phrase 精确短语 - filters.add(Query.of(q -> q.matchPhrase(m -> m.field("snCode").query(snCode)))); + filters.addObject().putObject("match_phrase").putObject("snCode").put("query", snCode); if (messageType != null) { - filters.add(Query.of(q -> q.term(t -> t.field("messageType") - .value(FieldValue.of(messageType.longValue()))))); + filters.addObject().putObject("term").putObject("messageType").put("value", messageType); } // 老代码 rangeQuery from(start,true).to(end,true),闭区间 - filters.add(Query.of(q -> q.range(r -> r.field("createTime") - .gte(JsonData.of(startTime)).lte(JsonData.of(endTime))))); + ObjectNode range = filters.addObject().putObject("range").putObject("createTime"); + range.put("gte", startTime).put("lte", endTime); + + ObjectNode body = mapper.createObjectNode(); + body.put("from", from).put("size", size); + body.put("track_total_hits", true); + ObjectNode bool = body.putObject("query").putObject("bool"); + bool.set("filter", filters); + body.putArray("sort").addObject().putObject("createTime").put("order", "asc"); try { - SearchResponse response = localClient.search(s -> s - .index(index) - .from(from) - .size(size) - .trackTotalHits(t -> t.enabled(true)) - .query(q -> q.bool(b -> b.filter(filters))) - .sort(so -> so.field(f -> f.field("createTime").order(SortOrder.Asc))), - JsonData.class); + Request req = new Request("POST", "/" + index + "/_search"); + // setJsonEntity 自动带 Content-Type: application/json(无 vendor 头) + req.setJsonEntity(mapper.writeValueAsString(body)); + Response resp = localRestClient.performRequest(req); + JsonNode root = mapper.readTree(resp.getEntity().getContent()); List records = new ArrayList<>(); - for (Hit hit : response.hits().hits()) { - JsonData source = hit.source(); + for (JsonNode hit : root.at("/hits/hits")) { + JsonNode source = hit.get("_source"); if (source == null) { continue; } - JsonObject obj = source.toJson().asJsonObject(); - records.add(new Record(createTimeText(obj.get("createTime")), - integerValue(obj.get("messageType")), - integerValue(obj.get("value")))); + records.add(new Record(createTimeText(source.get("createTime")), + integerValue(source.get("messageType")), + integerValue(source.get("value")))); } - return new Result(response.hits().total().value(), records); + return new Result(root.at("/hits/total/value").asLong(), records); } catch (IOException | RuntimeException e) { throw new IllegalStateException("ES 查询电量消息记录失败: " + e.getMessage(), e); } @@ -87,21 +90,20 @@ public class ElectricityMessageEsGatewayImpl implements ElectricityMessageEsGate * createTime 在索引里可能是 yyyy-MM-dd HH:mm:ss 文本(老 range 直接传字符串可证 ES 接受该格式), * 也可能是 epoch 毫秒数;两种形态都归一为 yyyy-MM-dd HH:mm:ss 文本输出。 */ - private static String createTimeText(JsonValue value) { - if (value == null || value.getValueType() == JsonValue.ValueType.NULL) { + private static String createTimeText(JsonNode value) { + if (value == null || value.isNull()) { return null; } - if (value instanceof JsonString) { - return ((JsonString) value).getString(); + if (value.isTextual()) { + return value.asText(); } - if (value instanceof JsonNumber) { - return TIME_FORMATTER.format(Instant.ofEpochMilli(((JsonNumber) value).longValue())); + if (value.isNumber()) { + return TIME_FORMATTER.format(Instant.ofEpochMilli(value.longValue())); } return null; } - private static Integer integerValue(JsonValue value) { - return value != null && value.getValueType() == JsonValue.ValueType.NUMBER - ? ((JsonNumber) value).intValue() : null; + private static Integer integerValue(JsonNode value) { + return value != null && value.isNumber() ? value.intValue() : null; } } diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/es/PaperUseLogEsGatewayImpl.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/es/PaperUseLogEsGatewayImpl.java index 32c4edc..a1b182c 100644 --- a/xiaoqu-agent-service/src/main/java/me/iiv/agent/es/PaperUseLogEsGatewayImpl.java +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/es/PaperUseLogEsGatewayImpl.java @@ -1,19 +1,24 @@ package me.iiv.agent.es; -import co.elastic.clients.elasticsearch.ElasticsearchClient; -import co.elastic.clients.elasticsearch._types.FieldValue; -import co.elastic.clients.elasticsearch._types.query_dsl.Query; -import co.elastic.clients.json.JsonData; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.elasticsearch.client.Request; +import org.elasticsearch.client.Response; +import org.elasticsearch.client.RestClient; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; import org.springframework.util.StringUtils; import java.io.IOException; -import java.util.ArrayList; -import java.util.List; /** - * paperuselog ES 查询实现(elasticsearch-java 7.17 强类型 API)。 + * paperuselog ES 查询实现。 + * 降级原因:生产 ES 集群为 7.7.1,elasticsearch-java 7.17 高阶客户端(ElasticsearchClient) + * 请求自带 vendor Content-Type(application/vnd.elasticsearch+json; compatible-with=7)被 + * 7.7.1 以 406 Not Acceptable 拒绝,且 7.17 客户端校验响应 X-Elastic-Product 头(7.14 才引入), + * 7.7.1 没有,二次兼容也过不了;故改用低阶 RestClient(只发普通 application/json、无产品头校验)。 * 查询口径对齐老 MonthlyStatisticsController.getMonthlyStatisticsMap: * term districtId/gender、range createTime [gte,lte](含边界)、deviceType 用 match(老代码字符串参数走 match), * positionMode 厕位=positionsId>0、厕墙=positionsId=0。 @@ -25,31 +30,37 @@ public class PaperUseLogEsGatewayImpl implements PaperUseLogEsGateway { /** rfid terms 聚合去重上限,超出部分不计入 */ private static final int RFID_TERMS_SIZE = 10000; - private final ElasticsearchClient cloudClient; - private final ElasticsearchClient localClient; + private final RestClient cloudRestClient; + private final RestClient localRestClient; + private final ObjectMapper mapper = new ObjectMapper(); - public PaperUseLogEsGatewayImpl(@Qualifier("esCloudClient") ElasticsearchClient cloudClient, - @Qualifier("esLocalClient") ElasticsearchClient localClient) { - this.cloudClient = cloudClient; - this.localClient = localClient; + public PaperUseLogEsGatewayImpl(@Qualifier("esCloudRestClient") RestClient cloudRestClient, + @Qualifier("esLocalRestClient") RestClient localRestClient) { + this.cloudRestClient = cloudRestClient; + this.localRestClient = localRestClient; } @Override public long distinctRfidCount(String index, boolean localCluster, Long districtId, Integer gender, Long fromMillis, Long toMillis, Integer deviceType, String positionMode) { - List filters = baseFilters(districtId, gender, fromMillis, toMillis, deviceType, positionMode); + ArrayNode filters = baseFilters(districtId, gender, fromMillis, toMillis, deviceType, positionMode); // 上纸(2)+撤纸(4) 合并后 rfid 去重(老 :302-342) - filters.add(Query.of(q -> q.terms(t -> t.field("state") - .terms(tt -> tt.value(List.of(FieldValue.of(2), FieldValue.of(4))))))); + ArrayNode stateTerms = filters.addObject().putObject("terms").putObject("state").putArray("terms"); + stateTerms.add(2).add(4); try { - return client(localCluster).search(s -> s - .index(index) - .size(0) - .trackTotalHits(t -> t.enabled(true)) - .query(q -> q.bool(b -> b.filter(filters))) - .aggregations("rfids", a -> a.terms(t -> t.field("rfid").size(RFID_TERMS_SIZE))), - Void.class) - .aggregations().get("rfids").sterms().buckets().array().size(); + ObjectNode body = mapper.createObjectNode(); + body.put("size", 0); + body.put("track_total_hits", false); + ObjectNode bool = body.putObject("query").putObject("bool"); + bool.set("filter", filters); + body.putObject("aggs").putObject("rfids").putObject("terms") + .put("field", "rfid").put("size", RFID_TERMS_SIZE); + + Request req = new Request("POST", "/" + index + "/_search"); + req.setJsonEntity(mapper.writeValueAsString(body)); + Response resp = client(localCluster).performRequest(req); + JsonNode root = mapper.readTree(resp.getEntity().getContent()); + return root.at("/aggregations/rfids/buckets").size(); } catch (IOException | RuntimeException e) { throw new IllegalStateException("ES 查询换纸 rfid 去重失败: " + e.getMessage(), e); } @@ -58,40 +69,47 @@ public class PaperUseLogEsGatewayImpl implements PaperUseLogEsGateway { @Override public long countState2(String index, boolean localCluster, Long districtId, Integer gender, Long fromMillis, Long toMillis, Integer deviceType, String positionMode) { - List filters = baseFilters(districtId, gender, fromMillis, toMillis, deviceType, positionMode); - filters.add(Query.of(q -> q.term(t -> t.field("state").value(FieldValue.of(2))))); + ArrayNode filters = baseFilters(districtId, gender, fromMillis, toMillis, deviceType, positionMode); + filters.addObject().putObject("term").putObject("state").put("value", 2); try { - return client(localCluster).count(c -> c.index(index) - .query(q -> q.bool(b -> b.filter(filters)))).count(); + ObjectNode body = mapper.createObjectNode(); + ObjectNode bool = body.putObject("query").putObject("bool"); + bool.set("filter", filters); + + Request req = new Request("POST", "/" + index + "/_count"); + req.setJsonEntity(mapper.writeValueAsString(body)); + Response resp = client(localCluster).performRequest(req); + JsonNode root = mapper.readTree(resp.getEntity().getContent()); + return root.get("count").asLong(); } catch (IOException | RuntimeException e) { throw new IllegalStateException("ES 查询换纸上纸记录数失败: " + e.getMessage(), e); } } - private List baseFilters(Long districtId, Integer gender, Long fromMillis, Long toMillis, - Integer deviceType, String positionMode) { - List filters = new ArrayList<>(); - filters.add(Query.of(q -> q.term(t -> t.field("districtId").value(FieldValue.of(districtId))))); - filters.add(Query.of(q -> q.term(t -> t.field("gender").value(FieldValue.of(gender))))); + private ArrayNode baseFilters(Long districtId, Integer gender, Long fromMillis, Long toMillis, + Integer deviceType, String positionMode) { + ArrayNode filters = mapper.createArrayNode(); + filters.addObject().putObject("term").putObject("districtId").put("value", districtId); + filters.addObject().putObject("term").putObject("gender").put("value", gender); // createTime 为 epoch 毫秒;含边界(老 rangeQuery from/to 均 inclusive) - filters.add(Query.of(q -> q.range(r -> r.field("createTime") - .gte(JsonData.of(fromMillis)).lte(JsonData.of(toMillis))))); + ObjectNode range = filters.addObject().putObject("range").putObject("createTime"); + range.put("gte", fromMillis).put("lte", toMillis); if (deviceType != null) { // 老代码 deviceType 为字符串参数走 matchQuery(BaseSearchSupportImpl2 type=0 非数值分支) - filters.add(Query.of(q -> q.match(m -> m.field("deviceType") - .query(FieldValue.of(String.valueOf(deviceType)))))); + filters.addObject().putObject("match").putObject("deviceType") + .put("query", String.valueOf(deviceType)); } if (StringUtils.hasText(positionMode)) { if ("position".equals(positionMode)) { - filters.add(Query.of(q -> q.range(r -> r.field("positionsId").gt(JsonData.of(0))))); + filters.addObject().putObject("range").putObject("positionsId").put("gt", 0); } else { - filters.add(Query.of(q -> q.term(t -> t.field("positionsId").value(FieldValue.of(0))))); + filters.addObject().putObject("term").putObject("positionsId").put("value", 0); } } return filters; } - private ElasticsearchClient client(boolean localCluster) { - return localCluster ? localClient : cloudClient; + private RestClient client(boolean localCluster) { + return localCluster ? localRestClient : cloudRestClient; } } diff --git a/xiaoqu-agent-service/src/test/java/me/iiv/agent/es/ElectricityMessageEsGatewayImplTest.java b/xiaoqu-agent-service/src/test/java/me/iiv/agent/es/ElectricityMessageEsGatewayImplTest.java index 40f39b6..0aa2ea1 100644 --- a/xiaoqu-agent-service/src/test/java/me/iiv/agent/es/ElectricityMessageEsGatewayImplTest.java +++ b/xiaoqu-agent-service/src/test/java/me/iiv/agent/es/ElectricityMessageEsGatewayImplTest.java @@ -1,12 +1,13 @@ package me.iiv.agent.es; -import co.elastic.clients.elasticsearch.ElasticsearchClient; -import co.elastic.clients.elasticsearch.core.SearchRequest; -import co.elastic.clients.elasticsearch.core.SearchResponse; -import co.elastic.clients.elasticsearch.core.search.Hit; -import co.elastic.clients.elasticsearch.core.search.TotalHitsRelation; -import co.elastic.clients.json.JsonData; -import co.elastic.clients.util.ObjectBuilder; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.http.entity.ContentType; +import org.apache.http.entity.StringEntity; +import org.apache.http.util.EntityUtils; +import org.elasticsearch.client.Request; +import org.elasticsearch.client.Response; +import org.elasticsearch.client.RestClient; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -14,103 +15,107 @@ import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; -import java.util.List; -import java.util.function.Function; +import java.io.IOException; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @ExtendWith(MockitoExtension.class) class ElectricityMessageEsGatewayImplTest { + /** 2026-09-21 14:00:00 Asia/Shanghai 的 epoch 毫秒(= 2026-09-21T06:00:00Z) */ + private static final long EPOCH_MILLIS = 1789970400000L; + + private static final String RESPONSE_JSON = "{" + + "\"took\":1,\"timed_out\":false," + + "\"_shards\":{\"total\":1,\"successful\":1,\"skipped\":0,\"failed\":0}," + + "\"hits\":{\"total\":{\"value\":2,\"relation\":\"eq\"}," + + "\"hits\":[" + + "{\"_index\":\"electricitymessage_202609\"," + + "\"_source\":{\"createTime\":\"2026-09-21 10:00:00\",\"messageType\":0,\"value\":88}}," + + "{\"_index\":\"electricitymessage_202609\"," + + "\"_source\":{\"createTime\":" + EPOCH_MILLIS + ",\"messageType\":4,\"value\":3700}}" + + "]}}"; + @Mock - ElasticsearchClient localClient; + RestClient localRestClient; ElectricityMessageEsGatewayImpl gateway; + private final ObjectMapper mapper = new ObjectMapper(); + @BeforeEach void setUp() { - gateway = new ElectricityMessageEsGatewayImpl(localClient); + gateway = new ElectricityMessageEsGatewayImpl(localRestClient); } - private SearchResponse responseWith(String... sources) { - List> hits = java.util.Arrays.stream(sources) - .map(json -> Hit.of(hit -> hit - .index("electricitymessage_202609") - .source(JsonData.fromJson(json)))) - .toList(); - return SearchResponse.of(r -> r - .took(1) - .timedOut(false) - .shards(s -> s.total(1).successful(1).skipped(0).failed(0)) - .hits(h -> h.total(t -> t.value(2).relation(TotalHitsRelation.Eq)) - .hits(hits))); + private static Response esResponse(String jsonBody) throws IOException { + Response response = org.mockito.Mockito.mock(Response.class); + when(response.getEntity()).thenReturn(new StringEntity(jsonBody, ContentType.APPLICATION_JSON)); + return response; } - @SuppressWarnings("unchecked") - private SearchRequest captureRequest(SearchResponse response, Integer messageType) throws Exception { - when(localClient.search(any(Function.class), eq(JsonData.class))).thenReturn(response); + private JsonNode captureRequestBody(String responseJson, Integer messageType) throws Exception { + Response esResponse = esResponse(responseJson); + when(localRestClient.performRequest(any(Request.class))).thenReturn(esResponse); gateway.search("electricitymessage_202609", "AB-123", messageType, "2026-09-21 00:00:00", "2026-09-21 23:59:59", 0, 500); - ArgumentCaptor>> captor = - ArgumentCaptor.forClass(Function.class); - verify(localClient).search(captor.capture(), eq(JsonData.class)); - return captor.getValue().apply(new SearchRequest.Builder()).build(); + ArgumentCaptor captor = ArgumentCaptor.forClass(Request.class); + verify(localRestClient).performRequest(captor.capture()); + Request captured = captor.getValue(); + assertEquals("POST", captured.getMethod()); + assertEquals("/electricitymessage_202609/_search", captured.getEndpoint()); + return mapper.readTree(EntityUtils.toString(captured.getEntity())); + } + + private static boolean anyFilter(JsonNode filters, java.util.function.Predicate predicate) { + for (JsonNode filter : filters) { + if (predicate.test(filter)) { + return true; + } + } + return false; } @Test - @SuppressWarnings("unchecked") void buildsMatchPhraseOptionalTermRangeSortAndPagination() throws Exception { - SearchRequest request = captureRequest(responseWith( - "{\"createTime\":\"2026-09-21 10:00:00\",\"messageType\":0,\"value\":88}"), 0); + JsonNode body = captureRequestBody(RESPONSE_JSON, 0); - assertEquals(List.of("electricitymessage_202609"), request.index()); - assertEquals(Integer.valueOf(0), request.from()); - assertEquals(Integer.valueOf(500), request.size()); - assertTrue(request.trackTotalHits().enabled()); - assertEquals(1, request.sort().size()); - assertEquals("createTime", request.sort().get(0).field().field()); - assertEquals(co.elastic.clients.elasticsearch._types.SortOrder.Asc, - request.sort().get(0).field().order()); + assertEquals(0, body.get("from").asInt()); + assertEquals(500, body.get("size").asInt()); + assertTrue(body.get("track_total_hits").asBoolean()); + assertEquals("asc", body.get("sort").get(0).get("createTime").get("order").asText()); - List filters = - request.query().bool().filter(); + JsonNode filters = body.at("/query/bool/filter"); // snCode 必须 match_phrase(老系统 matchQuery 分词误配的修复) - assertTrue(filters.stream().anyMatch(q -> q.isMatchPhrase() - && q.matchPhrase().field().equals("snCode") - && q.matchPhrase().query().equals("AB-123"))); + assertTrue(anyFilter(filters, f -> f.has("match_phrase") + && f.get("match_phrase").get("snCode").get("query").asText().equals("AB-123"))); // messageType 传了才加 term - assertTrue(filters.stream().anyMatch(q -> q.isTerm() - && q.term().field().equals("messageType") - && q.term().value().longValue() == 0)); + assertTrue(anyFilter(filters, f -> f.has("term") + && f.get("term").get("messageType").get("value").asInt() == 0)); // createTime 闭区间 range - assertTrue(filters.stream().anyMatch(q -> q.isRange() - && q.range().field().equals("createTime") - && q.range().gte().to(String.class).equals("2026-09-21 00:00:00") - && q.range().lte().to(String.class).equals("2026-09-21 23:59:59"))); + assertTrue(anyFilter(filters, f -> f.has("range") + && f.get("range").get("createTime").get("gte").asText().equals("2026-09-21 00:00:00") + && f.get("range").get("createTime").get("lte").asText().equals("2026-09-21 23:59:59"))); } @Test - @SuppressWarnings("unchecked") void omitsMessageTypeTermWhenNull() throws Exception { - SearchRequest request = captureRequest(responseWith( - "{\"createTime\":\"2026-09-21 10:00:00\",\"messageType\":1,\"value\":-60}"), null); - assertTrue(request.query().bool().filter().stream().noneMatch(q -> q.isTerm())); + JsonNode body = captureRequestBody(RESPONSE_JSON, null); + for (JsonNode filter : body.at("/query/bool/filter")) { + assertTrue(!filter.has("term") || filter.get("term").findValue("messageType") == null); + } } @Test - @SuppressWarnings("unchecked") void mapsHitsAndEpochMillisCreateTime() throws Exception { - long epochMillis = java.time.Instant.parse("2026-09-21T06:00:00Z").toEpochMilli(); - when(localClient.search(any(Function.class), eq(JsonData.class))).thenReturn(responseWith( - "{\"createTime\":\"2026-09-21 10:00:00\",\"messageType\":0,\"value\":88}", - "{\"createTime\":" + epochMillis + ",\"messageType\":4,\"value\":3700}")); + Response esResponse = esResponse(RESPONSE_JSON); + when(localRestClient.performRequest(any(Request.class))).thenReturn(esResponse); ElectricityMessageEsGateway.Result result = gateway.search("electricitymessage_202609", "AB-123", null, "2026-09-21 00:00:00", "2026-09-21 23:59:59", 0, 500); @@ -124,13 +129,15 @@ class ElectricityMessageEsGatewayImplTest { // epoch 毫秒形态归一为 Asia/Shanghai yyyy-MM-dd HH:mm:ss assertEquals("2026-09-21 14:00:00", result.getRecords().get(1).getCreateTime()); assertEquals(Integer.valueOf(4), result.getRecords().get(1).getMessageType()); + assertEquals(Integer.valueOf(3700), result.getRecords().get(1).getValue()); } @Test - @SuppressWarnings("unchecked") void mapsNullAndMissingFields() throws Exception { - when(localClient.search(any(Function.class), eq(JsonData.class))).thenReturn(responseWith( - "{\"createTime\":null}")); + String response = "{\"hits\":{\"total\":{\"value\":1,\"relation\":\"eq\"}," + + "\"hits\":[{\"_source\":{\"createTime\":null}}]}}"; + Response esResponse = esResponse(response); + when(localRestClient.performRequest(any(Request.class))).thenReturn(esResponse); ElectricityMessageEsGateway.Record record = gateway.search("electricitymessage_202609", "AB-123", null, "2026-09-21 00:00:00", "2026-09-21 23:59:59", 0, 500) @@ -142,10 +149,9 @@ class ElectricityMessageEsGatewayImplTest { } @Test - @SuppressWarnings("unchecked") void wrapsEsFailureInIllegalState() throws Exception { - when(localClient.search(any(Function.class), eq(JsonData.class))) - .thenThrow(new java.io.IOException("connection reset")); + when(localRestClient.performRequest(any(Request.class))) + .thenThrow(new IOException("connection reset")); IllegalStateException exception = assertThrows(IllegalStateException.class, () -> gateway.search("electricitymessage_202609", "AB-123", null,