bugfix collector can not auto reconnect when channel idle

collector/src/main/java/org/dromara/hertzbeat/collector/dispatch/entrance/CollectServer.java

@@ -132,12 +132,16 @@ public void onChannelActive(Channel channel) {
                 scheduledExecutor = Executors.newSingleThreadScheduledExecutor(threadFactory);
                 // schedule send heartbeat message
                 scheduledExecutor.scheduleAtFixedRate(() -> {
-                    ClusterMsg.Message heartbeat = ClusterMsg.Message.newBuilder()
-                            .setIdentity(identity)
-                            .setType(ClusterMsg.MessageType.HEARTBEAT)
-                            .build();
-                    CollectServer.this.sendMsg(heartbeat);
-                    log.info("collector send cluster server heartbeat, time: {}.", System.currentTimeMillis());
+                    try {
+                        ClusterMsg.Message heartbeat = ClusterMsg.Message.newBuilder()
+                                .setIdentity(identity)
+                                .setType(ClusterMsg.MessageType.HEARTBEAT)
+                                .build();
+                        CollectServer.this.sendMsg(heartbeat);
+                        log.info("collector send cluster server heartbeat, time: {}.", System.currentTimeMillis());   
+                    } catch (Exception e) {
+                        log.error(e.getMessage());
+                    }
                 }, 5, 5, TimeUnit.SECONDS);
             }
         }

manager/src/main/java/org/dromara/hertzbeat/manager/scheduler/netty/ManageServer.java

@@ -66,7 +66,7 @@ private void init(final SchedulerProperties schedulerProperties, final CommonThr
         this.remotingServer.registerHook(Lists.newArrayList(new NettyHook() {
             @Override
             public void doBeforeRequest(ChannelHandlerContext ctx, ClusterMsg.Message message) {
-                ManageServer.this.clientChannelTable.put(message.getIdentity(), ctx.channel());
+                // do something before processor list
             }
         }));
 
@@ -84,13 +84,17 @@ public void start() {
         this.remotingServer.start();
 
         this.channelSchedule.scheduleAtFixedRate(() -> {
-            ManageServer.this.clientChannelTable.forEach((collector, channel) -> {
-                if (!channel.isActive()) {
-                    channel.closeFuture();
-                    ManageServer.this.clientChannelTable.remove(collector);
-                    ManageServer.this.collectorAndJobScheduler.collectorGoOffline(collector);
-                }
-            });
+            try {
+                this.clientChannelTable.forEach((collector, channel) -> {
+                    if (!channel.isActive()) {
+                        channel.closeFuture();
+                        this.clientChannelTable.remove(collector);
+                        this.collectorAndJobScheduler.collectorGoOffline(collector);
+                    }
+                });   
+            } catch (Exception e) {
+                log.error(e.getMessage(), e);
+            }
         }, 10, 3, TimeUnit.SECONDS);
     }
 
@@ -113,6 +117,14 @@ public Channel getChannel(final String identity) {
         return channel;
     }
 
+    public void addChannel(final String identity, Channel channel) {
+        Channel preChannel = this.clientChannelTable.get(identity);
+        if (preChannel != null && channel.isActive()) {
+            preChannel.close();
+        }
+        this.clientChannelTable.put(identity, channel);
+    }
+
     public void closeChannel(final String identity) {
         Channel channel = this.getChannel(identity);
         if (channel != null) {
@@ -165,6 +177,7 @@ public void onChannelIdle(Channel channel) {
             if (identity != null) {
                 ManageServer.this.clientChannelTable.remove(identity);
                 ManageServer.this.collectorAndJobScheduler.collectorGoOffline(identity);
+                channel.close();
                 log.info("handle idle event triggered. the client {} is going offline.", identity);
             }
         }

manager/src/main/java/org/dromara/hertzbeat/manager/scheduler/netty/process/CollectorOnlineProcessor.java

@@ -24,6 +24,7 @@ public ClusterMsg.Message handle(ChannelHandlerContext ctx, ClusterMsg.Message m
         String collector = message.getIdentity();
         log.info("the collector {} actively requests to go online.", collector);
         CollectorInfo collectorInfo = JsonUtil.fromJson(message.getMsg(), CollectorInfo.class);
+        this.manageServer.addChannel(collector, ctx.channel());
         this.manageServer.getCollectorAndJobScheduler().collectorGoOnline(collector, collectorInfo);
         return null;
     }

manager/src/main/java/org/dromara/hertzbeat/manager/scheduler/netty/process/HeartbeatProcessor.java

@@ -23,11 +23,10 @@ public ClusterMsg.Message handle(ChannelHandlerContext ctx, ClusterMsg.Message m
         String identity = message.getIdentity();
         boolean isChannelExist = this.manageServer.isChannelExist(identity);
         if (!isChannelExist) {
-            log.info("the collector {} has reconnected and to go online.", identity);
-            this.manageServer.getCollectorAndJobScheduler().collectorGoOnline(identity, null);
+            log.info("the collector {} is not online.", identity);
         }
         if (log.isDebugEnabled()) {
-            log.debug("server receive collector heartbeat");
+            log.debug("server receive collector {} heartbeat", message.getIdentity());
         }
         return ClusterMsg.Message.newBuilder()
                 .setType(ClusterMsg.MessageType.HEARTBEAT)

remoting/src/main/java/org/dromara/hertzbeat/remoting/netty/NettyRemotingServer.java

@@ -138,7 +138,7 @@ private void initChannel(final SocketChannel channel) {
         pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
         pipeline.addLast(new ProtobufEncoder());
         // idle state
-        pipeline.addLast(new IdleStateHandler(0, 0, 30));
+        pipeline.addLast(new IdleStateHandler(0, 0, 100));
         pipeline.addLast(new NettyServerHandler());
     }