package id.iptek.utms.agent.queue;

import id.iptek.utms.agent.db.DatabaseException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author jakar
 * @param <T>
 */
public class DelayMessageQueue<T> {
    
    private final Logger logger = LoggerFactory.getLogger(getClass());
    
    private ExecutorService executorservice;
    private int capacity = 10;
    private ConsumerMode mode = ConsumerMode.SINGLE;
    private QueueMessageHandler<T> handler;
    private final ConcurrentLinkedQueue<T> queue = new ConcurrentLinkedQueue<>();
    private boolean running = false;
    private int totalThread = 1;
    private ExecutorService internalExec = null;
    private Future internalWorker = null;
    private long batchFlushIntervalMillis = 1000L;

    public DelayMessageQueue() {
    }

    public DelayMessageQueue(ExecutorService executorservice, QueueMessageHandler<T> handler) {
        this(executorservice, handler, 10);
    }

    public DelayMessageQueue(ExecutorService executorservice, QueueMessageHandler<T> handler, int capacity) {
        this.executorservice = executorservice;
        this.handler = handler;
        this.capacity = capacity;
    }

    public void setExecutorservice(ExecutorService executorservice) {
        this.executorservice = executorservice;
    }

    public void setCapacity(int capacity) {
        this.capacity = capacity;
    }

    public int getCapacity() {
        return capacity;
    }

    public void setMode(ConsumerMode mode) {
        this.mode = mode;
    }

    public void setHandler(QueueMessageHandler<T> handler) {
        this.handler = handler;
    }

    public void setTotalThread(int totalThread) {
        this.totalThread = totalThread;
    }

    public void setBatchFlushIntervalMillis(long batchFlushIntervalMillis) {
        this.batchFlushIntervalMillis = batchFlushIntervalMillis;
    }
    
    public void run() {
        logger.info("Run queue worker ...");
        this.running = true;
        
        internalExec = Executors.newFixedThreadPool(totalThread);
        internalWorker = this.executorservice.submit(new QueueRunnable());
    }
    
    public void stop(boolean interrupt) {
        logger.info("Stopping queue worker ...");
        stop(interrupt, 0L, TimeUnit.MILLISECONDS);
    }

    public boolean stop(boolean interrupt, long timeout, TimeUnit unit) {
        logger.info("Stopping queue worker ...");
        this.running = false;
        if(internalWorker == null) {
            logger.warn("Queue worker has not been run!");
            return true;
        }
        try {
            if(interrupt) {
                internalWorker.cancel(true);
            } else if(!internalWorker.isDone()) {
                if(timeout > 0L) {
                    internalWorker.get(timeout, unit);
                } else {
                    internalWorker.get();
                }
            }
        } catch(Exception ex) {
            logger.warn("Queue worker did not stop gracefully: {}", ex.getMessage());
        }

        internalExec.shutdown();
        try {
            boolean terminated = timeout > 0L ? internalExec.awaitTermination(timeout, unit) : internalExec.awaitTermination(0L, TimeUnit.MILLISECONDS);
            if(!terminated) {
                logger.warn("Queue handler tasks did not finish within {} {}", timeout, unit);
                internalExec.shutdownNow();
                return false;
            }
        } catch(InterruptedException ex) {
            logger.warn("Interrupted while waiting queue handler tasks: {}", ex.getMessage());
            internalExec.shutdownNow();
            Thread.currentThread().interrupt();
            return false;
        }
        return true;
    }
    
    public boolean add(T message) {
        return queue.add(message);
    }
    
    class QueueRunnable implements Runnable {

        private final List<T> internalList = new ArrayList<>();
        private long lastFlushTime = System.currentTimeMillis();
        
        @Override
        public void run() {
            while(running || !queue.isEmpty()) {
                try {
                    T message = queue.poll();
                    if(message != null) {
                        if(mode == ConsumerMode.BATCH) {
                            // handled batches
                            internalList.add(message);

                            // check capacity
                            if(internalList.size() >= capacity) {
                                flushBatch();
                            }
                        } else if(mode == ConsumerMode.SINGLE) {
                            // run job using separate runnable
                            Runnable run = new Runnable() {
                                @Override
                                public void run() {
                                    try {
                                        // handle single
                                        boolean handled = handler.handleMessage(message);
                                        logger.debug("Message handled? {}", handled);
                                    } catch(DatabaseException ex) {
                                        logger.error("Error saving heartbeats: {}", ex.getMessage(), ex);
                                    }
                                }
                            };
                            internalExec.execute(run);
                        }
                    } else {
                        if(mode == ConsumerMode.BATCH && !internalList.isEmpty()
                                && System.currentTimeMillis() - lastFlushTime >= batchFlushIntervalMillis) {
                            flushBatch();
                        }
                        TimeUnit.MILLISECONDS.sleep(100L);
                    }
                    if(mode == ConsumerMode.BATCH && !internalList.isEmpty()
                            && System.currentTimeMillis() - lastFlushTime >= batchFlushIntervalMillis) {
                        flushBatch();
                    }
                } catch(InterruptedException ex) {
                    logger.debug("Queue worker interrupted: {}", ex.getMessage());
                    Thread.currentThread().interrupt();
                    break;
                } catch(Exception ex) {
                    logger.error("Error consuming queue : {}", ex.getMessage(), ex);
                }
            }
            if(mode == ConsumerMode.BATCH && !internalList.isEmpty()) {
                flushBatch();
            }
        }

        private void flushBatch() {
            List<T> workingList = new ArrayList<>();
            workingList.addAll(internalList);
            internalList.clear();
            lastFlushTime = System.currentTimeMillis();
            
            // run job using separate runnable
            Runnable run = new Runnable() {
                @Override
                public void run() {
                    try {
                        boolean handled = handler.handleMessages(workingList);
                        logger.debug("Batch messages handled? {}", handled);
                    } catch(DatabaseException ex) {
                        logger.error("Error saving heartbeats: {}", ex.getMessage(), ex);
                    }
                }
            };
            internalExec.execute(run);
        }
        
    }
}
