From 46a77a196a4ac0d059f859426f389d919114abd0 Mon Sep 17 00:00:00 2001
From: vinci <1552443053@qq.com>
Date: Sun, 22 Oct 2023 23:10:47 +0800
Subject: [PATCH] support push style for multiple messages (#1292)

---
 .../collect/push/PushCollectImpl.java         | 23 ++++++++++---------
 .../common/entity/push/PushMetricsDto.java    |  2 +-
 .../push/service/impl/PushServiceImpl.java    |  7 +++---
 3 files changed, 16 insertions(+), 16 deletions(-)

diff --git a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/push/PushCollectImpl.java b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/push/PushCollectImpl.java
index 314c7da30c3..57c454d2349 100644
--- a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/push/PushCollectImpl.java
+++ b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/push/PushCollectImpl.java
@@ -138,11 +138,6 @@ private HttpUriRequest createHttpRequest(PushProtocol pushProtocol, Long monitor
     }
 
     private void parseResponse(CollectRep.MetricsData.Builder builder, String resp, Metrics metric) {
-//        Map<String, Object> jsonMap = JsonUtil.fromJson(resp, new TypeReference<Map<String, Object>>() {
-//        });
-//        if (jsonMap == null) {
-//            throw new NullPointerException("parse result is null");
-//        }
         Message<PushMetricsDto> msg = JsonUtil.fromJson(resp, new TypeReference<Message<PushMetricsDto>>() {
         });
         if (msg == null) {
@@ -153,13 +148,19 @@ private void parseResponse(CollectRep.MetricsData.Builder builder, String resp,
             throw new NullPointerException("parse result is null");
         }
         for (PushMetricsDto.Metrics pushMetrics : pushMetricsDto.getMetricsList()) {
-            List<String> metricColumn = new ArrayList<>();
-            for (Metrics.Field field : metric.getFields()) {
-                metricColumn.add(pushMetrics.getMetrics().get(field.getField()));
+            List<CollectRep.ValueRow> rows = new ArrayList<>();
+            for (Map<String, String> metrics : pushMetrics.getMetrics()) {
+                List<String> metricColumn = new ArrayList<>();
+                for (Metrics.Field field : metric.getFields()) {
+                    metricColumn.add(metrics.get(field.getField()));
+                }
+                CollectRep.ValueRow valueRow = CollectRep.ValueRow.newBuilder()
+                        .addAllColumns(metricColumn).build();
+                rows.add(valueRow);
             }
-            CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder()
-                    .addAllColumns(metricColumn);
-            builder.addValues(valueRowBuilder.build());
+
+
+            builder.addAllValues(rows);
         }
         builder.setTime(System.currentTimeMillis());
     }
diff --git a/common/src/main/java/org/dromara/hertzbeat/common/entity/push/PushMetricsDto.java b/common/src/main/java/org/dromara/hertzbeat/common/entity/push/PushMetricsDto.java
index e4a8ca27166..b764d58652d 100644
--- a/common/src/main/java/org/dromara/hertzbeat/common/entity/push/PushMetricsDto.java
+++ b/common/src/main/java/org/dromara/hertzbeat/common/entity/push/PushMetricsDto.java
@@ -32,6 +32,6 @@ public PushMetricsDto() {
     public static class Metrics {
         private long monitorId;
         private Long time;
-        private Map<String, String> metrics;
+        private List<Map<String, String>> metrics;
     }
 }
diff --git a/push/src/main/java/org/dromara/hertzbeat/push/service/impl/PushServiceImpl.java b/push/src/main/java/org/dromara/hertzbeat/push/service/impl/PushServiceImpl.java
index 6c4b9a04203..0faf8a1da49 100644
--- a/push/src/main/java/org/dromara/hertzbeat/push/service/impl/PushServiceImpl.java
+++ b/push/src/main/java/org/dromara/hertzbeat/push/service/impl/PushServiceImpl.java
@@ -69,7 +69,7 @@ public void pushMetricsData(PushMetricsDto pushMetricsDto) throws RuntimeExcepti
             long monitorId = metrics.getMonitorId();
             metrics.setTime(curTime);
 
-            if (!monitorIdCache.containsKey(monitorId) && (monitorIdCache.containsKey(monitorId) && curTime > monitorIdCache.get(monitorId) + cacheTimeout)) {
+            if (!monitorIdCache.containsKey(monitorId) || (monitorIdCache.containsKey(monitorId) && curTime > monitorIdCache.get(monitorId) + cacheTimeout)) {
                 Optional<Monitor> queryOption = monitorDao.findById(monitorId);
                 if (queryOption.isEmpty()) {
                     monitorIdCache.remove(monitorId);
@@ -103,7 +103,7 @@ public PushMetricsDto getPushMetricData(final Long monitorId, final Long time) {
                 if (pushMetrics == null || pushMetrics.getMetrics() == null) {
                     return pushMetricsDto;
                 }
-                Map<String, String> jsonMap = JsonUtil.fromJson(pushMetrics.getMetrics(), new TypeReference<Map<String, String>>() {
+                List<Map<String, String>> jsonMap = JsonUtil.fromJson(pushMetrics.getMetrics(), new TypeReference<List<Map<String, String>>>() {
                 });
                 metrics = PushMetricsDto.Metrics.builder().monitorId(monitorId).metrics(jsonMap).time(pushMetrics.getTime()).build();
                 lastPushMetrics.put(monitorId, metrics);
@@ -114,11 +114,10 @@ public PushMetricsDto getPushMetricData(final Long monitorId, final Long time) {
             }
         }
         if (time > metrics.getTime()) {
+            // return void because time param is invalid
             return pushMetricsDto;
         }
         pushMetricsDto.getMetricsList().add(metrics);
-        // 目前先不删除
-        // metricsDao.deleteAllById(toBeDelMetricsId);
         return pushMetricsDto;
     }