package com.hammurapi.common.concurrent;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * This class does not queue jobs - it either executes them immediately in a background thread
 * or in a caller thread if all background threads are busy.
 * @author Pavel Vlasov
 *
 */
public class NonQueueingThreadPoolExecutorService extends AbstractExecutorService {
	
	private AtomicBoolean shutdown = new AtomicBoolean(); 
	
	private class NonQueueingRunnable implements Runnable {
		
		private int idx;

		public NonQueueingRunnable(int idx) {
			this.idx = idx;
		}
		
		Runnable[] toRun = {null};

		@Override
		public void run() {
			try {
				while (true) {
					synchronized (toRun) {
						while (toRun[0]==null) {
							if (shutdown.get()) {
								toRun.notifyAll();
								return;
							}
							try {
								toRun.wait();
							} catch (InterruptedException e) {
								handleInterruptedException(e);
							}
						}
					}
					try {
						toRun[0].run();
					} catch (Exception e) {
						handleException(toRun[0], e);
					}
					synchronized (toRun) {
						toRun[0] = null;
					}
				}
			} finally {
				runnables[idx] = null;
			}
		}

		boolean offer(Runnable command) {
			synchronized (toRun) {
				if (this.toRun[0]==null) {
					this.toRun[0]=command;
					this.toRun.notifyAll();
					return true;
				}
				return false;
			}
		}
		
		void shutdown() {
			synchronized (toRun) {
				toRun.notifyAll();
			}
		}

		public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
			synchronized (toRun) {
				if (toRun[0]!=null) {					
					if (timeout==0) {
						toRun.wait();
					} else {
						toRun.wait(unit.toMillis(timeout));
					}
					if (toRun[0]!=null) {
						return false;
					}
				}
			}
			return true;
		}
	}
	
	private NonQueueingRunnable[] runnables;
	
	public NonQueueingThreadPoolExecutorService(ThreadFactory threadFactory, int size) {
		runnables = new NonQueueingRunnable[size];
		for (int i=0; i<runnables.length; ++i) {
			runnables[i] = new NonQueueingRunnable(i);
			Thread thread = threadFactory.newThread(runnables[i]);
			if (thread == null) {
				runnables[i] = null;
			} else {
				thread.start();
			}
		}
	}

	@Override
	public void shutdown() {		
		shutdown.set(true);
		for (NonQueueingRunnable nqr: runnables) {
			if (nqr!=null) {
				nqr.shutdown();
			}
		}
	}

	@Override
	public List<Runnable> shutdownNow() {
		shutdown();
		return Collections.emptyList();
	}

	@Override
	public boolean isShutdown() {
		return shutdown.get();
	}

	@Override
	public boolean isTerminated() {
		for (NonQueueingRunnable nqr: runnables) {
			synchronized (nqr.toRun) {
				if (nqr.toRun[0]!=null) {
					return false;
				}
			}
		}
		return false;
	}

	@Override
	public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
		for (NonQueueingRunnable nqr: runnables) {
			if (nqr!=null && !nqr.awaitTermination(timeout, unit)) {
				return false;
			}
		}
		return true;
	}

	@Override
	public void execute(Runnable command) {
		for (NonQueueingRunnable nqr: runnables) {
			if (nqr.offer(command)) {
				return;
			}
		}
		onRejectedExecution(command);
	}
	
	protected void handleException(Runnable task, Exception e) {
		LOGGER.log(Level.SEVERE, "Exception in task "+task+": "+e, e);
	}

	/**
	 * This method is invoked when all threads are busy and cannot accept command for
	 * immediate execution. This implementation executes the command in the caller thread.
	 * @param command
	 */
	protected void onRejectedExecution(Runnable command) {
		command.run();
	}
	
	private static final Logger LOGGER = Logger.getLogger(NonQueueingThreadPoolExecutorService.class.getName());

	protected void handleInterruptedException(InterruptedException e) {
		LOGGER.log(Level.SEVERE, "Interrupted exception: "+e, e);
	}
	
    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
        return new NonBlockingFutureTask<T>(runnable, value);
    }

    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new NonBlockingFutureTask<T>(callable);
    }
		
}
