/*
 * Copyright (c) 2018. Dark Phoenixs (Open-Source Organization).
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.darkphoenixs.kafka.pool;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * The type Kafka message receiver monitor.
 *
 * @param <T> the type parameter
 */
public class KafkaMessageReceiverMonitor<T> {

    private static final Logger logger = LoggerFactory.getLogger(KafkaMessageReceiverMonitor.class);

    private final int monitorPoolSize = 1;

    private final long monitorIntervalTime;

    private final int monitorPercentage;

    /**
     * The Monitor pool.
     */
    protected final ExecutorService monitorPool;

    /**
     * The Monitor threads.
     */
    protected final List<MonitorThread> monitorThreads;

    /**
     * Instantiates a new Kafka message receiver monitor.
     *
     * @param topic               the topic
     * @param monitorIntervalTime the monitor interval time
     * @param monitorPercentage   the monitor percentage
     * @param blockingQueue       the blocking queue
     */
    public KafkaMessageReceiverMonitor(String topic, long monitorIntervalTime, int monitorPercentage, BlockingQueue<T> blockingQueue) {

        this.monitorIntervalTime = monitorIntervalTime;

        this.monitorPercentage = monitorPercentage;

        this.monitorThreads = new ArrayList<MonitorThread>(monitorPoolSize);

        this.monitorPool = Executors.newFixedThreadPool(monitorPoolSize, new KafkaPoolThreadFactory(MonitorThread.tagger + "-" + topic));

        MonitorThread monitorThread = new MonitorThread(blockingQueue);

        this.monitorThreads.add(monitorThread);

        this.monitorPool.submit(monitorThread);
    }

    /**
     * Destroy.
     */
    public void destroy() {

        for (MonitorThread thread : monitorThreads)

            thread.shutdown();

        if (monitorPool != null) {

            monitorPool.shutdown();

            logger.info("Monitor pool closed.");
        }
    }

    /**
     * The type Monitor thread.
     */
    class MonitorThread implements Runnable {

        /**
         * The constant tagger.
         */
        public static final String tagger = "MonitorThread";

        private final AtomicBoolean closed = new AtomicBoolean(false);

        private final BlockingQueue<T> blockingQueue;

        /**
         * Instantiates a new Monitor thread.
         *
         * @param blockingQueue the blocking queue
         */
        public MonitorThread(BlockingQueue<T> blockingQueue) {

            this.blockingQueue = blockingQueue;
        }

        @Override
        public void run() {

            logger.info(Thread.currentThread().getName() + " start.");

            while (!closed.get()) {

                // queue usage size
                int usageSize = blockingQueue.size();
                // queue free size
                int freeSize = blockingQueue.remainingCapacity();
                // queue usage rate (%)
                double usageRate = (usageSize * 1.0) / ((usageSize + freeSize) * 1.0) * 100;
                // usageRate > monitorPercentage
                if (usageRate > monitorPercentage)

                    logger.warn("BlockingQueue usage rate: {}%. usage size: {}. free size: {}.", usageRate, usageSize, freeSize);

                try {
                    TimeUnit.MILLISECONDS.sleep(monitorIntervalTime);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            logger.info(Thread.currentThread().getName() + " end.");
        }

        /**
         * Shutdown.
         */
        public void shutdown() {

            closed.set(true);
        }
    }

}

@@ -725,59 +725,8 @@
 y;%0A%0A
-import java.util.ArrayList;%0Aimport java.util.List;%0A
 impo
@@ -800,23 +800,17 @@
 Executor
-Service
+s
 ;%0Aimport
@@ -835,55 +835,32 @@
 ent.
-Executors;%0Aimport java.util.concurrent.TimeUnit
+ScheduledExecutorService
 ;%0Aim
@@ -889,28 +889,16 @@
 ent.
-atomic.AtomicBoolean
+TimeUnit
 ;%0A%0A/
@@ -1181,53 +1181,8 @@
 1;%0A%0A
-    private final long monitorIntervalTime;%0A%0A
     
@@ -1280,16 +1280,25 @@
 d final 
+Scheduled
 Executor
@@ -1323,109 +1323,8 @@
 l;%0A%0A
-    /**%0A     * The Monitor threads.%0A     */%0A    protected final List%3CMonitorThread%3E monitorThreads;%0A%0A
     
@@ -1775,173 +1775,38 @@
 itor
-IntervalTime = monitorIntervalTime;%0A%0A        this.monitorPercentage = monitorPercentage;%0A%0A        this.monitorThreads = new ArrayList%3CMonitorThread%3E(monitorPoolSize)
+Percentage = monitorPercentage
 ;%0A%0A 
@@ -1848,11 +1848,15 @@
 .new
-Fix
+Schedul
 edTh
@@ -1959,321 +1959,193 @@
     
-MonitorThread monitorThread = new MonitorThread(blockingQueue);%0A%0A        this.monitorThreads.add(monitorThread);%0A%0A        this.monitorPool.submit(monitorThread);%0A    %7D%0A%0A    /**%0A     * Destroy.%0A     */%0A    public void destroy() %7B%0A%0A        for (MonitorThread thread : monitorThreads)%0A%0A            thread.shutdown();
+this.monitorPool.scheduleAtFixedRate(new MonitorThread(blockingQueue), 0, monitorIntervalTime, TimeUnit.MILLISECONDS);%0A    %7D%0A%0A    /**%0A     * Destroy.%0A     */%0A    public void destroy() %7B
 %0A%0A  
@@ -2498,80 +2498,8 @@
 %22;%0A%0A
-        private final AtomicBoolean closed = new AtomicBoolean(false);%0A%0A
     
@@ -2855,121 +2855,8 @@
  %7B%0A%0A
-            logger.info(Thread.currentThread().getName() + %22 start.%22);%0A%0A            while (!closed.get()) %7B%0A%0A    
     
@@ -2879,28 +2879,24 @@
  usage size%0A
-    
             
@@ -2929,28 +2929,25 @@
 eue.size();%0A
-    
+%0A
             
@@ -2973,28 +2973,24 @@
             
-    
 int freeSize
@@ -3023,28 +3023,25 @@
 Capacity();%0A
-    
+%0A
             
@@ -3076,20 +3076,16 @@
         
-    
 double u
@@ -3153,19 +3153,17 @@
  * 100;%0A
-   
+%0A
         
@@ -3166,17 +3166,16 @@
         
- 
 // usage
@@ -3199,20 +3199,16 @@
 centage%0A
-    
         
@@ -3247,20 +3247,16 @@
 ntage)%0A%0A
-    
         
@@ -3377,415 +3377,8 @@
 e);%0A
-%0A                try %7B%0A                    TimeUnit.MILLISECONDS.sleep(monitorIntervalTime);%0A                %7D catch (InterruptedException e) %7B%0A                    e.printStackTrace();%0A                %7D%0A            %7D%0A%0A            logger.info(Thread.currentThread().getName() + %22 end.%22);%0A        %7D%0A%0A        /**%0A         * Shutdown.%0A         */%0A        public void shutdown() %7B%0A%0A            closed.set(true);%0A
     
@@ -3374,26 +3374,27 @@
 Size);%0A        %7D
+%0A
 %0A    %7D%0A%0A%7D%0A

use ScheduledExecutorService fixed rate.