From 1e8edd88ef0da9500509308d7a8c1eae0d805d64 Mon Sep 17 00:00:00 2001
From: linrf <linrf@qq.com>
Date: Fri, 11 Aug 2023 14:08:39 +0800
Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=90=AF=E5=8A=A8=E7=AD=89?=
 =?UTF-8?q?=E5=BE=85=E9=87=87=E9=9B=86=E5=99=A8=E8=BF=9E=E6=8E=A5=E4=BA=8B?=
 =?UTF-8?q?=E4=BB=B6?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../server/TerminalToChannelCacheMap.java     |  1 +
 .../common/LoadPackageCommandLineRunner.java  |  4 ++
 .../server/common/netty/ChannelAttribute.java |  2 +-
 .../common/netty/client/NettyClient.java      | 13 ++++++
 .../netty/client/handler/ClientHandler.java   | 11 +----
 .../common/netty/event/StartUpEvent.java      | 27 ++++++++++++
 .../netty/event/StartUpEventListener.java     | 41 +++++++++++++++++++
 7 files changed, 88 insertions(+), 11 deletions(-)
 create mode 100644 collector-terminal-management/src/main/java/com/docus/server/common/netty/event/StartUpEvent.java
 create mode 100644 collector-terminal-management/src/main/java/com/docus/server/common/netty/event/StartUpEventListener.java

diff --git a/collector-scheduling-management/src/main/java/com/docus/server/common/netty/server/TerminalToChannelCacheMap.java b/collector-scheduling-management/src/main/java/com/docus/server/common/netty/server/TerminalToChannelCacheMap.java
index 039bd13..c60b522 100644
--- a/collector-scheduling-management/src/main/java/com/docus/server/common/netty/server/TerminalToChannelCacheMap.java
+++ b/collector-scheduling-management/src/main/java/com/docus/server/common/netty/server/TerminalToChannelCacheMap.java
@@ -99,6 +99,7 @@ public class TerminalToChannelCacheMap {
         IP_CHANNEL_CACHE_MAP.remove(key);
         IP_TERMINATOR_CACHE_MAP.remove(key);
         NettyTerminatorDTO nettyTerminatorDTO = new NettyTerminatorDTO();
+        nettyTerminatorDTO.setBusyState(BusyStateEnum.IDLE);
         nettyTerminatorDTO.setOnlineState(OnlineStateEnum.OFFLINE);
         iSchTerminatorService.saveOrUpdate(key, nettyTerminatorDTO);
     }
diff --git a/collector-terminal-management/src/main/java/com/docus/server/common/LoadPackageCommandLineRunner.java b/collector-terminal-management/src/main/java/com/docus/server/common/LoadPackageCommandLineRunner.java
index 63be06e..c5436b9 100644
--- a/collector-terminal-management/src/main/java/com/docus/server/common/LoadPackageCommandLineRunner.java
+++ b/collector-terminal-management/src/main/java/com/docus/server/common/LoadPackageCommandLineRunner.java
@@ -136,6 +136,10 @@ public class LoadPackageCommandLineRunner implements CommandLineRunner {
             updatePart(loadSchCollectorVOList, value);
         }
 
+        loadSchCollectorVOList.forEach(p -> {
+            p.setDeployPath(saveCollectorPackagePath + p.getCollectorId() + "\\collector");
+        });
+
         //根据服务器端口区分，每次更新了哪些采集器版本
         template.opsForValue().set(key, JSON.toJSON(loadSchCollectorVOList));
     }
diff --git a/collector-terminal-management/src/main/java/com/docus/server/common/netty/ChannelAttribute.java b/collector-terminal-management/src/main/java/com/docus/server/common/netty/ChannelAttribute.java
index 5d3f659..f06e8d0 100644
--- a/collector-terminal-management/src/main/java/com/docus/server/common/netty/ChannelAttribute.java
+++ b/collector-terminal-management/src/main/java/com/docus/server/common/netty/ChannelAttribute.java
@@ -4,7 +4,7 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
 
-public class ChannelAttribute<T> {
+public class  ChannelAttribute<T> {
 
     /**
      * 属性键名称
diff --git a/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/NettyClient.java b/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/NettyClient.java
index 43c9ab7..250d23d 100644
--- a/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/NettyClient.java
+++ b/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/NettyClient.java
@@ -2,6 +2,7 @@ package com.docus.server.common.netty.client;
 
 import com.docus.core.util.json.JSON;
 import com.docus.server.common.CommMsg;
+import com.docus.server.common.netty.event.StartUpEvent;
 import com.docus.server.common.netty.server.CollectorChannelCacheMap;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.ChannelFuture;
@@ -13,6 +14,7 @@ import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.ApplicationContext;
 import org.springframework.core.env.Environment;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
@@ -46,6 +48,8 @@ public class NettyClient {
     private StringRedisTemplate template;
     @Resource
     private Environment env;
+    @Resource
+    private ApplicationContext applicationContext;
 
     /**
      * 发送消息给服务端
@@ -59,6 +63,15 @@ public class NettyClient {
         }
     }
 
+    /**
+     * 发送消息给服务端
+     *
+     * @param message 消息内容
+     */
+    public void publishEvent(String message, CommMsg commMsg) {
+        applicationContext.publishEvent(new StartUpEvent(this, message, commMsg));
+    }
+
     @PostConstruct
     public void start() {
         final EventLoopGroup group = new NioEventLoopGroup();
diff --git a/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/handler/ClientHandler.java b/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/handler/ClientHandler.java
index e66a66a..1093a21 100644
--- a/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/handler/ClientHandler.java
+++ b/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/handler/ClientHandler.java
@@ -138,16 +138,7 @@ public class ClientHandler extends SimpleChannelInboundHandler<CommMsg<Serializa
                     //不存在进程，则启动指定采集器进程,等待启动完成下发任务
                     StartUpExeUtils.startUpExeOnly(saveCollectorPackagePath + collectorId + "\\collector\\" + processName);
 
-                    while (true) {
-                        Channel channel = collectorChannelCacheMap.get(collectorId);
-                        log.info("等待采集器={}启动中", collectorId);
-                        if (null != channel && channel.isOpen()) {
-                            log.info("等待采集器={}启动成功并注册到终端", collectorId);
-                            collectorChannelCacheMap.writeAndFlush(collectorId, commMsg);
-                            break;
-                        }
-                        Thread.sleep(5000);
-                    }
+                    nettyClient.publishEvent(collectorId, commMsg);
                 }
             }
         }
diff --git a/collector-terminal-management/src/main/java/com/docus/server/common/netty/event/StartUpEvent.java b/collector-terminal-management/src/main/java/com/docus/server/common/netty/event/StartUpEvent.java
new file mode 100644
index 0000000..77c43ea
--- /dev/null
+++ b/collector-terminal-management/src/main/java/com/docus/server/common/netty/event/StartUpEvent.java
@@ -0,0 +1,27 @@
+package com.docus.server.common.netty.event;
+
+import com.docus.server.common.CommMsg;
+import lombok.Getter;
+import org.springframework.context.ApplicationEvent;
+
+@Getter
+public class StartUpEvent extends ApplicationEvent {
+    /**
+     * 下载对应任务id ,接收处理
+     */
+    private final String message;
+
+    private final CommMsg commMsg;
+
+    /**
+     * @param source  触发该事件的对象
+     * @param message 该事件携带的信息
+     */
+    public StartUpEvent(Object source, String message, CommMsg commMsg) {
+        super(source);
+        this.message = message;
+        this.commMsg = commMsg;
+    }
+
+
+}
diff --git a/collector-terminal-management/src/main/java/com/docus/server/common/netty/event/StartUpEventListener.java b/collector-terminal-management/src/main/java/com/docus/server/common/netty/event/StartUpEventListener.java
new file mode 100644
index 0000000..5a508c9
--- /dev/null
+++ b/collector-terminal-management/src/main/java/com/docus/server/common/netty/event/StartUpEventListener.java
@@ -0,0 +1,41 @@
+package com.docus.server.common.netty.event;
+
+import com.docus.server.common.netty.server.CollectorChannelCacheMap;
+import io.netty.channel.Channel;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.event.EventListener;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+@Component
+@Slf4j
+public class StartUpEventListener {
+
+    @Resource
+    private CollectorChannelCacheMap collectorChannelCacheMap;
+
+
+    @EventListener
+    @Async("threadPoolExecutor")
+    public void startUp(StartUpEvent startUpEvent) {
+        while (true) {
+            String collectorId = startUpEvent.getMessage();
+            Channel channel = collectorChannelCacheMap.get(collectorId);
+            log.info("等待采集器={}启动中", collectorId);
+            if (null != channel && channel.isOpen()) {
+                log.info("等待采集器={}启动成功并注册到终端", collectorId);
+                collectorChannelCacheMap.writeAndFlush(collectorId, startUpEvent.getCommMsg());
+                break;
+            }
+            try {
+                Thread.sleep(5000);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+
+}