package de.renew.util;

import java.util.Enumeration;
import java.util.Iterator;
import java.util.TreeMap;
import java.util.Vector;


/**
 * The Scheduler class allows to delay the execution of runnables.
 * Every runnable is executed in the scheduler thread.
 * <p>
 * The scheduler thread is a non-daemon thread and will not
 * terminate as long as any runnable is scheduled. If you want
 * the Java VM to shut down after all application threads have
 * finished their execution, you have to take care that all
 * scheduled runnables get {@link #cancel cancelled}. Otherwise,
 * the Scheduler thread will prevent the Java VM from termination
 * until all scheduled runnables have been executed.
 * </p>
 **/
public class Scheduler implements Runnable {
    private final static org.apache.log4j.Logger _logger =
        org.apache.log4j.Logger.getLogger(Scheduler.class);

    /**
     * The singleton {@link Scheduler} instance.
     */
    private static Scheduler _instance;

    /**
     * The thread used for execution of the scheduler and all
     * scheduled runnables. The thread is created on demand
     * (i.e. when the first runnable gets registered) and stopped
     * when no runnables are left. When stopped, this variable is
     * set back to <code>null</code>.
     **/
    private Thread _schedulerThread = null;

    /**
     * A map from time stamps represented as longs
     * to pairs of time stamp and queued elements. This map is always
     * accessed by take and insert methods.
     */
    private final TreeMap<Long, SchedulerPair> _queues = new TreeMap<>();

    private Scheduler() {
        // nothing to do.
        // The scheduler thread gets started whenever the first
        // runnable is enqueued.
        if (_logger.isDebugEnabled()) {
            _logger.debug(this + ": created instance.");
        }
    }

    /**
     * Factory method to get a Scheduler instance for use.
     * @return the Scheduler instance.  
     */
    public static synchronized Scheduler instance() {
        if (_instance == null) {
            _instance = new Scheduler();
        }
        return _instance;
    }

    private void ensureRunningThread() {
        if (_schedulerThread == null) {
            _schedulerThread = new Thread(this, "Renew Scheduler");
            _schedulerThread.start();
            if (_logger.isDebugEnabled()) {
                _logger.debug(this + ": created thread.");
            }
        }
    }

    private SchedulerPair takeEarliest() {
        if (_queues.isEmpty()) {
            return null;
        } else {
            Long key = _queues.firstKey();
            return _queues.remove(key);
        }
    }

    private SchedulerPair takePairAt(long timeMillis) {
        Long key = timeMillis;
        SchedulerPair result = _queues.remove(key);
        return result == null ? new SchedulerPair(key) : result;
    }

    /**
     * Reinsert the given scheduler pair if it contains
     * at least one runnable.
     *
     * @param pair the <code>SchedulerPair</code> to be inserted
     */
    private void putPair(SchedulerPair pair) {
        if (!pair.list.isEmpty()) {
            _queues.put(pair.key, pair);
        }
    }

    /**
     * Delays the execution of a runnable for a given period of time
     * A runnable that is executed may call this method to reinsert itself
     * or to schedule other runnables.
     *
     * @param runnable the <code>Runnable</code> to be executed
     * @param deltaMillis the number of milliseconds to delay the execution.
     */
    public synchronized void executeIn(Runnable runnable, long deltaMillis) {
        executeAt(runnable, System.currentTimeMillis() + deltaMillis);
    }

    /**
     * Delays the execution of a runnable until a given point of time
     * A runnable that is executed may call this method to reinsert itself
     * or to schedule other runnables.
     *
     * @param runnable the <code>Runnable</code> to be executed
     * @param timeMillis the time in milliseconds of desired the execution.
     */
    public synchronized void executeAt(Runnable runnable, long timeMillis) {
        SchedulerPair pair = takePairAt(timeMillis);
        pair.list.addFirst(runnable);
        putPair(pair);
        if (_logger.isTraceEnabled()) {
            _logger.trace(this + ": scheduled " + runnable + " for " + timeMillis);
        }


        // Make sure that the scheduler thread wakes up to process
        // the new runnnable, if it happens to be the earliest runnable.
        ensureRunningThread();
        notifyAll();
    }

    /**
     * Cancels any scheduled excecution for the given
     * Runnable. The cancellation will not affect a current
     * execution of the Runnable. This method has no effect if
     * the Runnable was not scheduled.
     *
     * @param runnable the <code>Runnable</code> to be removed
     *                 from the schedule
     */
    public synchronized void cancel(Runnable runnable) {
        if (_logger.isTraceEnabled()) {
            _logger.trace(this + ": request to cancel " + runnable);
        }
        Iterator<SchedulerPair> allElements = _queues.values().iterator();
        Vector<Long> toRemove = new Vector<>(_queues.size());
        while (allElements.hasNext()) {
            SchedulerPair pair = allElements.next();
            pair.list.remove(runnable);
            if (pair.list.isEmpty()) {
                toRemove.add(pair.key);
            }
        }
        Enumeration<Long> removeElements = toRemove.elements();
        while (removeElements.hasMoreElements()) {
            _queues.remove(removeElements.nextElement());
        }


        // Make sure that the scheduler thread wakes up to check
        // whether any runnables are left in the queue.
        notifyAll();
    }

    @Override
    public synchronized void run() {
        SchedulerPair pair;
        do {
            pair = takeEarliest();
            if (pair != null) {
                long timeMillis = pair.key;
                long now = System.currentTimeMillis();
                if (timeMillis > now) {
                    // No runnables are ready yet.
                    putPair(pair);
                    try {
                        wait(timeMillis - now);
                    } catch (InterruptedException e) {
                        // This is expected.
                    }
                } else {
                    // Some runnables have become executable.
                    // After this point of time, we must not assume
                    // that the map is unchanged, because the
                    // invoked runnables might have inserted further
                    // items into the queue.
                    while (!pair.list.isEmpty()) {
                        Runnable runnable = pair.list.removeLast();
                        if (_logger.isTraceEnabled()) {
                            _logger.trace(this + ": executing at " + now + ": " + runnable);
                        }
                        runnable.run();
                    }

                    // No wait required. Check next time stamp immediately.
                }
            }
        } while (pair != null);

        if (_logger.isDebugEnabled()) {
            _logger.debug(this + ": quitting thread.");
        }

        // No runnables in the queue
        // Instead of waiting, stop the thread.
        _schedulerThread = null;
    }
}