1.新增entity delete线程 2.update 操作去除delete操作

3 files changed, 84 insertions, 18 deletions

diff --git a/src/main/java/com/example/nis/Scheduled/Entity.java b/src/main/java/com/example/nis/Scheduled/Entity.java
index 31c9252..0c27a3d 100644
--- a/src/main/java/com/example/nis/Scheduled/Entity.java
+++ b/src/main/java/com/example/nis/Scheduled/Entity.java
@@ -1,5 +1,7 @@
 package com.example.nis.Scheduled;
 
+import cn.hutool.core.date.DateTime;
+import cn.hutool.core.date.DateUtil;
 import cn.hutool.core.util.BooleanUtil;
 import cn.hutool.core.util.StrUtil;
 import cn.hutool.http.HttpResponse;
@@ -22,6 +24,7 @@ import org.springframework.stereotype.Component;
 import org.springframework.web.bind.annotation.RequestBody;
 import sun.net.util.IPAddressUtil;
 
+import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -48,27 +51,29 @@ public class Entity {
     @Value("${entity.update.sql}")
     private String sql;
 
+
+    @Value("${entity.delete.ipObjectId}")
+    private Integer deleteIpObjectId;
+    @Value("${entity.delete.ipObjectName}")
+    private String deleteIpObjectName;
+    @Value("${entity.delete.enable}")
+    private Boolean deleteEnable;
+    @Value("${entity.delete.offset}")
+    private Integer deleteOffset;
+
     @Scheduled(cron = "${entity.update.cron}")
-    public ResponseData scheduledExecutorOfPsiphon() {
+    public ResponseData scheduledExecutorOfUpdate() {
+        log.info("start entity update schedule");
         if (BooleanUtil.isFalse(updateEnable)) {
             log.warn("update schedule disable");
             return ResponseData.ok();
         }
-        ResponseData responseData = getAllByIpObjectId(updateIpObjectId);
-        if (!Code.SUCCESS.getCode().equals(responseData.get("code"))) {
-            String msg = String.valueOf(responseData.get("msg"));
-            log.error("get all itemList error: {}", msg);
-            return ResponseData.error(msg);
-        }
-
-        //String sql = "select distinct client_ip from psiphon_client_detection_event where event_time >= now()-600 AND event_time < now() AND notEmpty(client_ip)";
         HttpResponse httpResponse = ToCKDBUtil.exeQuery(sql);
         if (httpResponse.getStatus() != Code.SUCCESS.getCode()) {
             String msg = String.valueOf(httpResponse);
             log.error("get client ip error: {}", msg);
             return ResponseData.error(msg);
         }
-
         Map responseBody = JSONUtil.toBean(httpResponse.body(), Map.class);
         JSONArray dataJson = JSONUtil.parseArray(responseBody.get("data"));
         List<Map> data = JSONUtil.toList(dataJson, Map.class);
@@ -76,15 +81,8 @@ public class Entity {
             log.warn("ips is empty, No update");
             return ResponseData.ok();
         }
-
+        log.info("wait to add count: {}", data.size());
         IpSource ipSource = new IpSource();
-        List<IpObject> deleteList = new ArrayList<>();
-        JSONArray jsonArray = JSONUtil.parseArray(responseData.get("data"));
-        for (int i = 0; i < jsonArray.size(); i++) {
-            JSONObject obj = jsonArray.getJSONObject(i);
-            deleteList.add(JSONUtil.toBean(obj, IpObject.class));
-        }
-        ipSource.setDeleteItemList(deleteList);
         List<IpObject> addList = new ArrayList<>();
         for (Map map : data) {
             if (StrUtil.isBlankIfStr(map) || IpController.excludeList.contains(map)) {
@@ -104,6 +102,33 @@ public class Entity {
         return putPsiphon(ipSource);
     }
 
+    @Scheduled(cron = "${entity.delete.cron}")
+    public ResponseData scheduledExecutorOfDelete() {
+        log.info("start entity delete schedule");
+        if (BooleanUtil.isFalse(deleteEnable)) {
+            log.warn("delete schedule disable");
+            return ResponseData.ok();
+        }
+        ResponseData responseData = getDeleteByIpObjectId(deleteIpObjectId);
+        if (!Code.SUCCESS.getCode().equals(responseData.get("code"))) {
+            String msg = String.valueOf(responseData.get("msg"));
+            log.error("get all itemList error: {}", msg);
+            return ResponseData.error(msg);
+        }
+        IpSource ipSource = new IpSource();
+        List<IpObject> deleteList = new ArrayList<>();
+        JSONArray jsonArray = JSONUtil.parseArray(responseData.get("data"));
+        for (int i = 0; i < jsonArray.size(); i++) {
+            JSONObject obj = jsonArray.getJSONObject(i);
+            deleteList.add(JSONUtil.toBean(obj, IpObject.class));
+        }
+        if (deleteList.isEmpty()) {
+            log.info("deleteList is empty, no execute delete schedule");
+        }
+        ipSource.setDeleteItemList(deleteList);
+        return putPsiphon(ipSource);
+    }
+
 
     public ResponseData getAllByIpObjectId(Integer ipObjectId) {
         try {
@@ -127,6 +152,39 @@ public class Entity {
         }
     }
 
+
+    public ResponseData getDeleteByIpObjectId(Integer ipObjectId) {
+        try {
+            JSONArray itemList = tsgService.getItemList(ipObjectId, Constant.TSG_ITEM_IP, null, null, null, null, null);
+            if (itemList == null || itemList.isEmpty()) {
+                return ResponseData.ok(new ArrayList<>());
+            }
+            List<IpObject> result = new ArrayList<>(itemList.size());
+            DateTime now = DateUtil.offsetSecond(DateTime.now(), deleteOffset);
+            for (int i = 0; i < itemList.size(); i++) {
+                JSONObject jsonObject = itemList.getJSONObject(i);
+                DateTime dateTime = new DateTime(jsonObject.get(Constant.TSG_OP_TIME).toString(), DateTimeFormatter.ISO_DATE_TIME);
+                if (DateUtil.compare(now, dateTime) <= 0) {
+                    continue;
+                }
+                IpObject ipObject = new IpObject();
+                ipObject.setItemId(Integer.parseInt(jsonObject.get(Constant.TSG_ITEM_ID).toString()));
+                ipObject.setIp(jsonObject.get(Constant.TSG_ITEM_IP).toString());
+                ipObject.setPort(jsonObject.get(Constant.TSG_ITEM_PORT).toString());
+                result.add(ipObject);
+            }
+            if (itemList.size() == result.size()) {
+                log.info("all itemList expired, won't set delete items empty");
+                result = new ArrayList<>();
+            }
+            log.info("wait to delete count: {}", result.size());
+            return ResponseData.ok(result);
+        } catch (Exception e) {
+            log.error("get All itemList error: {}", e.getMessage());
+            return ResponseData.error();
+        }
+    }
+
     public ResponseData putPsiphon(@RequestBody IpSource source) {
         try {
             JSONArray array = new JSONArray();
diff --git a/src/main/java/com/example/nis/util/Constant.java b/src/main/java/com/example/nis/util/Constant.java
index 2f5b6c1..c364c48 100644
--- a/src/main/java/com/example/nis/util/Constant.java
+++ b/src/main/java/com/example/nis/util/Constant.java
@@ -105,6 +105,7 @@ public class Constant {
     public static String TSG_OBJECT_ID = "objectId";
     public static String TSG_ITEM_LIST = "itemList";
     public static String TSG_ITEM_ID = "itemId";
+    public static String TSG_OP_TIME = "opTime";
     public static String TSG_ITEM_NAME = "itemName";
     public static String TSG_ITEM_DESC = "itemDesc";
     public static String TSG_ITEM_IDS = "itemIds";
diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
index edf83c7..861ff44 100644
--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@ -36,6 +36,13 @@ entity:
     sql: SELECT DISTINCT client_ip FROM psiphon_client_detection_event WHERE event_time >= now()-600 AND event_time < now() AND notEmpty(client_ip)
     ipObjectId: 8533
     ipObjectName: Psiphon_Client_IP
+  delete:
+    cron: 0 0/5 * * * ?
+    enable: true
+    #偏移秒数，正数向未来偏移，负数向历史偏移
+    offset: -600
+    ipObjectId: 8533
+    ipObjectName: Psiphon_Client_IP
 ## ClickhHouse configuration
 clickhouse:
   url: http://192.168.44.12:8123