/*
Copyright (C) 2001, 2006 United States Government
as represented by the Administrator of the
National Aeronautics and Space Administration.
All Rights Reserved.
*/
package gov.nasa.worldwind;

/**
 * @author Tom Gaskins
 * @version $Id: ThreadedTaskService.java 1792 2007-05-08 21:28:37Z tgaskins $
 */
public class ThreadedTaskService extends WWObjectImpl // TODO: Extract interface
    implements Thread.UncaughtExceptionHandler
{
    static final private int DEFAULT_CORE_POOL_SIZE = 1;
    static final private int DEFAULT_QUEUE_SIZE = 10;
    private static final String RUNNING_THREAD_NAME_PREFIX = WorldWind.retrieveMessage(
        "ThreadedTaskService.RUNNING_THREAD_NAME_PREFIX", "ThreadStrings");
    private static final String IDLE_THREAD_NAME_PREFIX = WorldWind.retrieveMessage(
        "ThreadedTaskService.IDLE_THREAD_NAME_PREFIX", "ThreadStrings");
    private java.util.concurrent.ConcurrentLinkedQueue<Runnable> activeTasks; // tasks currently allocated a thread
    private TaskExecutor executor; // thread pool for running retrievers

    public ThreadedTaskService()
    {
        Integer poolSize = Configuration.getIntegerValue(AVKey.THREADED_TASK_POOL_SIZE, DEFAULT_CORE_POOL_SIZE);
        Integer queueSize = Configuration.getIntegerValue(AVKey.THREADED_TASK_QUEUE_SIZE, DEFAULT_QUEUE_SIZE);

        // this.executor runs the tasks, each in their own thread
        this.executor = new TaskExecutor(poolSize, queueSize);

        // this.activeTasks holds the list of currently executing tasks
        this.activeTasks = new java.util.concurrent.ConcurrentLinkedQueue<Runnable>();
    }

    public void uncaughtException(Thread thread, Throwable throwable)
    {
        if (throwable instanceof WWDuplicateRequestException)
            return;

        String message = WorldWind.retrieveErrMsg("ThreadedTaskService.UncaughtExceptionDuringTask") + thread.getName();
        WorldWind.logger().log(java.util.logging.Level.FINE, message);
        Thread.currentThread().getThreadGroup().uncaughtException(thread, throwable);
    }

    private class TaskExecutor extends java.util.concurrent.ThreadPoolExecutor
    {
        private static final long THREAD_TIMEOUT = 2; // keep idle threads alive this many seconds

        private TaskExecutor(int poolSize, int queueSize)
        {
            super(poolSize, poolSize, THREAD_TIMEOUT, java.util.concurrent.TimeUnit.SECONDS,
                new java.util.concurrent.ArrayBlockingQueue<Runnable>(queueSize),
                new java.util.concurrent.ThreadFactory()
                {
                    public Thread newThread(Runnable runnable)
                    {
                        Thread thread = new Thread(runnable);
                        thread.setDaemon(true);
                        thread.setPriority(Thread.MIN_PRIORITY);
                        thread.setUncaughtExceptionHandler(ThreadedTaskService.this);
                        return thread;
                    }
                }, new java.util.concurrent.ThreadPoolExecutor.DiscardPolicy() // abandon task when queue is full
            {
                public void rejectedExecution(Runnable runnable,
                    java.util.concurrent.ThreadPoolExecutor threadPoolExecutor)
                {
                    // Interposes logging for rejected execution
                    String message = WorldWind.retrieveErrMsg("ThreadedTaskService.ResourceRejected") + runnable;
                    WorldWind.logger().log(java.util.logging.Level.FINEST, message);
                    super.rejectedExecution(runnable, threadPoolExecutor);
                }
            });
        }

        protected void beforeExecute(Thread thread, Runnable runnable)
        {
            WorldWind.logger().log(java.util.logging.Level.FINEST, WorldWind.retrieveErrMsg(
                "ThreadedTaskService.EnteringBeforeExecute"));

            if (thread == null)
            {
                String msg = WorldWind.retrieveErrMsg("nullValue.ThreadIsNull");
                WorldWind.logger().log(java.util.logging.Level.FINE, msg);
                throw new IllegalArgumentException(msg);
            }

            if (runnable == null)
            {
                String msg = WorldWind.retrieveErrMsg("nullValue.RunnableIsNull");
                WorldWind.logger().log(java.util.logging.Level.FINE, msg);
                throw new IllegalArgumentException(msg);
            }

            if (ThreadedTaskService.this.activeTasks.contains(runnable))
            {
                String message = WorldWind.retrieveErrMsg("ThreadedTaskService.CancellingDuplicateTask") + runnable;
                WorldWind.logger().log(java.util.logging.Level.FINER, message);
                throw new WWDuplicateRequestException(message);
            }

            ThreadedTaskService.this.activeTasks.add(runnable);

            if (RUNNING_THREAD_NAME_PREFIX != null)
                thread.setName(RUNNING_THREAD_NAME_PREFIX + runnable);
            thread.setPriority(Thread.MIN_PRIORITY);
            thread.setUncaughtExceptionHandler(ThreadedTaskService.this);

            super.beforeExecute(thread, runnable);

            WorldWind.logger().log(java.util.logging.Level.FINEST, WorldWind.retrieveErrMsg(
                "ThreadedTaskService.LeavingBeforeExecute"));
        }

        protected void afterExecute(Runnable runnable, Throwable throwable)
        {
            WorldWind.logger().log(java.util.logging.Level.FINEST, WorldWind.retrieveErrMsg(
                "ThreadedTaskService.EnteringAfterExecute"));

            if (runnable == null)
            {
                String msg = WorldWind.retrieveErrMsg("nullValue.RunnableIsNull");
                WorldWind.logger().log(java.util.logging.Level.FINE, msg);
                throw new IllegalArgumentException(msg);
            }

            super.afterExecute(runnable, throwable);

            ThreadedTaskService.this.activeTasks.remove(runnable);

            if (throwable == null && IDLE_THREAD_NAME_PREFIX != null)
                Thread.currentThread().setName(IDLE_THREAD_NAME_PREFIX);
        }
    }

    public synchronized boolean contains(Runnable runnable)
    {
        //noinspection SimplifiableIfStatement
        if (runnable == null)
            return false;

        return (this.activeTasks.contains(runnable) || this.executor.getQueue().contains(runnable));
    }

    /**
     * Enqueues a task to run.
     * @param runnable the task to add
     * @throws IllegalArgumentException if <code>runnable</code> is null
     */
    public synchronized void addTask(Runnable runnable)
    {
        if (runnable == null)
        {
            String message = WorldWind.retrieveErrMsg("nullValue.RunnableIsNull");
            WorldWind.logger().log(java.util.logging.Level.FINE, message);
            throw new IllegalArgumentException(message);
        }

        // Do not queue duplicates.
        if (this.activeTasks.contains(runnable) || this.executor.getQueue().contains(runnable))
            return;

        this.executor.execute(runnable);
    }

    public boolean isFull()
    {
        return this.executor.getQueue().remainingCapacity() == 0;
    }
}
