class CollectNettyEventListener implements NettyEventListener {

        @Override
        public void onChannelActive(Channel channel) {<FILL_FUNCTION_BODY>}

        @Override
        public void onChannelIdle(Channel channel) {
            log.info("handle idle event triggered. collector is going offline.");
        }
    }

            String identity = CollectServer.this.collectJobService.getCollectorIdentity();
            String mode = CollectServer.this.collectJobService.getCollectorMode();
            CollectorInfo collectorInfo = CollectorInfo.builder()
                    .name(identity)
                    .ip(IpDomainUtil.getLocalhostIp())
                    .mode(mode)
                    // todo more info
                    .build();
            timerDispatch.goOnline();
            // send online message
            ClusterMsg.Message message = ClusterMsg.Message.newBuilder()
                    .setIdentity(identity)
                    .setType(ClusterMsg.MessageType.GO_ONLINE)
                    .setMsg(JsonUtil.toJson(collectorInfo))
                    .build();
            CollectServer.this.sendMsg(message);

            if (scheduledExecutor == null) {
                ThreadFactory threadFactory = new ThreadFactoryBuilder()
                        .setUncaughtExceptionHandler((thread, throwable) -> {
                            log.error("HeartBeat Scheduler has uncaughtException.");
                            log.error(throwable.getMessage(), throwable);
                        })
                        .setDaemon(true)
                        .setNameFormat("heartbeat-worker-%d")
                        .build();
                scheduledExecutor = Executors.newSingleThreadScheduledExecutor(threadFactory);
                // schedule send heartbeat message
                scheduledExecutor.scheduleAtFixedRate(() -> {
                    try {
                        ClusterMsg.Message heartbeat = ClusterMsg.Message.newBuilder()
                                .setIdentity(identity)
                                .setDirection(ClusterMsg.Direction.REQUEST)
                                .setType(ClusterMsg.MessageType.HEARTBEAT)
                                .build();
                        CollectServer.this.sendMsg(heartbeat);
                        log.info("collector send cluster server heartbeat, time: {}.", System.currentTimeMillis());   
                    } catch (Exception e) {
                        log.error("schedule send heartbeat to server error.{}", e.getMessage());
                    }
                }, 5, 5, TimeUnit.SECONDS);
            }
        
<no_super_class>