package psl;
   
   import java.util.concurrent.LinkedBlockingQueue;
   import java.util.concurrent.ThreadPoolExecutor;
   import java.util.concurrent.TimeUnit;
   
   import ensure.Ensure;
   
   /**
   * Dispatcher used to run asynchronous requests. A dispatcher runs several
   * threads that execute tasks submitted. It is commonly used to register
   * {@link AsyncDispatchRunnable} tasks but any runnable task may be registered.
   */
  public class AsyncDispatcher {
  	/**
  	 * Time to wait between polls to check whether threads have terminated
  	 * or not.
  	 */
  	private static final long TERMINATION_POLLING_WAIT_TIME_MS = 25;
  	
  	/**
  	 * The executor, <code>null</code> if not alive.
  	 */
  	private ThreadPoolExecutor m_executor;
  	
  	/**
  	 * The thread factory.
  	 */
  	private DispatcherThreadFactory m_factory;
  	
  	/**
  	 * Number of tasks waiting to be executed.
  	 */
  	private long m_waiting;
  	
  	/**
  	 * Number of tasks currently executing.
  	 */
  	private long m_executing;
  	
  	/**
  	 * Creates a new dispatcher.
  	 * @param name the dispatcher name
  	 * @param handler the exception handler; may be <code>null</code> in which
  	 * case the default handler will be used 
  	 * @param threads number of threads to use in the dispatcher
  	 */
  	public AsyncDispatcher(String name, ExceptionHandler handler, int threads) {
  		Ensure.not_null(name, "name == null");
  		Ensure.greater_equal(threads, 1, "threads < 1");
  		
  		final ExceptionHandler h;
  		if (handler == null) {
  			h = new DefaultExceptionHandler();
  		} else {
  			h = handler;
  		}
  		
  		m_factory = DispatcherThreadFactory.make(name, h);
  		m_executor = new ThreadPoolExecutor(threads, threads, 0,
  				TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(),
  				m_factory) {
  			@Override
  			protected void afterExecute(Runnable r, Throwable t) {
  				super.afterExecute(r, t);
  				if (t != null) {
  					h.thrown(r, t, Thread.currentThread());
  				}
  			}
  		};
  		
  		m_executing = 0;
  		m_waiting = 0;
  	}
  	
  	/**
  	 * Checks whether the dispatcher is still alive.
  	 * @return is the dispatcher alive?
  	 */
  	public synchronized boolean alive() {
  		return m_executor != null;
  	}
  	
  	/**
  	 * Shuts down the dispatcher.
  	 */
  	public void shutdown() {
  		ThreadPoolExecutor executor;
  		
  		synchronized (this) {
  			if (m_executor == null) {
  				return;
  			}
  			
  			executor = m_executor;
  			m_executor = null;
  		}
  		
  		executor.shutdownNow();
 		
 		while (true) {
 			boolean shutdown = executor.isTerminated();
 			if (shutdown) {
 				m_factory.shutdown();
 				return;
 			}
 			
 			try {
 				Thread.sleep(TERMINATION_POLLING_WAIT_TIME_MS);
 			} catch (InterruptedException e) {
 				/*
 				 * Ignored.
 				 */
 			}
 		}
 	}
 	
 	/**
 	 * Submits a task for execution.
 	 * @param r the task, cannot be <code>null</code>
 	 */
 	public synchronized void submit(final Runnable r) {
 		Ensure.not_null(r, "r == null");
 		Ensure.not_null(m_executor, "m_executor == null");
 		
 		m_waiting++;
 		m_executor.submit(new Runnable() {
 			@Override
 			public void run() {
 				synchronized (AsyncDispatcher.this) {
 					Ensure.greater(m_waiting, 0, "m_waiting <= 0");
 					m_waiting--;
 					m_executing++;
 				}
 				
 				try {
 					new ExecutorRunnable(r).run();
 				} finally {
 					synchronized (AsyncDispatcher.this) {
 						Ensure.greater(m_executing, 0, "m_executing <= 0");
 						m_executing--;
 						AsyncDispatcher.this.notifyAll();
 					}
 				}
 			}
 		});
 	}
 	
 	/**
 	 * Waits until there are no more tasks to process. Because this method is
 	 * unsafe as tasks can be submitted right after the method has returned,
 	 * it is usually only useful in tightly-controlled environments such as
 	 * unit tests. 
 	 */
 	public synchronized void wait_clear() {
 		while (m_waiting > 0 || m_executing > 0) {
 			try {
 				wait();
 			} catch (InterruptedException e) {
 				/*
 				 * Just cycle again.
 				 */
 			}
 		}
 	}
 }