package system;

import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * a generic scheduler that may be extended to provide fixed-delay execution to tasks
 * @author Willie Scholtz
 * @param <T> the type of bean to be scheduled
 */
public abstract class Scheduler<T extends Callable<T>> {

    private static final Log LOG = LogFactory.getLog(Scheduler.class);
    private static final int MAX_THREADS = Runtime.getRuntime().availableProcessors();

    private Timer timer = null;
    private final int delay;

    /**
     * creates a new scheduler
     * @param delay the delay in seconds between executions
     */
    public Scheduler(final int delay) {
        this.delay = delay;
    }

    /**
     * creates a new timer for executing tasks
     * @param seconds number of seconds between each execution
     * @return a Timer
     */
    private Timer getSchedulerTimer(int seconds) {
        final String cName = this.getClass().getSimpleName();

        final Timer sTimer = new Timer(cName + " Scheduler", false);
        final TimerTask sTask = new TimerTask() {
            @Override
            public void run() {
                LOG.debug("before running " + cName + " timer");
                Scheduler.this.runScheduler();
                LOG.debug("after running " + cName + " timer");
            }
        };

        sTimer.schedule(sTask, 0, (1000 * seconds));
        return sTimer;
    }

    /**
     * starts running this Scheduler, if the scheduler is currently executing,
     * tasks will be canceled and a new timer will be scheduled.
     */
    public void start() {
        this.stop();

        LOG.info("starting scheduler[" + getClass().getSimpleName() + "]...");
        this.timer = getSchedulerTimer(this.delay);
    }

    /**
     * stops the execution of this Scheduler.
     */
    public void stop() {
        LOG.info("stopping scheduler[" + this.getClass().getSimpleName() + "]...");
        if (this.timer != null) {
            this.timer.cancel();
        }
    }

    /**
     * retrieves a list of tasks to execute
     * @param currentDate the current date of the scheduler
     * @return a non-null List of tasks
     */
    public abstract List<T> getTasksForExecution(final Date currentDate);

    /**
     * runs the scheduler according to the specified delay
     */
    private void runScheduler() {
        final Set<Future<T>> futures = new HashSet<Future<T>>();
        final ExecutorService pool = Executors.newFixedThreadPool(MAX_THREADS);
        final Date now  = new Date();

        try {
            final List<T> tasks = this.getTasksForExecution(now);
            if (!tasks.isEmpty()) {
                LOG.info("executing " + tasks.size() + " task" + (tasks.size() != 1 ? "s" : ""));

                // submit messages for execution
                for (final T task : tasks) {
                    futures.add(pool.submit(task));
                }

                // wait for completion
                for (final Future<T> future : futures) {
                    try {
                        // max wait time for 1 minute
                        final T sendTaskOp = future.get(1L, TimeUnit.MINUTES);
                        LOG.info("task[" + sendTaskOp + "] executed...");
                    } catch (InterruptedException ex) {
                        LOG.error("interupted while executing task - " + ex.getMessage(), ex);
                    } catch (ExecutionException ex) {
                        LOG.error("error while executiong task  - " + ex.getMessage(), ex);
                    } catch (TimeoutException ex) {
                        LOG.error("executing the task timed out! - " + ex.getMessage(), ex);
                    }
                }
            }
        } finally {
            pool.shutdown();
        }
    }
}