EventExecutor

gistfile1.java

import java.util.Queue;
import java.util.concurrent.PriorityBlockingQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class EventExecutor {
	private static Log log = LogFactory.getLog(EventExecutor.class);
	private final Queue<Event> queues;
	private Thread running;
	private long cycleTime = 100;

	public EventExecutor() {
		super();
		queues = new PriorityBlockingQueue<Event>();
	}

	public EventExecutor(long cycleTime) {
		this();
		this.cycleTime = cycleTime;
	}

	public Event doAsyncEvent(Runnable runnable) {
		return doAsyncEvent(runnable, 0L);
	}

	public Event doAsyncEvent(Runnable runnable, long delay) {
		return addEvent(runnable, delay + System.currentTimeMillis());
	}

	private Event addEvent(Runnable runnable, long time) {
		Event action = new Event(runnable, time);
		queues.add(action);
		return action;
	}

	public long getNextEventTime() {
		Event action = queues.peek();
		return action == null ? Long.MAX_VALUE : action.getTime();
	}

	public synchronized boolean start() {
		if (running != null) {
			return true;
		}
		try {
			running = new Thread() {
				@Override
				public void run() {
					while (running == this) {
						try {
							doEvent();
						} catch (Throwable e) {
							if (e instanceof Error) {
								log.fatal("fatal error occurs.");
								log.fatal(e.getMessage(), e);
								throw (Error)e;
							} else {
								log.warn(e.getMessage(), e);
							}
						}
					}
				}
			};
			running.start();
			return true;

		} catch (Exception e) {
			log.error(e.getMessage(), e);
			return false;
		}
	}

	public synchronized boolean stop() {
		if (running == null) {
			return false;
		}
		running.interrupt();
		running = null;
		return true;
	}

	private void doEvent() throws Throwable {
		while (getNextEventTime() <= System.currentTimeMillis()) {
			queues.poll().run();
		}

		if (cycleTime > 0L) {
			Thread.sleep(cycleTime);
		}
	}

	public int getQueueSize() {
		return queues.size();
	}

}