content: /*
   * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   *
   * This code is free software; you can redistribute it and/or modify it
   * under the terms of the GNU General Public License version 2 only, as
   * published by the Free Software Foundation.  Oracle designates this
   * particular file as subject to the "Classpath" exception as provided
   * by Oracle in the LICENSE file that accompanied this code.
   *
   * This code is distributed in the hope that it will be useful, but WITHOUT
   * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
   * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
   * version 2 for more details (a copy is included in the LICENSE file that
   * accompanied this code).
   *
   * You should have received a copy of the GNU General Public License version
   * 2 along with this work; if not, write to the Free Software Foundation,
   * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
   *
   * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
   * or visit www.oracle.com if you need additional information or have any
   * questions.
   */
  
  /*
   * This file is available under and governed by the GNU General Public
   * License version 2 only, as published by the Free Software Foundation.
   * However, the following notice accompanied the original version of this
   * file:
   *
   * Written by Doug Lea with assistance from members of JCP JSR-166
   * Expert Group and released to the public domain, as explained at
   * http://creativecommons.org/publicdomain/zero/1.0/
   */
  
  package java.util.concurrent;
  
  /**
   * A {@link CompletionService} that uses a supplied {@link Executor}
   * to execute tasks.  This class arranges that submitted tasks are,
   * upon completion, placed on a queue accessible using {@code take}.
   * The class is lightweight enough to be suitable for transient use
   * when processing groups of tasks.
   *
   * <p>
   *
   * <b>Usage Examples.</b>
   *
   * Suppose you have a set of solvers for a certain problem, each
   * returning a value of some type {@code Result}, and would like to
   * run them concurrently, processing the results of each of them that
   * return a non-null value, in some method {@code use(Result r)}. You
   * could write this as:
   *
   * <pre> {@code
   * void solve(Executor e,
   *            Collection<Callable<Result>> solvers)
   *     throws InterruptedException, ExecutionException {
   *   CompletionService<Result> ecs
   *       = new ExecutorCompletionService<Result>(e);
   *   for (Callable<Result> s : solvers)
   *     ecs.submit(s);
   *   int n = solvers.size();
   *   for (int i = 0; i < n; ++i) {
   *     Result r = ecs.take().get();
   *     if (r != null)
   *       use(r);
   *   }
   * }}</pre>
   *
   * Suppose instead that you would like to use the first non-null result
   * of the set of tasks, ignoring any that encounter exceptions,
   * and cancelling all other tasks when the first one is ready:
   *
   * <pre> {@code
   * void solve(Executor e,
   *            Collection<Callable<Result>> solvers)
   *     throws InterruptedException {
   *   CompletionService<Result> ecs
   *       = new ExecutorCompletionService<Result>(e);
   *   int n = solvers.size();
   *   List<Future<Result>> futures = new ArrayList<>(n);
   *   Result result = null;
   *   try {
   *     for (Callable<Result> s : solvers)
   *       futures.add(ecs.submit(s));
   *     for (int i = 0; i < n; ++i) {
   *       try {
   *         Result r = ecs.take().get();
   *         if (r != null) {
   *           result = r;
   *           break;
   *         }
   *       } catch (ExecutionException ignore) {}
   *     }
   *   }
   *   finally {
   *     for (Future<Result> f : futures)
   *       f.cancel(true);
   *   }
   *
   *   if (result != null)
   *     use(result);
   * }}</pre>
   */
  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 static class QueueingFuture<V> extends FutureTask<Void> {
          QueueingFuture(RunnableFuture<V> task,
                         BlockingQueue<Future<V>> completionQueue) {
              super(task, null);
              this.task = task;
              this.completionQueue = completionQueue;
          }
          private final Future<V> task;
          private final BlockingQueue<Future<V>> completionQueue;
          protected void done() { completionQueue.add(task); }
      }
  
      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);
      }
  
      /**
       * 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 {@code null}
       */
      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. This
       *        queue is treated as unbounded -- failed attempted
       *        {@code Queue.add} operations for completed tasks cause
       *        them not to be retrievable.
       * @throws NullPointerException if executor or completionQueue are {@code null}
       */
      public ExecutorCompletionService(Executor executor,
                                       BlockingQueue<Future<V>> completionQueue) {
          if (executor == null || completionQueue == null)
              throw new NullPointerException();
          this.executor = executor;
          this.aes = (executor instanceof AbstractExecutorService) ?
              (AbstractExecutorService) executor : null;
          this.completionQueue = completionQueue;
      }
  
      public Future<V> submit(Callable<V> task) {
          if (task == null) throw new NullPointerException();
          RunnableFuture<V> f = newTaskFor(task);
          executor.execute(new QueueingFuture<V>(f, completionQueue));
          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<V>(f, completionQueue));
          return f;
      }
  
      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);
      }
  
  }
  