From c036483680e7c60b26def1eba34b9c867de0fde7 Mon Sep 17 00:00:00 2001
From: zengqiao <zengqiao_cn@163.com>
Date: Tue, 11 Jan 2022 17:48:24 +0800
Subject: [PATCH] add region created event and handle it to cal region capacity
 immediately

---
 .../common/events/RegionCreatedEvent.java     | 20 ++++++++++
 .../service/impl/RegionServiceImpl.java       |  4 ++
 .../listener/biz/RegionCreatedListener.java   | 38 +++++++++++++++++++
 3 files changed, 62 insertions(+)
 create mode 100644 kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/events/RegionCreatedEvent.java
 create mode 100644 kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/biz/RegionCreatedListener.java

diff --git a/kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/events/RegionCreatedEvent.java b/kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/events/RegionCreatedEvent.java
new file mode 100644
index 00000000..b8d72de9
--- /dev/null
+++ b/kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/events/RegionCreatedEvent.java
@@ -0,0 +1,20 @@
+package com.xiaojukeji.kafka.manager.common.events;
+
+import com.xiaojukeji.kafka.manager.common.entity.pojo.RegionDO;
+import lombok.Getter;
+import org.springframework.context.ApplicationEvent;
+
+/**
+ * Region创建事件
+ * @author zengqiao
+ * @date 22/01/1
+ */
+@Getter
+public class RegionCreatedEvent extends ApplicationEvent {
+    private final RegionDO regionDO;
+
+    public RegionCreatedEvent(Object source, RegionDO regionDO) {
+        super(source);
+        this.regionDO = regionDO;
+    }
+}
diff --git a/kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/RegionServiceImpl.java b/kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/RegionServiceImpl.java
index 8f957b02..3b2df843 100644
--- a/kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/RegionServiceImpl.java
+++ b/kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/RegionServiceImpl.java
@@ -2,6 +2,8 @@ package com.xiaojukeji.kafka.manager.service.service.impl;
 
 import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
 import com.xiaojukeji.kafka.manager.common.entity.pojo.RegionDO;
+import com.xiaojukeji.kafka.manager.common.events.RegionCreatedEvent;
+import com.xiaojukeji.kafka.manager.common.utils.SpringTool;
 import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
 import com.xiaojukeji.kafka.manager.common.zookeeper.znode.brokers.TopicMetadata;
 import com.xiaojukeji.kafka.manager.dao.RegionDao;
@@ -59,6 +61,8 @@ public class RegionServiceImpl implements RegionService {
                 return ResultStatus.BROKER_NOT_EXIST;
             }
             if (regionDao.insert(regionDO) > 0) {
+                // 发布region创建事件
+                SpringTool.publish(new RegionCreatedEvent(this, regionDO));
                 return ResultStatus.SUCCESS;
             }
         } catch (DuplicateKeyException e) {
diff --git a/kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/biz/RegionCreatedListener.java b/kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/biz/RegionCreatedListener.java
new file mode 100644
index 00000000..5daa0e9e
--- /dev/null
+++ b/kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/biz/RegionCreatedListener.java
@@ -0,0 +1,38 @@
+package com.xiaojukeji.kafka.manager.task.listener.biz;
+
+import com.xiaojukeji.kafka.manager.common.events.RegionCreatedEvent;
+import com.xiaojukeji.kafka.manager.task.dispatch.biz.CalRegionCapacity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationListener;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+
+/**
+ * Region创建监听器,
+ * TODO 后续需要将其移动到core模块
+ * @author zengqiao
+ * @date 22/01/11
+ */
+@Component
+public class RegionCreatedListener implements ApplicationListener<RegionCreatedEvent> {
+    private static final Logger logger = LoggerFactory.getLogger(RegionCreatedListener.class);
+
+    @Autowired
+    private CalRegionCapacity calRegionCapacity;
+
+    @Async
+    @Override
+    public void onApplicationEvent(RegionCreatedEvent event) {
+        try {
+            logger.info("cal region capacity started when region created, regionDO:{}.", event.getRegionDO());
+
+            calRegionCapacity.processTask(event.getRegionDO());
+
+            logger.info("cal region capacity finished when region created, regionDO:{}.", event.getRegionDO());
+        } catch (Exception e) {
+            logger.error("cal region capacity failed when region created, regionDO:{}.", event.getRegionDO(), e);
+        }
+    }
+}