import java.util.ArrayList;

import java.util.Collection;

import java.util.Set;

import java.util.concurrent.Callable;

import java.util.concurrent.CancellationException;

import java.util.concurrent.CountDownLatch;

import java.util.concurrent.Delayed;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.Future;

import java.util.concurrent.FutureTask;

import java.util.concurrent.RejectedExecutionException;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.ScheduledFuture;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.TimeoutException;

import java.util.concurrent.atomic.AtomicBoolean;

import java.util.concurrent.atomic.AtomicLong;

import java.util.concurrent.atomic.AtomicReference;

    /**

     * Shut down this thread pool as defined in

     * {@link java.util.concurrent.ExecutorService#shutdown()}.<p/>If this method

     * is called on the <a href="#getDefault()">default RequestProcessor</a>,

     * throws an IllegalStateException.

     * @since 8.2

     */

    @Override

    public void shutdown() {

        if (this == UNLIMITED) {

            throw new IllegalStateException ("Cannot shut down the default " + //NOI18N

                    "request processor"); //NOI18N

        }

        stop();

    }

    /**

     * Shut down this thread pool as defined in

     * {@link java.util.concurrent.ExecutorService#shutdownNow()}, returning

     * a list of un-run or unfinished runnables.  <p/> If this method

     * is called on the <a href="#getDefault()">default RequestProcessor</a>,

     * throws an IllegalStateException.

     * @since 8.2

     */

    @Override

    public List<Runnable> shutdownNow() {

        if (this == UNLIMITED) {

            throw new IllegalStateException ("Cannot shut down the default " + //NOI18N

                    "request processor"); //NOI18N

        }

        //XXX more aggressive shutdown?

        stop();

        synchronized (processorLock) {

            List<Runnable> result = new ArrayList<Runnable>(processors.size());

            for (Processor p : processors) {

                if (p != null && p.todo != null && p.todo.run != null) {

                    Runnable r = p.todo.run;

                    if (r instanceof RunnableWrapper) {

                        Runnable other = ((RunnableWrapper) r).getRunnable();

                        r = other == null ? r : other;

                    }

                    result.add(r);

                }

            }

            return result;

        }

    }

    /**

     * Determine if this thread pool has been shut down, in accordance with

     * the specification of {@link java.util.concurrent.ExecutorService#shutdown()}

     * @return

     * @since 8.2

     */

    @Override

    public boolean isShutdown() {

        return stopped;

    }

    /**

     * Determine if this thread pool has been terminated, in accordance with

     * the specification of {@link java.util.concurrent.ExecutorService#isTerminated()}

     * @return

     * @since 8.2

     */

    @Override

    public boolean isTerminated() {

        boolean result = true;

        Set<Processor> set = collectProcessors(new HashSet<Processor>());

        for (Processor p : set) {

            if (p.isAlive() && p.belongsTo(this)) {

                result = false;

                break;

            }

        }

        return result;

    }

    /**

     * Determine if this thread pool has been shut down, in accordance with

     * the specification of {@link java.util.concurrent.ExecutorService#shutdown()}

     * @return True if the request processor is shut down before the timeout

     * @since 8.2

     */

    @Override

    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {

        Parameters.notNull("unit", unit); //NOI18N

        long timeoutMillis = unit.convert(timeout, TimeUnit.MILLISECONDS);

        boolean result = stopped;

        long doneTime = System.currentTimeMillis() + timeoutMillis;

        Set<Processor> procs = new HashSet<Processor>();

outer:  do {

            procs = collectProcessors(procs);

            if (procs.isEmpty()) {

                return true;

            }

            for (Processor p : procs) {

                long remaining = doneTime - System.currentTimeMillis();

                if (remaining <= 0) {

                    result = collectProcessors(procs).isEmpty();

                    break outer;

                }

                if (p.belongsTo(this)) {

                    p.join(remaining);

                }

                result = !p.isAlive() || !p.belongsTo(this);

            }

            procs.clear();

        } while (!procs.isEmpty());

        return result;

    }

    private Set<Processor> collectProcessors (Set<Processor> procs) {

        procs.clear();

        synchronized (processorLock) {

            for (Processor p : processors) {

                if (p.belongsTo(this)) {

                    procs.add(p);

                }

            }

        }

        return procs;

    }

    /**

     * Submit a job to be run, and get back a {@link java.util.concurrent.Future}

     * which can be waited on and asked for a result, following the contract of

     * {@link java.util.concurrent.ExecutorService#submit(java.util.concurrent.Callable)}

     * <p/>

     * <b>Note:</b> If the passed {@link java.util.concurrent.Callable} implements

     * {@link org.openide.util.Cancellable}, then that object's {@link org.openide.util.Cancellable#cancel()}

     * method will be called if {@link java.util.concurrent.Future#cancel(boolean)} is invoked.

     * If <code>Cancellable.cancel()</code> returns false, then <i>the job will <u>not</u> be

     * cancelled</i>.

     *

     * @param <T> The return type of the passed job

     * @param task The work to do

     * @return A Future that can indicate work status and fetch a result

     * @throws RejectedExecutionException if this RequestProcessor has been

     * shut down

     * @since 8.2

     */

    @Override

    public <T> Future<T> submit(Callable<T> task) {

        Parameters.notNull("task", task); //NOI18N

        if (stopped) {

            throw new RejectedExecutionException("Request Processor already " + //NOI18N

                    "stopped"); //NOI18N

        }

        RPFutureTask<T> result = new RPFutureTask<T>(task);

        Task t = create(result);

        result.setTask(t);

        t.schedule(0);

        return result;

    }

    /**

     * Submit a job to be run, and get back a {@link java.util.concurrent.Future}

     * which can be waited on and asked for a result, following the contract of

     * {@link java.util.concurrent.ExecutorService#submit(java.lang.Runnable, T)}

     * <p/>

     * <b>Note:</b> If the passed {@link java.lang.Runnable} implements

     * {@link org.openide.util.Cancellable}, then that object's {@link org.openide.util.Cancellable#cancel()}

     * method will be called if {@link java.util.concurrent.Future#cancel(boolean)} is invoked.

     * If <code>Cancellable.cancel()</code> returns false, then <i>the job will <u>not</u> be

     * cancelled</i>.

     *

     * @param <T> The return type of the passed job

     * @param task The work to do

     * @param predefinedResult The return value for the resulting Future's

     * <code>get()</code> method

     * @return A Future that can indicate work status and fetch the result

     * @throws RejectedExecutionException if this RequestProcessor has been

     * shut down

     * @since 8.2

     */

    @Override

    public <T> Future<T> submit(Runnable task, T predefinedResult) {

        Parameters.notNull("task", task); //NOI18N

        if (stopped) {

            throw new RejectedExecutionException("Request Processor already " + //NOI18N

                    "stopped"); //NOI18N

        }

        RPFutureTask<T> result = new RPFutureTask<T>(task, predefinedResult);

        Task t = create(result);

        result.setTask(t);

        t.schedule(0);

        return result;

    }

    /**

     * Submit a job to be run, and get back a {@link java.util.concurrent.Future}

     * which can be waited on and asked for a result, following the contract of

     * {@link java.util.concurrent.ExecutorService#submit(java.lang.Runnable)}

     * <p/>

     * <b>Note:</b> If the passed {@link java.lang.Runnable} implements

     * {@link org.openide.util.Cancellable}, then that object's {@link org.openide.util.Cancellable#cancel()}

     * method will be called if {@link java.util.concurrent.Future#cancel(boolean)} is invoked.

     * If <code>Cancellable.cancel()</code> returns false, then <i>the job will <u>not</u> be

     * cancelled</i>.

     *

     * @param <T> The return type of the passed job

     * @param task The work to do

     * @return A Future that can indicate work status and fetch a result

     * @throws RejectedExecutionException if this RequestProcessor has been

     * shut down

     * @since 8.2

     */

    @Override

    public Future<?> submit(Runnable task) {

        return this.<Void>submit (task, null);

    }

    /**

     * Invoke a collection of tasks, as defined in the specification of

     * {@link java.util.concurrent.ExecutorService#invokeAll(java.util.Collection)}

     * @param <T> The type of returned object

     * @param tasks A collection of Callables with the same return type

     * @return A list of futures which can be monitored for completion

     * @throws InterruptedException If the timeout expires

     * @since 8.2

     */

    @Override

    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {

        Parameters.notNull("tasks", tasks); //NOI18N

        List<Future<T>> result = new ArrayList<Future<T>>(tasks.size());

        CountDownLatch wait = new CountDownLatch(tasks.size());

        for (Callable<T> c : tasks) {

            if (c == null) {

                    throw new NullPointerException ("Contains null tasks: " +  //NOI18N

                            tasks);

            }

            Callable<T> delegate = new WaitableCallable<T>(c, wait);

            result.add (submit(delegate));

        }

        wait.await();

        return result;

    }

    /**

     * Executes the given tasks, returning a list of Futures holding their

     * status and results when all complete or the timeout expires, whichever

     * happens first, as specified in

     * {@link java.util.concurrent.ExecutorService#invokeAll((java.util.Collection,long,java.util.concurrent.TimeUnit))}

     * @param <T> The result type

     * @param tasks A collection of callables

     * @param timeout The maximum time to wait for completion, in the specified time units

     * @param unit The time unit

     * @return A list of futures

     * @throws InterruptedException if the timeout expires

     * @since 8.2

     */

    @Override

    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {

        Parameters.notNull("unit", unit); //NOI18N

        Parameters.notNull("tasks", tasks); //NOI18N

        CountDownLatch wait = new CountDownLatch(tasks.size());

        List<Future<T>> result = new ArrayList<Future<T>>(tasks.size());

        for (Callable<T> c : tasks) {

            if (c == null) {

                throw new NullPointerException ("Contains null tasks: " + tasks); //NOI18N

            }

            Callable<T> delegate = new WaitableCallable<T>(c, wait);

            result.add (submit(delegate));

        }

        if (!wait.await(timeout, unit)) {

            for (Future<T> f : result) {

                RPFutureTask<?> ft = (RPFutureTask<?>) f;

                ft.cancel(true);

            }

        }

        return result;

    }

    /**

     * Executes the given tasks, returning the result of one which has

     * completed and cancelling any incomplete tasks, as specified in

     * {@link java.util.concurrent.ExecutorService#invokeAny((java.util.Collection))}

     * @param <T> The result type

     * @param tasks A collection of callables

     * @return A list of futures

     * @throws InterruptedException if execution is interrupted

     * @since 8.2

     */

    @Override

    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {

        Parameters.notNull("tasks", tasks); //NOI18N

        CountDownLatch wait = new CountDownLatch(1);

        List<Future<T>> result = new ArrayList<Future<T>>(tasks.size());

        AtomicReference<T> ref = new AtomicReference<T>();

        try {

            for (Callable<T> c : tasks) {

                if (c == null) {

                    throw new NullPointerException ("Contains null tasks: " +  //NOI18N

                            tasks);

                }

                Callable<T> delegate = new WaitableCallable<T>(c, ref, wait);

                result.add (submit(delegate));

            }

            wait.await();

        } finally {

            for (Future<T> f : result) {

                RPFutureTask<?> ft = (RPFutureTask<?>) f;

                ft.cancel(true);

            }

        }

        return ref.get();

    }

    /**

     * Executes the given tasks, returning a list of Futures holding their

     * status and results when all complete or the timeout expires, whichever

     * happens first, as specified in

     * {@link java.util.concurrent.ExecutorService#invokeAny((java.util.Collection))}

     * @param <T> The result type

     * @param tasks A collection of callables

     * @param timeout The maximum time to wait for completion, in the specified time units

     * @param unit The time unit

     * @return A list of futures

     * @throws InterruptedException if the timeout expires or execution is interrupted

     * @since 8.2

     */

    @Override

    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {

        Parameters.notNull("unit", unit); //NOI18N

        Parameters.notNull("tasks", tasks); //NOI18N

        CountDownLatch wait = new CountDownLatch(1);

        List<Future<T>> result = new ArrayList<Future<T>>(tasks.size());

        AtomicReference<T> ref = new AtomicReference<T>();

        try {

            for (Callable<T> c : tasks) {

                if (c == null) {

                    throw new NullPointerException ("Contains null tasks: " +  //NOI18N

                            tasks);

                }

                Callable<T> delegate = new WaitableCallable(c, ref, wait);

                result.add (submit(delegate));

            }

            wait.await(timeout, unit);

        } finally {

            for (Future<T> f : result) {

                RPFutureTask ft = (RPFutureTask) f;

                ft.cancel(true);

            }

        }

        return ref.get();

    }

    /**

     * Schedule a runnable to execute after the specified delay.

     * @param command The runnable

     * @param delay The delay

     * @param unit The time units of the delay

     * @return A future which can be monitored for completion

     * @since 8.2

     */

    @Override

    public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {

        Parameters.notNull("command", command); //NOI18N

        Parameters.notNull("unit", unit); //NOI18N

        if (delay < 0) {

            throw new IllegalArgumentException ("Negative delay: " + delay);

        }

        if (stopped) {

            throw new RejectedExecutionException("Request Processor already stopped"); //NOI18N

        }

        long delayMillis = unit.convert(delay, TimeUnit.MILLISECONDS);

        if (delayMillis > Integer.MAX_VALUE) {

            throw new IllegalArgumentException ("Requested delay " + delayMillis +  //NOI18N

                    " is > Integer.MAX_VALUE"); //NOI18N

        }

        ScheduledRPFutureTask<Void> result = new ScheduledRPFutureTask<Void>(command, null, delayMillis);

        Task t = create(result);

        result.setTask(t);

        t.schedule((int) delayMillis);

        return result;

    }

    /**

     * Schedule a {@link java.util.concurrent.Callable} to execute on another thread

     * after the specified delay.

     * @param command The work to run

     * @param delay The delay

     * @param unit The time units of the delay

     * @return A future which can be monitored for completion

     * @since 8.2

     */

    @Override

    public <T> ScheduledFuture<T> schedule(Callable<T> callable, long delay, TimeUnit unit) {

        Parameters.notNull("unit", unit); //NOI18N

        Parameters.notNull("callable", callable); //NOI18N

        if (delay < 0) {

            throw new IllegalArgumentException ("Negative delay: " + delay);

        }

        if (stopped) {

            throw new RejectedExecutionException("Request Processor already " + //NOI18N

                    "stopped"); //NOI18N

        }

        long delayMillis = unit.convert(delay, TimeUnit.MILLISECONDS);

        if (delayMillis > Integer.MAX_VALUE) {

            throw new IllegalArgumentException ("Requested delay " +  //NOI18N

                    delayMillis + " is > Integer.MAX_VALUE"); //NOI18N

        }

        ScheduledRPFutureTask<T> result = new ScheduledRPFutureTask<T>(callable, delayMillis);

        Task t = create(result);

        result.setTask(t);

        t.schedule((int) delayMillis);

        return result;

    }

    /**

     * Schedule a runnable which will run with a given frequency, regardless

     * of how long execution takes, with the exception that if execution takes

     * longer than the specified delay, execution will be delayed but will

     * never be run on two threads concurrently.

     * As specified in

     * {@link java.util.concurrent.ExecutorService.scheduleAtFixedRate(java.lang.Runnable,long,long,java.util.concurrent.TimeUnit)}

     * @param command The runnable

     * @param initialDelay The delay before first running

     * @param period The frequency with which the runnable should run after the

     * first run

     * @param unit The time units in which the initial delay and period are

     * specified

     * @return A future which can be monitored for completion, or cancelled

     * @since 8.2

     */

    @Override

    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {

        return scheduleFixed(command, initialDelay, period, unit, false);

    }

    /**

     * Schedule a runnable which will run repeatedly after the specified initial

     * delay, with the specified delay between the completion of one run and

     * the start of the next.

     * As specified in

     * {@link java.util.concurrent.ExecutorService.scheduleWithFixedDelay(java.lang.Runnable,long,long,java.util.concurrent.TimeUnit)}

     * @param command The runnable

     * @param initialDelay The delay before first run

     * @param delay The delay between runs

     * @param unit The time units of the delay parameters

     * @return A future which can be monitored for completion, or cancelled

     * @since 8.2

     */

    @Override

    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {

        return scheduleFixed(command, initialDelay, delay, unit, true);

    }

    private ScheduledFuture<?> scheduleFixed (Runnable command, long initialDelay, long period, TimeUnit unit, boolean fixedDelay) {

        Parameters.notNull("unit", unit); //NOI18N

        Parameters.notNull("command", command); //NOI18N

        if (period < 0) {

            throw new IllegalArgumentException ("Negative delay: " + period); //NOI18N

        }

        if (initialDelay < 0) {

            throw new IllegalArgumentException ("Negative initialDelay: "  //NOI18N

                    + initialDelay);

        }

        if (stopped) {

            throw new RejectedExecutionException("Request Processor already " + //NOI18N

                    "stopped"); //NOI18N

        }

        long initialDelayMillis = unit.convert(initialDelay, TimeUnit.MILLISECONDS);

        if (initialDelayMillis > Integer.MAX_VALUE) {

            throw new IllegalArgumentException("Initial delay > " + //NOI18N

                    "Integer.MAX_VALUE milliseconds: " + initialDelayMillis); //NOI18N

        }

        long periodMillis = unit.convert(period, TimeUnit.MILLISECONDS);

        if (periodMillis > Integer.MAX_VALUE) {

            throw new IllegalArgumentException("Initial delay > " + //NOI18N

                    "Integer.MAX_VALUE milliseconds: " + periodMillis); //NOI18N

        }

        TaskFutureWrapper wrap = fixedDelay ? 

            new FixedDelayTask(command, initialDelayMillis, periodMillis) :

            new FixedRateTask(command, initialDelay, periodMillis);

        Task t = create(wrap);

        wrap.t = t;

        t.cancelled = wrap.cancelled;

        t.schedule ((int) initialDelayMillis);

        return wrap;

    }

    private static abstract class TaskFutureWrapper implements ScheduledFuture<Void>, Runnable, RunnableWrapper {

        volatile Task t;

        protected final Runnable toRun;

        protected final long initialDelay;

        protected final long period;

        final AtomicBoolean cancelled = new AtomicBoolean();

        TaskFutureWrapper(Runnable run, long initialDelay, long period) {

            this.toRun = run;

            this.initialDelay = initialDelay;

            this.period = period;

        }

        @Override

        public final Runnable getRunnable() {

            return toRun;

        }

        @Override

        public int compareTo(Delayed o) {

            long other = o.getDelay(TimeUnit.MILLISECONDS);

            long ours = getDelay(TimeUnit.MILLISECONDS);

            //Might overflow on, say, ms compared to Long.MAX_VALUE, TimeUnit.DAYS

            return (int) (ours - other);

        }

        @Override

        public boolean cancel(boolean mayInterruptIfRunning) {

            boolean result = true;

            if (toRun instanceof Cancellable) {

                result = ((Cancellable) toRun).cancel();

            }

            if (result) {

                //will invoke cancelled.set(true)

                result = t.cancel(mayInterruptIfRunning);

            }

            return result;

        }

        @Override

        public boolean isCancelled() {

            return cancelled.get();

        }

        @Override

        public boolean isDone() {

            return cancelled.get() || t.isFinished();

        }

        @Override

        public Void get() throws InterruptedException, ExecutionException {

            if (cancelled.get()) {

                throw new CancellationException();

            }

            t.waitFinished();

            if (cancelled.get()) {

                throw new CancellationException();

            }

            return null;

        }

        @Override

        public Void get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {

            if (cancelled.get()) {

                throw new CancellationException();

            }

            long millis = unit.convert(timeout, TimeUnit.MILLISECONDS);

            t.waitFinished(millis);

            if (cancelled.get()) {

                throw new CancellationException();

            }

            return null;

        }

    }

    private static final class FixedRateTask extends TaskFutureWrapper {

        private final Object runLock = new Object();

        private final Object timeLock = new Object();

        //must be accessed holding timeLock

        private int runCount;

        private long nextRunTime;

        private long start = Long.MIN_VALUE;

        volatile boolean firstRun = true;

        FixedRateTask (Runnable run, long initialDelay, long period) {

            super (run, initialDelay, period);

        }

        @Override

        public void run() {

            if (firstRun) {

                synchronized (timeLock) {

                    start = System.currentTimeMillis();

                    firstRun = false;

                }

            }

            try {

                synchronized(runLock) {

                    toRun.run();

                }

            } catch (RuntimeException e) {

                cancel(true);

                throw e;

            }

            reschedule();

        }

        private void reschedule() {

            //All access to nextRunTime & runCount under lock.

            long interval;

            synchronized (timeLock) {

                nextRunTime = start + (initialDelay + period * runCount);

                runCount++;

                interval = Math.max(0,  nextRunTime - System.currentTimeMillis());

                if (interval > Integer.MAX_VALUE) {

                    throw new IllegalStateException ("Interval > Integer.MAX_VALUE: " + interval); //NOI18N

                }

            }

            boolean canContinue = !cancelled.get() && !Thread.currentThread().isInterrupted();

            if (canContinue) {

                t.schedule((int) interval);

            }

        }

        @Override

        public long getDelay(TimeUnit unit) {

            if (isCancelled()) {

                return Long.MAX_VALUE;

            }

            long delay;

            synchronized (timeLock) {

                delay = Math.min(0, nextRunTime - System.currentTimeMillis());

            }

            return unit.convert(delay, TimeUnit.MILLISECONDS);

        }

    }

    private static final class FixedDelayTask extends TaskFutureWrapper {

        private volatile boolean firstRun = true;

        private final AtomicLong nextRunTime = new AtomicLong();

        FixedDelayTask(Runnable run, long initialDelay, long period)  {

            super (run, initialDelay, period);

        }

        @Override

        public long getDelay(TimeUnit unit) {

            long next = nextRunTime.get();

            return unit.convert (next - System.currentTimeMillis(), TimeUnit.MILLISECONDS);

        }

        @Override

        public void run() {

            if (!fini()) {

                toRun.run();

            }

            if (!fini()) {

                reschedule();

            }

        }

        private boolean fini() {

            boolean result = cancelled.get() || Thread.currentThread().isInterrupted();

            return result;

        }

        private void reschedule() {

            long delay;

            if (firstRun) {

                delay = initialDelay;

            } else {

                delay = period;

            }

            nextRunTime.set(System.currentTimeMillis() + delay);

            firstRun = false;

            if (!fini()) {

                t.schedule((int) delay);

            }

        }

    }

    private interface RunnableWrapper {

        Runnable getRunnable();

    }

    private static final class WaitableCallable<T> implements Callable<T>, Cancellable {

        private final CountDownLatch countdown;

        private final Callable<T> delegate;

        private final AtomicReference<T> ref;

        private volatile boolean failed;

        WaitableCallable(Callable<T> delegate, CountDownLatch countdown) {

            this (delegate, null, countdown);

        }

        WaitableCallable(Callable<T> delegate, AtomicReference<T> ref, CountDownLatch countdown) {

            this.delegate = delegate;

            this.countdown = countdown;

            this.ref = ref;

        }

        boolean failed() {

            return failed;

        }

        @Override

        public T call() throws Exception {

            try {

                T result = delegate.call();

                if (ref != null) {

                    ref.set(result);

                }

                return result;

            } catch (RuntimeException e) {

                failed = true;

                throw e;

            } catch (Error e) {

                failed = true;

                throw e;

            } finally {

                if (!failed || ref == null) {

                    countdown.countDown();

                }

            }

        }

        @Override

        public boolean cancel() {

            return delegate instanceof Cancellable ? ((Cancellable) delegate).cancel() : true;

        }

    }

    private static class RPFutureTask<T> extends FutureTask<T> implements RunnableWrapper {

        protected volatile Task task;

        private final Runnable runnable;

        private final Cancellable cancellable;

        RPFutureTask(Callable<T> c) {

            super (c);

            this.runnable = null;

            this.cancellable = c instanceof Cancellable ? (Cancellable) c : null;

        }

        RPFutureTask(Runnable r, T result) {

            super (r, result);

            this.runnable = r;

            this.cancellable = r instanceof Cancellable ? (Cancellable) r : null;

        }

        void setTask(Task task) {

            this.task = task;

        }

        RPFutureTask(Callable<T> c, T predefinedResult) {

            this (c);

            set(predefinedResult);

        }

        @Override

        public boolean cancel(boolean mayInterruptIfRunning) {

            if (cancellable != null) {

                boolean result = cancellable.cancel();

                if (result) {

                    //note & not && - task.cancel() and super.cancel() must be invoked,

                    //if and only if the underlying cancellable is really null or

                    //returned true from cancel().

                    return task.cancel() & super.cancel(mayInterruptIfRunning);

                } else {

                    return result;

                }

            } else {

                return task.cancel() & super.cancel(mayInterruptIfRunning);

            }

        }

        @Override

        public Runnable getRunnable() {

            return this.runnable;

        }

    }

    private static final class ScheduledRPFutureTask<T> extends RPFutureTask<T> implements ScheduledFuture<T> {

        protected final long delayMillis;

        ScheduledRPFutureTask(Callable<T> c, long delayMillis) {

            super (c);

            this.delayMillis = delayMillis;

        }

        ScheduledRPFutureTask(Runnable r, T result, long delayMillis) {

            super (r, result);

            this.delayMillis = delayMillis;

        }

        @Override

        public long getDelay(TimeUnit unit) {

            return TimeUnit.MILLISECONDS.convert(delayMillis, unit);

        }

        @Override

        public int compareTo(Delayed o) {

            //Can overflow, if one delay is, say, days, and the other, microseconds

            long otherDelayMillis = o.getDelay(TimeUnit.MILLISECONDS);

            return (int) (delayMillis - otherDelayMillis);

        }

    }

        private AtomicBoolean cancelled;

                if (cancelled != null) {

                    cancelled.set(false);

                }

        /**

         * Implementation of cancel for use with Future objects, to guarantee

         * that the thread will be interrupted, no matter what the setting

         * on the owning RP.

         * @param interrupt If true, the thread should be interrupted

         * @return true if cancellation occurred

         */

        boolean cancel (boolean interrupt) {

            synchronized (processorLock) {

                if (cancelled != null) {

                    boolean wasCancelled = !cancelled.getAndSet(true);

                    if (wasCancelled) {

                        return false;

                    }

                }

                boolean success;

                if (item == null) {

                    success = false;

                } else {

                    Processor p = item.getProcessor();

                    success = item.clear(null);

                    if (p != null) {

                        if (interrupt) {

                            success = p.interrupt(this, RequestProcessor.this);

                        } else {

                            //despite its name, will not actually interrupt

                            //unless the RP specifies that it should

                            p.interruptTask(this, RequestProcessor.this);

                        }

                        if (success) {

                            item = null;

                        }

                    }

                }

                if (success) {

                    notifyFinished(); // mark it as finished

                }

                return success;

            }

        }

        boolean belongsTo(RequestProcessor r) {

            synchronized (lock) {

                return source == r;

            }

        }

        boolean interrupt (Task t, RequestProcessor src) {

            if (t != todo) {

                return false;

            }

            interrupt();

            return true;

        }

/*

 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.

 *

 * Copyright 2010 Sun Microsystems, Inc. All rights reserved.

 *

 * The contents of this file are subject to the terms of either the GNU

 * General Public License Version 2 only ("GPL") or the Common

 * Development and Distribution License("CDDL") (collectively, the

 * "License"). You may not use this file except in compliance with the

 * License. You can obtain a copy of the License at

 * http://www.netbeans.org/cddl-gplv2.html

 * or nbbuild/licenses/CDDL-GPL-2-CP. See the License for the

 * specific language governing permissions and limitations under the

 * License.  When distributing the software, include this License Header

 * Notice in each file and include the License file at

 * nbbuild/licenses/CDDL-GPL-2-CP.  Sun designates this

 * particular file as subject to the "Classpath" exception as provided

 * by Sun in the GPL Version 2 section of the License file that

 * accompanied this code. If applicable, add the following below the

 * License Header, with the fields enclosed by brackets [] replaced by

 * your own identifying information:

 * "Portions Copyrighted [year] [name of copyright owner]"

 *

 * If you wish your version of this file to be governed by only the CDDL

 * or only the GPL Version 2, indicate your decision by adding

 * "[Contributor] elects to include this software in this distribution

 * under the [CDDL or GPL Version 2] license." If you do not indicate a

 * single choice of license, a recipient has the option to distribute

 * your version of this file under either the CDDL, the GPL Version 2 or

 * to extend the choice of license to its licensees as provided above.

 * However, if you add GPL Version 2 code and therefore, elected the GPL

 * Version 2 license, then the option applies only if the new code is

 * made subject to such option by the copyright holder.

 *

 * Contributor(s):

 *

 * Portions Copyrighted 2010 Sun Microsystems, Inc.

 */

package org.openide.util;

import java.util.ArrayList;

import java.util.Collections;

import java.util.HashSet;

import java.util.List;

import java.util.Set;

import java.util.concurrent.Callable;

import java.util.concurrent.CancellationException;

import java.util.concurrent.CopyOnWriteArrayList;

import java.util.concurrent.CountDownLatch;

import java.util.concurrent.Future;

import java.util.concurrent.ScheduledFuture;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.logging.Level;

import junit.framework.Test;

import org.netbeans.junit.NbTestCase;

import org.netbeans.junit.NbTestSuite;

import org.netbeans.junit.RandomlyFails;

/**

 *

 * @author Tim Boudreau

 */

public class RequestProcessor180386Test extends NbTestCase {

    public RequestProcessor180386Test(java.lang.String testName) {

        super(testName);

    }

    public static Test suite() {

        Test t = null;

        if (t == null) {

            t = new NbTestSuite(RequestProcessor180386Test.class);

        }

        return t;

    }

    static {

        RequestProcessor.logger().setLevel(Level.ALL);

    }

    public void testSubmit() throws Exception {

        class C implements Callable<String> {

            volatile boolean hasRun;

            @Override

            public String call() throws Exception {

                String result = "Hello";

                hasRun = true;

                return result;

            }

        }

        C c = new C();

        Future<String> f = RequestProcessor.getDefault().submit(c);

        assertEquals("Hello", f.get());

        assertTrue(c.hasRun);

        class R implements Runnable {

            volatile boolean hasRun;

            @Override

            public void run() {

                hasRun = true;

            }

        }

        R r = new R();

        f = RequestProcessor.getDefault().submit(r, "Goodbye");

        assertEquals("Goodbye", f.get());

        assertTrue(r.hasRun);

    }

    public void testSomeTasksNotRunIfShutDown() throws Exception {

        final Object lock = new Object();

        int count = 10;

        final CountDownLatch waitAllLaunched = new CountDownLatch(count);

        final CountDownLatch waitOneFinished = new CountDownLatch(1);

        RequestProcessor rp = new RequestProcessor("TestRP", count * 2);

        class R implements Runnable {

            volatile boolean hasStarted;

            volatile boolean hasFinished;

            @Override

            public void run() {

                hasStarted = true;

                waitAllLaunched.countDown();

                synchronized (lock) {

                    try {

                        lock.wait();

                        if (Thread.interrupted()) {

                            return;

                        }

                    } catch (InterruptedException ex) {

                        return;

                    } finally {

                        waitOneFinished.countDown();

                    }

                    hasFinished = true;

                }

            }

        }

        Set<Future<String>> s = new HashSet<Future<String>>();

        Set<R> rs = new HashSet<R>();

        for (int i = 0; i < count; i++) {

            String currName = "Runnable " + i;

            R r = new R();

            rs.add(r);

            s.add(rp.submit(r, currName));

        }

        waitAllLaunched.await();

        synchronized (lock) {

            //Notify just one thread

            lock.notify();

        }

        waitOneFinished.await();

        List<Runnable> notRun = rp.shutdownNow();

        synchronized (lock) {

            lock.notifyAll();

        }

        boolean allFinished = true;

        int finishedCount = 0;

        for (R r : rs) {

            assertTrue(r.hasStarted);

            allFinished &= r.hasFinished;

            if (r.hasFinished) {

                finishedCount++;

            }

        }

        assertFalse("All tasks should not have completed", allFinished);

        assertTrue("At least one task shall complete", finishedCount >= 1);

        assertFalse(notRun.isEmpty());

        assertTrue(rp.isShutdown());

        //Technically not provable due to "spurious wakeups"

        //        assertEquals (1, finishedCount);

        try {

            RequestProcessor.getDefault().shutdown();

            fail("Should not be able to shutdown() default RP");

        } catch (Exception e) {

        }

        try {

            RequestProcessor.getDefault().shutdownNow();

            fail("Should not be able to shutdownNow() default RP");

        } catch (Exception e) {

        }

        for (Runnable nr : notRun) {

            assertTrue ("Shutdown is not returning submitted runnables - got a "

                    + nr.getClass().getName() + " instead of " +

                    R.class.getName(), nr.getClass() == R.class);

        }

    }

    public void testAwaitTermination() throws Exception {

        int count = 20;

        final Object lock = new Object();

        final CountDownLatch waitAllLaunched = new CountDownLatch(count);

        final CountDownLatch waitAll = new CountDownLatch(count);

        final RequestProcessor rp = new RequestProcessor("TestRP", count);

        class R implements Runnable {

            volatile boolean hasStarted;

            volatile boolean hasFinished;

            @Override

            public void run() {

                hasStarted = true;

                waitAllLaunched.countDown();

                synchronized (lock) {

                    try {

                        lock.wait();

                        if (Thread.interrupted()) {

                            return;

                        }

                    } catch (InterruptedException ex) {

                        return;

                    } finally {

                        hasFinished = true;

                        waitAll.countDown();

                    }

                }

            }

        }

        Set<Future<String>> s = new HashSet<Future<String>>();

        Set<R> rs = new HashSet<R>();

        for (int i = 0; i < count; i++) {

            String currName = "Runnable " + i;

            R r = new R();

            rs.add(r);

            s.add(rp.submit(r, currName));

        }

        waitAllLaunched.await();

        synchronized (lock) {

            //Notify just one thread

            lock.notifyAll();

        }

        rp.shutdown();

        boolean awaitTermination = rp.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);

        assertTrue(awaitTermination);

        assertTrue(rp.isShutdown());

        assertTrue(rp.isTerminated());

    }

    @RandomlyFails

    public void testAwaitTerminationWaitsForNewlyAddedThreads() throws Exception {

        final RequestProcessor rp = new RequestProcessor("testAwaitTerminationWaitsForNewlyAddedThreads", 50, false);

        int count = 30;

        final CountDownLatch waitLock = new CountDownLatch(1);

        class R implements Runnable {

            boolean done;

            @Override

            public void run() {

                try {

                    waitLock.await();

                } catch (InterruptedException ex) {

                    done = true;

                } finally {

                    done = true;

                }

            }

        }

        Set<R> rs = new HashSet<R>();

        for (int i= 0; i < count; i++) {

            R r = new R();

            rs.add(r);

            rp.submit(r);

        }

        final CountDownLatch shutdownBegun = new CountDownLatch(1);

        Runnable shutdowner = new Runnable() {

            @Override

            public void run() {

                try {

                    Thread.sleep(1000);

                    rp.shutdown();

                    shutdownBegun.countDown();

                } catch (InterruptedException ex) {

                    Exceptions.printStackTrace(ex);

                }

            }

        };

        waitLock.countDown();

        new Thread(shutdowner).start();

        assertTrue(rp.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS));

        Thread.sleep (600);

        assertTrue (rp.isTerminated());

    }

    public void testInvokeAll() throws Exception {

        int count = 20;

        final CountDownLatch waitAll = new CountDownLatch(count);

        final RequestProcessor rp = new RequestProcessor("TestRP", count);

        try {

            class C implements Callable<String> {

                private final String result;

                volatile boolean ran;

                C(String result) {

                    this.result = result;

                }

                @Override

                public String call() throws Exception {

                    ran = true;

                    waitAll.countDown();

                    return result;

                }

            }

            List<C> callables = new ArrayList<C>(count);

            List<Future<String>> fs;

            Set<String> names = new HashSet<String>(count);

            for (int i = 0; i < count; i++) {

                String name = "R" + i;

                names.add(name);

                C c = new C(name);

                callables.add(c);

            }

            fs = rp.invokeAll(callables);

            assertNotNull(fs);

            assertEquals(0, waitAll.getCount());

            for (Future<String> f : fs) {

                assertTrue (f.isDone());

            }

            for (C c : callables) {

                assertTrue (c.ran);

            }

            Set<String> s = new HashSet<String>(count);

            for (Future<String> f : fs) {

                s.add(f.get());

            }

            assertEquals(names, s);

        } finally {

            rp.stop();

        }

    }

    public void testInvokeAllWithTimeout() throws Exception {

        int count = 20;

        final CountDownLatch blocker = new CountDownLatch(1);

        final RequestProcessor rp = new RequestProcessor("TestRP", count);

        try {

            class C implements Callable<String> {

                volatile boolean iAmSpecial;

                private final String result;

                volatile boolean ran;

                C(String result) {

                    this.result = result;

                }

                @Override

                public String call() throws Exception {

                    //Only one will be allowed to run, the rest

                    //will be cancelled

                    if (!iAmSpecial) {

                        blocker.await();

                    }

                    ran = true;

                    return result;

                }

            }

            List<C> callables = new ArrayList<C>(count);

            C special = new C("Special");

            special.iAmSpecial = true;

            callables.add(special);

            List<Future<String>> fs;

            Set<String> names = new HashSet<String>(count);

            for (int i = 0; i < count; i++) {

                String name = "R" + i;

                names.add(name);

                C c = new C(name);

                callables.add(c);

            }

            fs = rp.invokeAll(callables, 1000, TimeUnit.MILLISECONDS);

            assertNotNull(fs);

            for (Future<String> f : fs) {

                assertTrue (f.isDone());

            }

            for (C c : callables) {

                if (c == special) {

                    assertTrue (c.ran);

                } else {

                    assertFalse(c.ran);

                }

            }

        } finally {

            rp.stop();

        }

    }

    public void testInvokeAllSingleThread() throws Exception {

        int count = 20;

        final CountDownLatch waitAll = new CountDownLatch(count);

        final RequestProcessor rp = new RequestProcessor("TestRP", 1);

        class C implements Callable<String> {

            private final String result;

            C(String result) {

                this.result = result;

            }

            @Override

            public String call() throws Exception {

                waitAll.countDown();

                return result;

            }

        }

        List<C> l = new ArrayList<C>(count);

        List<Future<String>> fs;

        Set<String> names = new HashSet<String>(count);

        for (int i = 0; i < count; i++) {

            String name = "R" + i;

            names.add(name);

            C c = new C(name);

            l.add(c);

        }

        fs = rp.invokeAll(l);

        assertNotNull(fs);

        Set<String> s = new HashSet<String>(count);

        for (Future<String> f : fs) {

            s.add(f.get());

        }

        assertEquals(names, s);

    }

    public void testInvokeAny() throws Exception {

        int count = 20;

        final RequestProcessor rp = new RequestProcessor("TestRP", count + 1);

        class C implements Callable<String> {

            private final String result;

            C(String result) {

                this.result = result;

            }

            @Override

            public String call() throws Exception {

                if (Thread.interrupted()) {

                    return null;

                }

                return result;

            }

        }

        List<C> l = new ArrayList<C>(count);

        Set<String> names = new HashSet<String>(count);

        for (int i = 0; i < count; i++) {

            String name = "R" + i;

            names.add(name);

            C c = new C(name);

            l.add(c);

        }

        String res = rp.invokeAny(l);

        assertNotNull(res);

        assertTrue(res.startsWith("R"));

    }

    public void testInvokeAnySingleThread() throws Exception {

        int count = 1000;

        final RequestProcessor rp = new RequestProcessor("TestRP", 20);

        final CountDownLatch latch = new CountDownLatch(count);

        class C implements Callable<String> {

            volatile boolean hasRun;

            private final String result;

            C(String result) {

                this.result = result;

            }

            @Override

            public String call() throws Exception {

                latch.countDown();

                if (!"R17".equals(result)) {

                    try {

                        //Block all but one thread until threads have entered

                        latch.await();

                    } catch (InterruptedException ie) {

                    }

                }

                Thread.yield();

                hasRun = true;

                return result;

            }

        }

        List<C> l = new ArrayList<C>(count);

        Set<String> names = new HashSet<String>(count);

        for (int i = 0; i < count; i++) {

            String name = "R" + i;

            names.add(name);

            C c = new C(name);

            l.add(c);

        }

        String res = rp.invokeAny(l);

        assertNotNull(res);

        assertTrue(res.startsWith("R"));

        int runCount = 0;

        for (C c : l) {

            if (c.hasRun) {

                runCount++;

            }

        }

        assertTrue("Not all " + count + " threads should have completed, but " + runCount + " did.", runCount < count);

    }

    public void testInvokeAnyWithTimeout() throws Exception {

        int count = 20;

        final RequestProcessor rp = new RequestProcessor("TestRP", count + 1);

        final CountDownLatch latch = new CountDownLatch(1);

        class C implements Callable<String> {

            volatile boolean hasRun;

            private final String result;

            C(String result) {

                this.result = result;

            }

            @Override

            public String call() throws Exception {

                latch.await();

                if (Thread.interrupted()) {

                    return null;

                }

                hasRun = true;

                return result;

            }

        }

        List<C> l = new ArrayList<C>(count);

        Set<String> names = new HashSet<String>(count);

        for (int i = 0; i < count; i++) {

            String name = "R" + i;

            names.add(name);

            C c = new C(name);

            l.add(c);

        }

        //All threads are waiting on latch;  we should time out

        String res = rp.invokeAny(l, 400, TimeUnit.MILLISECONDS);

        assertNull(res);

        for (C c : l) {

            assertFalse(c.hasRun);

        }

    }

    public void testCancellation() throws Exception {

        final CountDownLatch latch = new CountDownLatch(1);

        class C implements Callable<String> {

            volatile boolean hasRun;

            volatile boolean interrupted;

            @Override

            public String call() throws Exception {

                try {

                    latch.await();

                } catch (InterruptedException e) {

                    interrupted = true;

                    return null;

                }

                if (Thread.interrupted()) {

                    interrupted = true;

                    return null;

                }

                hasRun = true;

                return "Hello";

            }

        }

        C c = new C();

        Future<String> f = RequestProcessor.getDefault().submit(c);

        f.cancel(true);

        latch.countDown();

        String s = null;

        try {

            s = f.get();

            fail("CancellationException should have been thrown");

        } catch (CancellationException e) {

        }

        assertNull(s);

        assertTrue(c.interrupted || !c.hasRun);

        assertFalse(c.hasRun);

    }

    public void testCancellablesGetCancelInvokedWithCallable() throws Exception {

        final CountDownLatch latch = new CountDownLatch(1);

        final CountDownLatch exit = new CountDownLatch(1);

        class C implements Callable<String>, Cancellable {

            volatile boolean hasRun;

            volatile boolean interrupted;

            volatile boolean cancelled;

            @Override

            public String call() throws Exception {

                try {

                    try {

                        latch.await();

                    } catch (InterruptedException e) {

                        interrupted = true;

                        return null;

                    }

                    if (Thread.interrupted()) {

                        interrupted = true;

                        return null;

                    }

                    if (cancelled) {

                        return null;

                    }

                    hasRun = true;

                    return "Hello";

                } finally {

                    exit.countDown();

                }

            }

            @Override

            public boolean cancel() {

                cancelled = true;

                exit.countDown();

                return true;

            }

        }

        C c = new C();

        Future<String> f = RequestProcessor.getDefault().submit(c);

        f.cancel(true);

        assertTrue (c.cancelled);

        latch.countDown();

        exit.await();

        String s = null;

        try {

            s = f.get();

            fail ("Should have gotten cancellation exception");

        } catch (CancellationException e) {

        }

    }

    public void testCancellablesGetCancelInvokedWithRunnable() throws Exception {

        final CountDownLatch latch = new CountDownLatch(1);

        final CountDownLatch exit = new CountDownLatch(1);

        class C implements Runnable, Cancellable {

            volatile boolean hasRun;

            volatile boolean interrupted;

            volatile boolean cancelled;

            @Override

            public void run() {

                try {

                    try {

                        latch.await();

                    } catch (InterruptedException e) {

                        interrupted = true;

                        return;

                    }

                    if (Thread.interrupted()) {

                        interrupted = true;

                        return;

                    }

                    if (cancelled) {

                        return;

                    }

                    hasRun = true;

                } finally {

                    exit.countDown();

                }

            }

            @Override

            public boolean cancel() {

                cancelled = true;

                exit.countDown();

                return true;

            }

        }

        C c = new C();

        Future<?> f = RequestProcessor.getDefault().submit(c);

        f.cancel(true);

        assertTrue (c.cancelled);

        latch.countDown();

        exit.await();

        try {

            f.get();

            fail ("Should have gotten cancellation exception");

        } catch (CancellationException e) {

        }

        assertFalse (c.hasRun);

    }

    public void testCancellablesThatSayTheyCantBeCancelledAreNotCancelledViaFutureDotCancel() throws Exception {

        final CountDownLatch latch = new CountDownLatch(1);

        final CountDownLatch exit = new CountDownLatch(1);

        class C implements Runnable, Cancellable {

            volatile boolean hasRun;

            volatile boolean interrupted;

            volatile boolean cancelCalled;

            @Override

            public void run() {

                try {

                    try {

                        latch.await();

                    } catch (InterruptedException e) {

                        interrupted = true;

                        throw new AssertionError(e);

                    }

                    if (Thread.interrupted()) {

                        interrupted = true;

                        throw new AssertionError("Thread should not have been interrupted");

                    }

                    hasRun = true;

                } finally {

                    exit.countDown();

                }

            }

            @Override

            public boolean cancel() {

                cancelCalled = true;

                return false;

            }

        }

        C c = new C();

        Future<?> f = RequestProcessor.getDefault().submit(c);

        f.cancel(true);

        assertFalse (f.isCancelled());

        assertTrue (c.cancelCalled);

        latch.countDown();

        exit.await();

        f.get();

        assertFalse (f.isCancelled());

        assertTrue (c.hasRun);

    }

    public void testInvokeAllCancellation() throws Exception {

        int count = 20;

        final CountDownLatch waitAll = new CountDownLatch(count);

        final RequestProcessor rp = new RequestProcessor("TestRP", count);

        class C implements Callable<String>, Cancellable {

            private final String result;

            volatile boolean cancelCalled;

            C(String result) {

                this.result = result;

            }

            @Override

            public String call() throws Exception {

                waitAll.countDown();

                return cancelCalled ? null : result;

            }

            @Override

            public boolean cancel() {

                cancelCalled = true;

                return false;

            }

        }

        List<C> l = new ArrayList<C>(count);

        List<Future<String>> fs;

        Set<String> names = new HashSet<String>(count);

        for (int i = 0; i < count; i++) {

            String name = "R" + i;

            names.add(name);

            C c = new C(name);

            l.add(c);

        }

        fs = rp.invokeAll(l);

        assertNotNull(fs);

        Set<String> s = new HashSet<String>(count);

        for (Future<String> f : fs) {

            s.add(f.get());

        }

        assertEquals(names, s);

    }

    public void testCannotScheduleLongerThanIntegerMaxValue() throws Exception {

        Runnable r = new Runnable() {

            @Override

            public void run() {

                fail ("Should not have been run");

            }

        };

        try {

            Future<?> f = RequestProcessor.getDefault().schedule(r, Long.MAX_VALUE, TimeUnit.DAYS);

            f.cancel(true);

        } catch (Exception e) {}

    }

    public void testCannotScheduleNegativeDelay() throws Exception {

        Runnable r = new Runnable() {

            @Override

            public void run() {

                fail ("Should not have been run");

            }

        };

        try {

            RequestProcessor.getDefault().schedule(r, -1L, TimeUnit.MILLISECONDS);

            fail ("Negative value accepetd");

        } catch (Exception e) {}

        try {

            RequestProcessor.getDefault().scheduleAtFixedRate(r, -1L, 22L, TimeUnit.MILLISECONDS);

            fail ("Negative value accepetd");

        } catch (Exception e) {}

        try {

            RequestProcessor.getDefault().scheduleAtFixedRate(r, 200, -22L, TimeUnit.MILLISECONDS);

            fail ("Negative value accepetd");

        } catch (Exception e) {}

        try {

            RequestProcessor.getDefault().scheduleWithFixedDelay(r, -1L, 22L, TimeUnit.MILLISECONDS);

            fail ("Negative value accepetd");

        } catch (Exception e) {}

        try {

            RequestProcessor.getDefault().scheduleWithFixedDelay(r, 1L, -22L, TimeUnit.MILLISECONDS);

            fail ("Negative value accepetd");

        } catch (Exception e) {}

    }

    public void testTaskCanRescheduleItself() throws Exception {

        final CountDownLatch latch = new CountDownLatch(2);

        class R implements Runnable {

            volatile RequestProcessor.Task task;

            volatile int runCount;

            @Override

            public void run() {

                runCount++;

                if (runCount == 1) {

                    task.schedule(0);

                }

                latch.countDown();

            }

        }

        R r = new R();

        RequestProcessor.Task t = RequestProcessor.getDefault().create(r);

        r.task = t;

        t.schedule(0);

        latch.await ();

        assertEquals (r.runCount, 2);

    }

    public void testScheduleRepeatingSanityFixedRate() throws Exception {

        final CountDownLatch latch = new CountDownLatch(5);

        class C implements Runnable {

            volatile int runCount;

            @Override

            public void run() {

                runCount++;

                latch.countDown();

            }

        }

        C c = new C();

        RequestProcessor.getDefault().scheduleWithFixedDelay(c, 0, 20, TimeUnit.MILLISECONDS);

//        latch.await(5000, TimeUnit.MILLISECONDS);

        latch.await();

        assertEquals (5, c.runCount);

    }

    public void testScheduleRepeatingSanityFixedDelay() throws Exception {

        final CountDownLatch latch = new CountDownLatch(5);

        class C implements Runnable {

            volatile int runCount;

            @Override

            public void run() {

                runCount++;

                latch.countDown();

            }

        }

        C c = new C();

        RequestProcessor.getDefault().scheduleAtFixedRate(c, 0, 20, TimeUnit.MILLISECONDS);

        latch.await(2000, TimeUnit.MILLISECONDS);

        assertEquals (5, c.runCount);

    }

    public void testScheduleOneShot() throws Exception {

        RequestProcessor rp = new RequestProcessor ("testScheduleOneShot", 5, true, true);

        try {

            class C implements Callable<String> {

                volatile long start = System.currentTimeMillis();

                private volatile long end;

                @Override

                public String call() throws Exception {

                    synchronized(this) {

                        end = System.currentTimeMillis();

                    }

                    return "Hello";

                }

                synchronized long elapsed() {

                    return end - start;

                }

            }

            C c = new C();

            int delay = 5000;

            //Use a 20 second timeout to have a reasonable chance of accuracy

            ScheduledFuture<String> f = rp.schedule(c, delay, TimeUnit.MILLISECONDS);

            assertEquals (5000, f.getDelay(TimeUnit.MILLISECONDS));

            assertNotNull(f.get());

            //Allow 4 seconds fudge-factor

            assertTrue (c.elapsed() > 4600);

            assertTrue (f.isDone());

        } finally {

            rp.stop();

        }

    }

    public void testScheduleRepeatingIntervalsAreRoughlyCorrect() throws Exception {

        int runCount = 5;

        final CountDownLatch latch = new CountDownLatch(runCount);

        final List<Long> intervals = Collections.synchronizedList(new ArrayList<Long> (runCount));

//        long initialDelay = 30000;

//        long period = 20000;

//        long fudgeFactor = 4000;

        long initialDelay = 3000;

        long period = 2000;

        long fudgeFactor = 400;

        long expectedInitialDelay = initialDelay - fudgeFactor;

        long expectedPeriod = period - fudgeFactor;

        class C implements Runnable {

            volatile long start = System.currentTimeMillis();

            private int runCount;

            @Override

            public void run() {

                runCount++;

                try {

                    synchronized(this) {

                        long end = System.currentTimeMillis();

                        intervals.add (end - start);

                        start = end;

                    }

                } finally {

                    latch.countDown();

                }

            }

        }

        C c = new C();

        RequestProcessor rp = new RequestProcessor ("testScheduleRepeating", 5, true);

        try {

            Future<?> f = rp.scheduleWithFixedDelay(c, initialDelay, period, TimeUnit.MILLISECONDS);

    //        latch.await(initialDelay + fudgeFactor + ((runCount - 1) * (period + fudgeFactor)), TimeUnit.MILLISECONDS); //XXX

            latch.await();

            f.cancel(true);

            for (int i= 0; i < Math.min(runCount, intervals.size()); i++) {

                long expect = i == 0 ? expectedInitialDelay : expectedPeriod;

                assertTrue ("Expected at least " + expect + " milliseconds before run " + i + " but was " + intervals.get(i), intervals.get(i) >= expect);

            }

            //Ensure we have really exited

            try {

                f.get();

                fail ("CancellationException should have been thrown");

            } catch (CancellationException e) {}

            assertTrue(f.isCancelled());

            assertTrue(f.isDone());

        } finally {

            rp.stop();

        }

    }

    public void testScheduleFixedRateAreRoughlyCorrect() throws Exception {

        int runCount = 5;

        final CountDownLatch latch = new CountDownLatch(runCount);

        final List<Long> intervals = Collections.synchronizedList(new ArrayList<Long> (runCount));