public interface CompletionService<V> {
    /**
     * Submits a value-returning task for execution and returns a Future
     * representing the pending results of the task.  Upon completion,
     * this task may be taken or polled.
     *
     * @param task the task to submit
     * @return a Future representing pending completion of the task
     * @throws RejectedExecutionException if the task cannot be
     *         scheduled for execution
     * @throws NullPointerException if the task is null
     */
    // 1
    Future<V> submit(Callable<V> task);

    /**
     * Submits a Runnable task for execution and returns a Future
     * representing that task.  Upon completion, this task may be
     * taken or polled.
     *
     * @param task the task to submit
     * @param result the result to return upon successful completion
     * @return a Future representing pending completion of the task,
     *         and whose <tt>get()</tt> method will return the given
     *         result value upon completion
     * @throws RejectedExecutionException if the task cannot be
     *         scheduled for execution
     * @throws NullPointerException if the task is null
     */
    // 2
    Future<V> submit(Runnable task, V result);

    /**
     * Retrieves and removes the Future representing the next
     * completed task, waiting if none are yet present.
     *
     * @return the Future representing the next completed task
     * @throws InterruptedException if interrupted while waiting
     */
    // 3
    Future<V> take() throws InterruptedException;


    /**
     * Retrieves and removes the Future representing the next
     * completed task or <tt>null</tt> if none are present.
     *
     * @return the Future representing the next completed task, or
     *         <tt>null</tt> if none are present
     */
    // 4
    Future<V> poll();

    /**
     * Retrieves and removes the Future representing the next
     * completed task, waiting if necessary up to the specified wait
     * time if none are yet present.
     *
     * @param timeout how long to wait before giving up, in units of
     *        <tt>unit</tt>
     * @param unit a <tt>TimeUnit</tt> determining how to interpret the
     *        <tt>timeout</tt> parameter
     * @return the Future representing the next completed task or
     *         <tt>null</tt> if the specified waiting time elapses
     *         before one is present
     * @throws InterruptedException if interrupted while waiting
     */
    // 5
    Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;
}

public class ExecutorCompletionService<V> implements CompletionService<V> {
    private final Executor executor;
    private final AbstractExecutorService aes;
    private final BlockingQueue<Future<V>> completionQueue;

    /**
     * FutureTask extension to enqueue upon completion
     */
    private class QueueingFuture extends FutureTask<Void> {
        QueueingFuture(RunnableFuture<V> task) {
            super(task, null);
            this.task = task;
        }
        // 1
        protected void done() { completionQueue.add(task); }
        private final Future<V> task;
    }

/**
    * Creates an ExecutorCompletionService using the supplied
    * executor for base task execution and a
    * {@link LinkedBlockingQueue} as a completion queue.
    *
    * @param executor the executor to use
    * @throws NullPointerException if executor is <tt>null</tt>
    */
   public ExecutorCompletionService(Executor executor) {
       if (executor == null)
           throw new NullPointerException();
       this.executor = executor;
       this.aes = (executor instanceof AbstractExecutorService) ?
           (AbstractExecutorService) executor : null;
       this.completionQueue = new LinkedBlockingQueue<Future<V>>();
   }

   /**
    * Creates an ExecutorCompletionService using the supplied
    * executor for base task execution and the supplied queue as its
    * completion queue.
    *
    * @param executor the executor to use
    * @param completionQueue the queue to use as the completion queue
    * normally one dedicated for use by this service
    * @throws NullPointerException if executor or completionQueue are <tt>null</tt>
    */
   public ExecutorCompletionService(Executor executor,
                                    BlockingQueue<Future<V>> completionQueue) {
       if (executor == null || completionQueue == null)
           throw new NullPointerException();
       this.executor = executor;
       // 1
       this.aes = (executor instanceof AbstractExecutorService) ?
           (AbstractExecutorService) executor : null;
       this.completionQueue = completionQueue;
   }

private RunnableFuture<V> newTaskFor(Callable<V> task) {
    if (aes == null)
        return new FutureTask<V>(task);
    else
        return aes.newTaskFor(task);
}

private RunnableFuture<V> newTaskFor(Runnable task, V result) {
    if (aes == null)
        return new FutureTask<V>(task, result);
    else
        return aes.newTaskFor(task, result);
}

/**
 * Returns a <tt>RunnableFuture</tt> for the given callable task.
 *
 * @param callable the callable task being wrapped
 * @return a <tt>RunnableFuture</tt> which when run will call the
 * underlying callable and which, as a <tt>Future</tt>, will yield
 * the callable's result as its result and provide for
 * cancellation of the underlying task.
 * @since 1.6
 */
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    return new FutureTask<T>(callable);
}

public Future<V> submit(Callable<V> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<V> f = newTaskFor(task);
    executor.execute(new QueueingFuture(f));
    return f;
}

public Future<V> submit(Runnable task, V result) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<V> f = newTaskFor(task, result);
    executor.execute(new QueueingFuture(f));
    return f;
}

public interface Executor {

    /**
     * Executes the given command at some time in the future.  The command
     * may execute in a new thread, in a pooled thread, or in the calling
     * thread, at the discretion of the <tt>Executor</tt> implementation.
     *
     * @param command the runnable task
     * @throws RejectedExecutionException if this task cannot be
     * accepted for execution.
     * @throws NullPointerException if command is null
     */
    void execute(Runnable command);
}

/**
 * Executes command with zero required delay. This has effect
 * equivalent to <tt>schedule(command, 0, anyUnit)</tt>.  Note
 * that inspections of the queue and of the list returned by
 * <tt>shutdownNow</tt> will access the zero-delayed
 * {@link ScheduledFuture}, not the <tt>command</tt> itself.
 *
 * @param command the task to execute
 * @throws RejectedExecutionException at discretion of
 * <tt>RejectedExecutionHandler</tt>, if task cannot be accepted
 * for execution because the executor has been shut down.
 * @throws NullPointerException if command is null
 */
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    schedule(command, 0, TimeUnit.NANOSECONDS);
}

/**
 * Protected method invoked when this task transitions to state
 * <tt>isDone</tt> (whether normally or via cancellation). The
 * default implementation does nothing.  Subclasses may override
 * this method to invoke completion callbacks or perform
 * bookkeeping. Note that you can query status inside the
 * implementation of this method to determine whether this task
 * has been cancelled.
 */
protected void done() { }

/**
 * FutureTask extension to enqueue upon completion
 */
private class QueueingFuture extends FutureTask<Void> {
    QueueingFuture(RunnableFuture<V> task) {
        super(task, null);
        this.task = task;
    }
    protected void done() { completionQueue.add(task); }
    private final Future<V> task;
}

public Future<V> take() throws InterruptedException {
    return completionQueue.take();
}

public Future<V> poll() {
    return completionQueue.poll();
}

public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException {
    return completionQueue.poll(timeout, unit);
}