src/com/google/common/util/concurrent/Futures.java - platform/external/guava - Git at Google

 /*
  * Copyright (C) 2006 Google Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package com.google.common.util.concurrent;

 import com.google.common.base.Function;

 import java.lang.reflect.UndeclaredThrowableException;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;

 import javax.annotation.Nullable;

 import static com.google.common.base.Preconditions.checkNotNull;

 /**
  * Static utility methods pertaining to the {@link Future} interface.
  *
  * @author Kevin Bourrillion
  * @author Nishant Thakkar
  * @author Sven Mawson
  * @since 2009.09.15 <b>tentative</b>
  */
 public class Futures {
   private Futures() {}

   /**
    * Returns an uninterruptible view of a {@code Future}. If a thread is
    * interrupted during an attempt to {@code get()} from the returned future, it
    * continues to wait on the result until it is available or the timeout
    * elapses, and only then re-interrupts the thread.
    */
   public static <V> UninterruptibleFuture<V> makeUninterruptible(
       final Future<V> future) {
     checkNotNull(future);
     if (future instanceof UninterruptibleFuture) {
       return (UninterruptibleFuture<V>) future;
     }
     return new UninterruptibleFuture<V>() {
       public boolean cancel(boolean mayInterruptIfRunning) {
         return future.cancel(mayInterruptIfRunning);
       }
       public boolean isCancelled() {
         return future.isCancelled();
       }
       public boolean isDone() {
         return future.isDone();
       }

       public V get(long timeoutDuration, TimeUnit timeoutUnit)
           throws TimeoutException, ExecutionException {
         boolean interrupted = false;
         try {
           long timeoutNanos = timeoutUnit.toNanos(timeoutDuration);
           long end = System.nanoTime() + timeoutNanos;
           for (long remaining = timeoutNanos; remaining > 0;
               remaining = end - System.nanoTime()) {
             try {
               return future.get(remaining, TimeUnit.NANOSECONDS);
             } catch (InterruptedException ignored) {
               interrupted = true;
             }
           }
           throw new TimeoutException();
         } finally {
           if (interrupted) {
             Thread.currentThread().interrupt();
           }
         }
       }

       public V get() throws ExecutionException {
         boolean interrupted = false;
         try {
           while (true) {
             try {
               return future.get();
             } catch (InterruptedException ignored) {
               interrupted = true;
             }
           }
         } finally {
           if (interrupted) {
             Thread.currentThread().interrupt();
           }
         }
       }
     };
   }

   /**
    * Creates a {@link ListenableFuture} out of a normal {@link Future}. The
    * returned future will create a thread to wait for the source future to
    * complete before executing the listeners.
    *
    * <p>Callers who have a future that subclasses
    * {@link java.util.concurrent.FutureTask} may want to instead subclass
    * {@link ListenableFutureTask}, which adds the {@link ListenableFuture}
    * functionality to the standard {@code FutureTask} implementation.
    */
   public static <T> ListenableFuture<T> makeListenable(Future<T> future) {
     if (future instanceof ListenableFuture) {
       return (ListenableFuture<T>) future;
     }
     return new ListenableFutureAdapter<T>(future);
   }

   /**
    * Creates a {@link CheckedFuture} out of a normal {@link Future} and a
    * {@link Function} that maps from {@link Exception} instances into the
    * appropriate checked type.
    *
    * <p>The given mapping function will be applied to an
    * {@link InterruptedException}, a {@link CancellationException}, or an
    * {@link ExecutionException} with the actual cause of the exception.
    * See {@link Future#get()} for details on the exceptions thrown.
    */
   public static <T, E extends Exception> CheckedFuture<T, E> makeChecked(
       Future<T> future, Function<Exception, E> mapper) {
     return new MappingCheckedFuture<T, E>(makeListenable(future), mapper);
   }

   /**
    * Creates a {@code ListenableFuture} which has its value set immediately upon
    * construction. The getters just return the value. This {@code Future} can't
    * be canceled or timed out and its {@code isDone()} method always returns
    * {@code true}. It's useful for returning something that implements the
    * {@code ListenableFuture} interface but already has the result.
    */
   public static <T> ListenableFuture<T> immediateFuture(@Nullable T value) {
     ValueFuture<T> future = ValueFuture.create();
     future.set(value);
     return future;
   }

   /**
    * Creates a {@code CheckedFuture} which has its value set immediately upon
    * construction. The getters just return the value. This {@code Future} can't
    * be canceled or timed out and its {@code isDone()} method always returns
    * {@code true}. It's useful for returning something that implements the
    * {@code CheckedFuture} interface but already has the result.
    */
   public static <T, E extends Exception> CheckedFuture<T, E>
       immediateCheckedFuture(@Nullable T value) {
     ValueFuture<T> future = ValueFuture.create();
     future.set(value);
     return Futures.makeChecked(future, new Function<Exception, E>() {
       public E apply(Exception e) {
         throw new AssertionError("impossible");
       }
     });
   }

   /**
    * Creates a {@code ListenableFuture} which has an exception set immediately
    * upon construction. The getters just return the value. This {@code Future}
    * can't be canceled or timed out and its {@code isDone()} method always
    * returns {@code true}. It's useful for returning something that implements
    * the {@code ListenableFuture} interface but already has a failed
    * result. Calling {@code get()} will throw the provided {@code Throwable}
    * (wrapped in an {@code ExecutionException}).
    *
    * @throws Error if the throwable was an {@link Error}.
    */
   public static <T> ListenableFuture<T> immediateFailedFuture(
       Throwable throwable) {
     checkNotNull(throwable);
     ValueFuture<T> future = ValueFuture.create();
     future.setException(throwable);
     return future;
   }

   /**
    * Creates a {@code CheckedFuture} which has an exception set immediately
    * upon construction. The getters just return the value. This {@code Future}
    * can't be canceled or timed out and its {@code isDone()} method always
    * returns {@code true}. It's useful for returning something that implements
    * the {@code CheckedFuture} interface but already has a failed result.
    * Calling {@code get()} will throw the provided {@code Throwable} (wrapped in
    * an {@code ExecutionException}) and calling {@code checkedGet()} will throw
    * the provided exception itself.
    *
    * @throws Error if the throwable was an {@link Error}.
    */
   public static <T, E extends Exception> CheckedFuture<T, E>
       immediateFailedCheckedFuture(final E exception) {
     checkNotNull(exception);
     return makeChecked(Futures.<T>immediateFailedFuture(exception),
         new Function<Exception, E>() {
           public E apply(Exception e) {
             return exception;
           }
         });
   }

   /**
    * Creates a new {@code ListenableFuture} that wraps another
    * {@code ListenableFuture}.  The result of the new future is the result of
    * the provided function called on the result of the provided future.
    * The resulting future doesn't interrupt when aborted.
    *
    * <p>TODO: Add a version that accepts a normal {@code Future}
    *
    * <p>The typical use for this method would be when a RPC call is dependent on
    * the results of another RPC.  One would call the first RPC (input), create a
    * function that calls another RPC based on input's result, and then call
    * chain on input and that function to get a {@code ListenableFuture} of
    * the result.
    *
    * @param input The future to chain
    * @param function A function to chain the results of the provided future
    *     to the results of the returned future.  This will be run in the thread
    *     that notifies input it is complete.
    * @return A future that holds result of the chain.
    */
   public static <I, O> ListenableFuture<O> chain(ListenableFuture<I> input,
       Function<? super I, ? extends ListenableFuture<? extends O>> function) {
     return chain(input, function, Executors.sameThreadExecutor());
   }

   /**
    * Creates a new {@code ListenableFuture} that wraps another
    * {@code ListenableFuture}.  The result of the new future is the result of
    * the provided function called on the result of the provided future.
    * The resulting future doesn't interrupt when aborted.
    *
    * <p>This version allows an arbitrary executor to be passed in for running
    * the chained Function. When using {@link Executors#sameThreadExecutor}, the
    * thread chained Function executes in will be whichever thread set the
    * result of the input Future, which may be the network thread in the case of
    * RPC-based Futures.
    *
    * @param input The future to chain
    * @param function A function to chain the results of the provided future
    *     to the results of the returned future.
    * @param exec Executor to run the function in.
    * @return A future that holds result of the chain.
    */
   public static <I, O> ListenableFuture<O> chain(ListenableFuture<I> input,
       Function<? super I, ? extends ListenableFuture<? extends O>> function,
       Executor exec) {
     ChainingListenableFuture<I, O> chain =
         new ChainingListenableFuture<I, O>(function, input);
     input.addListener(chain, exec);
     return chain;
   }

   /**
    * Creates a new {@code ListenableFuture} that wraps another
    * {@code ListenableFuture}.  The result of the new future is the result of
    * the provided function called on the result of the provided future.
    * The resulting future doesn't interrupt when aborted.
    *
    * <p>An example use of this method is to convert a serializable object
    * returned from an RPC into a POJO.
    *
    * @param future The future to compose
    * @param function A Function to compose the results of the provided future
    *     to the results of the returned future.  This will be run in the thread
    *     that notifies input it is complete.
    * @return A future that holds result of the composition.
    */
   public static <I, O> ListenableFuture<O> compose(ListenableFuture<I> future,
       final Function<? super I, ? extends O> function) {
     return compose(future, function, Executors.sameThreadExecutor());
   }

   /**
    * Creates a new {@code ListenableFuture} that wraps another
    * {@code ListenableFuture}.  The result of the new future is the result of
    * the provided function called on the result of the provided future.
    * The resulting future doesn't interrupt when aborted.
    *
    * <p>An example use of this method is to convert a serializable object
    * returned from an RPC into a POJO.
    *
    * <p>This version allows an arbitrary executor to be passed in for running
    * the chained Function. When using {@link Executors#sameThreadExecutor}, the
    * thread chained Function executes in will be whichever thread set the result
    * of the input Future, which may be the network thread in the case of
    * RPC-based Futures.
    *
    * @param future The future to compose
    * @param function A Function to compose the results of the provided future
    *     to the results of the returned future.
    * @param exec Executor to run the function in.
    * @return A future that holds result of the composition.
    * @since 2010.01.04 <b>tentative</b>
    */
   public static <I, O> ListenableFuture<O> compose(ListenableFuture<I> future,
       final Function<? super I, ? extends O> function, Executor exec) {
     Function<I, ListenableFuture<O>> wrapperFunction
         = new Function<I, ListenableFuture<O>>() {
             /*@Override*/ public ListenableFuture<O> apply(I input) {
               O output = function.apply(input);
               return immediateFuture(output);
             }
         };
     return chain(future, wrapperFunction, exec);
   }

   /**
    * Creates a new {@code Future} that wraps another {@code Future}.
    * The result of the new future is the result of the provided function called
    * on the result of the provided future.
    *
    * <p>An example use of this method is to convert a Future that produces a
    * handle to an object to a future that produces the object itself.
    *
    * <p>Each call to {@code Future<O>.get(*)} results in a call to
    * {@code Future<I>.get(*)}, but {@code function} is only applied once, so it
    * is assumed that {@code Future<I>.get(*)} is idempotent.
    *
    * <p>When calling {@link Future#get(long, TimeUnit)} on the returned
    * future, the timeout only applies to the future passed in to this method.
    * Any additional time taken by applying {@code function} is not considered.
    *
    * @param future The future to compose
    * @param function A Function to compose the results of the provided future
    *     to the results of the returned future.  This will be run in the thread
    *     that calls one of the varieties of {@code get()}.
    * @return A future that computes result of the composition.
    */
   public static <I, O> Future<O> compose(final Future<I> future,
       final Function<? super I, ? extends O> function) {

     return new Future<O>() {

       /*
        * Concurrency detail:
        *
        * <p>To preserve the idempotency of calls to this.get(*) calls to the
        * function are only applied once. A lock is required to prevent multiple
        * applications of the function. The calls to future.get(*) are performed
        * outside the lock, as is required to prevent calls to
        * get(long, TimeUnit) to persist beyond their timeout.
        *
        * <p>Calls to future.get(*) on every call to this.get(*) also provide
        * the cancellation behavior for this.
        *
        * <p>(Consider: in thread A, call get(), in thread B call get(long,
        * TimeUnit). Thread B may have to wait for Thread A to finish, which
        * would be unacceptable.)
        *
        * <p>Note that each call to Future<O>.get(*) results in a call to
        * Future<I>.get(*), but the function is only applied once, so
        * Future<I>.get(*) is assumed to be idempotent.
        */

       private final Object lock = new Object();
       private boolean set = false;
       private O value = null;

       /*@Override*/
       public O get() throws InterruptedException, ExecutionException {
         return apply(future.get());
       }

       /*@Override*/
       public O get(long timeout, TimeUnit unit) throws InterruptedException,
           ExecutionException, TimeoutException {
         return apply(future.get(timeout, unit));
       }

       private O apply(I raw) {
         synchronized(lock) {
           if (!set) {
             value = function.apply(raw);
             set = true;
           }
           return value;
         }
       }

       /*@Override*/
       public boolean cancel(boolean mayInterruptIfRunning) {
         return future.cancel(mayInterruptIfRunning);
       }

       /*@Override*/
       public boolean isCancelled() {
         return future.isCancelled();
       }

       /*@Override*/
       public boolean isDone() {
         return future.isDone();
       }
     };
   }

   /**
    * An implementation of {@code ListenableFuture} that also implements
    * {@code Runnable} so that it can be used to nest ListenableFutures.
    * Once the passed-in {@code ListenableFuture} is complete, it calls the
    * passed-in {@code Function} to generate the result.
    * The resulting future doesn't interrupt when aborted.
    *
    * <p>If the function throws any checked exceptions, they should be wrapped
    * in a {@code UndeclaredThrowableException} so that this class can get
    * access to the cause.
    */
   private static class ChainingListenableFuture<I, O>
       extends AbstractListenableFuture<O> implements Runnable {

     private final Function<? super I, ? extends ListenableFuture<? extends O>>
         function;
     private final UninterruptibleFuture<? extends I> inputFuture;

     private ChainingListenableFuture(
         Function<? super I, ? extends ListenableFuture<? extends O>> function,
         ListenableFuture<? extends I> inputFuture) {
       this.function = function;
       this.inputFuture = makeUninterruptible(inputFuture);
     }

     public void run() {
       try {
         I sourceResult;
         try {
           sourceResult = inputFuture.get();
         } catch (CancellationException e) {
           // Cancel this future and return.
           cancel();
           return;
         } catch (ExecutionException e) {
           // Set the cause of the exception as this future's exception
           setException(e.getCause());
           return;
         }

         final ListenableFuture<? extends O> outputFuture =
             function.apply(sourceResult);
         outputFuture.addListener(new Runnable() {
             public void run() {
               try {
                 // Here it would have been nice to have had an
                 // UninterruptibleListenableFuture, but we don't want to start a
                 // combinatorial explosion of interfaces, so we have to make do.
                 set(makeUninterruptible(outputFuture).get());
               } catch (ExecutionException e) {
                 // Set the cause of the exception as this future's exception
                 setException(e.getCause());
               }
             }
           }, Executors.sameThreadExecutor());
       } catch (UndeclaredThrowableException e) {
         // Set the cause of the exception as this future's exception
         setException(e.getCause());
       } catch (RuntimeException e) {
         // This exception is irrelevant in this thread, but useful for the
         // client
         setException(e);
       } catch (Error e) {
         // This seems evil, but the client needs to know an error occured and
         // the error needs to be propagated ASAP.
         setException(e);
         throw e;
       }
     }
   }

   /**
    * A checked future that uses a function to map from exceptions to the
    * appropriate checked type.
    */
   private static class MappingCheckedFuture<T, E extends Exception> extends
       AbstractCheckedFuture<T, E> {

     final Function<Exception, E> mapper;

     MappingCheckedFuture(ListenableFuture<T> delegate,
         Function<Exception, E> mapper) {
       super(delegate);

       this.mapper = mapper;
     }

     @Override
     protected E mapException(Exception e) {
       return mapper.apply(e);
     }
   }

   /**
    * An adapter to turn a {@link Future} into a {@link ListenableFuture}.  This
    * will wait on the future to finish, and when it completes, run the
    * listeners.  This implementation will wait on the source future
    * indefinitely, so if the source future never completes, the adapter will
    * never complete either.
    *
    * <p>If the delegate future is interrupted or throws an unexpected unchecked
    * exception, the listeners will not be invoked.
    */
   private static class ListenableFutureAdapter<T> extends ForwardingFuture<T>
       implements ListenableFuture<T> {

     private static final Executor adapterExecutor =
         java.util.concurrent.Executors.newCachedThreadPool();

     // The execution list to hold our listeners.
     private final ExecutionList executionList = new ExecutionList();

     // This allows us to only start up a thread waiting on the delegate future
     // when the first listener is added.
     private final AtomicBoolean hasListeners = new AtomicBoolean(false);

     // The delegate future.
     private final Future<T> delegate;

     ListenableFutureAdapter(final Future<T> delegate) {
       this.delegate = delegate;
     }

     @Override
     protected Future<T> delegate() {
       return delegate;
     }

     /*@Override*/
     public void addListener(Runnable listener, Executor exec) {

       // When a listener is first added, we run a task that will wait for
       // the delegate to finish, and when it is done will run the listeners.
       if (!hasListeners.get() && hasListeners.compareAndSet(false, true)) {
         adapterExecutor.execute(new Runnable() {
           /*@Override*/
           public void run() {
             try {
               delegate.get();
             } catch (CancellationException e) {
               // The task was cancelled, so it is done, run the listeners.
             } catch (InterruptedException e) {
               // This thread was interrupted.  This should never happen, so we
               // throw an IllegalStateException.
               throw new IllegalStateException("Adapter thread interrupted!", e);
             } catch (ExecutionException e) {
               // The task caused an exception, so it is done, run the listeners.
             }
             executionList.run();
           }
         });
       }
       executionList.add(listener, exec);
     }
   }
 }
	/*
	* Copyright (C) 2006 Google Inc.
	*
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package com.google.common.util.concurrent;

	import com.google.common.base.Function;

	import java.lang.reflect.UndeclaredThrowableException;
	import java.util.concurrent.CancellationException;
	import java.util.concurrent.ExecutionException;
	import java.util.concurrent.Executor;
	import java.util.concurrent.Future;
	import java.util.concurrent.TimeUnit;
	import java.util.concurrent.TimeoutException;
	import java.util.concurrent.atomic.AtomicBoolean;

	import javax.annotation.Nullable;

	import static com.google.common.base.Preconditions.checkNotNull;

	/**
	* Static utility methods pertaining to the {@link Future} interface.
	*
	* @author Kevin Bourrillion
	* @author Nishant Thakkar
	* @author Sven Mawson
	* @since 2009.09.15 <b>tentative</b>
	*/
	public class Futures {
	private Futures() {}

	/**
	* Returns an uninterruptible view of a {@code Future}. If a thread is
	* interrupted during an attempt to {@code get()} from the returned future, it
	* continues to wait on the result until it is available or the timeout
	* elapses, and only then re-interrupts the thread.
	*/
	public static <V> UninterruptibleFuture<V> makeUninterruptible(
	final Future<V> future) {
	checkNotNull(future);
	if (future instanceof UninterruptibleFuture) {
	return (UninterruptibleFuture<V>) future;
	}
	return new UninterruptibleFuture<V>() {
	public boolean cancel(boolean mayInterruptIfRunning) {
	return future.cancel(mayInterruptIfRunning);
	}
	public boolean isCancelled() {
	return future.isCancelled();
	}
	public boolean isDone() {
	return future.isDone();
	}

	public V get(long timeoutDuration, TimeUnit timeoutUnit)
	throws TimeoutException, ExecutionException {
	boolean interrupted = false;
	try {
	long timeoutNanos = timeoutUnit.toNanos(timeoutDuration);
	long end = System.nanoTime() + timeoutNanos;
	for (long remaining = timeoutNanos; remaining > 0;
	remaining = end - System.nanoTime()) {
	try {
	return future.get(remaining, TimeUnit.NANOSECONDS);
	} catch (InterruptedException ignored) {
	interrupted = true;
	}
	}
	throw new TimeoutException();
	} finally {
	if (interrupted) {
	Thread.currentThread().interrupt();
	}
	}
	}

	public V get() throws ExecutionException {
	boolean interrupted = false;
	try {
	while (true) {
	try {
	return future.get();
	} catch (InterruptedException ignored) {
	interrupted = true;
	}
	}
	} finally {
	if (interrupted) {
	Thread.currentThread().interrupt();
	}
	}
	}
	};
	}

	/**
	* Creates a {@link ListenableFuture} out of a normal {@link Future}. The
	* returned future will create a thread to wait for the source future to
	* complete before executing the listeners.
	*
	* <p>Callers who have a future that subclasses
	* {@link java.util.concurrent.FutureTask} may want to instead subclass
	* {@link ListenableFutureTask}, which adds the {@link ListenableFuture}
	* functionality to the standard {@code FutureTask} implementation.
	*/
	public static <T> ListenableFuture<T> makeListenable(Future<T> future) {
	if (future instanceof ListenableFuture) {
	return (ListenableFuture<T>) future;
	}
	return new ListenableFutureAdapter<T>(future);
	}

	/**
	* Creates a {@link CheckedFuture} out of a normal {@link Future} and a
	* {@link Function} that maps from {@link Exception} instances into the
	* appropriate checked type.
	*
	* <p>The given mapping function will be applied to an
	* {@link InterruptedException}, a {@link CancellationException}, or an
	* {@link ExecutionException} with the actual cause of the exception.
	* See {@link Future#get()} for details on the exceptions thrown.
	*/
	public static <T, E extends Exception> CheckedFuture<T, E> makeChecked(
	Future<T> future, Function<Exception, E> mapper) {
	return new MappingCheckedFuture<T, E>(makeListenable(future), mapper);
	}

	/**
	* Creates a {@code ListenableFuture} which has its value set immediately upon
	* construction. The getters just return the value. This {@code Future} can't
	* be canceled or timed out and its {@code isDone()} method always returns
	* {@code true}. It's useful for returning something that implements the
	* {@code ListenableFuture} interface but already has the result.
	*/
	public static <T> ListenableFuture<T> immediateFuture(@Nullable T value) {
	ValueFuture<T> future = ValueFuture.create();
	future.set(value);
	return future;
	}

	/**
	* Creates a {@code CheckedFuture} which has its value set immediately upon
	* construction. The getters just return the value. This {@code Future} can't
	* be canceled or timed out and its {@code isDone()} method always returns
	* {@code true}. It's useful for returning something that implements the
	* {@code CheckedFuture} interface but already has the result.
	*/
	public static <T, E extends Exception> CheckedFuture<T, E>
	immediateCheckedFuture(@Nullable T value) {
	ValueFuture<T> future = ValueFuture.create();
	future.set(value);
	return Futures.makeChecked(future, new Function<Exception, E>() {
	public E apply(Exception e) {
	throw new AssertionError("impossible");
	}
	});
	}

	/**
	* Creates a {@code ListenableFuture} which has an exception set immediately
	* upon construction. The getters just return the value. This {@code Future}
	* can't be canceled or timed out and its {@code isDone()} method always
	* returns {@code true}. It's useful for returning something that implements
	* the {@code ListenableFuture} interface but already has a failed
	* result. Calling {@code get()} will throw the provided {@code Throwable}
	* (wrapped in an {@code ExecutionException}).
	*
	* @throws Error if the throwable was an {@link Error}.
	*/
	public static <T> ListenableFuture<T> immediateFailedFuture(
	Throwable throwable) {
	checkNotNull(throwable);
	ValueFuture<T> future = ValueFuture.create();
	future.setException(throwable);
	return future;
	}

	/**
	* Creates a {@code CheckedFuture} which has an exception set immediately
	* upon construction. The getters just return the value. This {@code Future}
	* can't be canceled or timed out and its {@code isDone()} method always
	* returns {@code true}. It's useful for returning something that implements
	* the {@code CheckedFuture} interface but already has a failed result.
	* Calling {@code get()} will throw the provided {@code Throwable} (wrapped in
	* an {@code ExecutionException}) and calling {@code checkedGet()} will throw
	* the provided exception itself.
	*
	* @throws Error if the throwable was an {@link Error}.
	*/
	public static <T, E extends Exception> CheckedFuture<T, E>
	immediateFailedCheckedFuture(final E exception) {
	checkNotNull(exception);
	return makeChecked(Futures.<T>immediateFailedFuture(exception),
	new Function<Exception, E>() {
	public E apply(Exception e) {
	return exception;
	}
	});
	}

	/**
	* Creates a new {@code ListenableFuture} that wraps another
	* {@code ListenableFuture}. The result of the new future is the result of
	* the provided function called on the result of the provided future.
	* The resulting future doesn't interrupt when aborted.
	*
	* <p>TODO: Add a version that accepts a normal {@code Future}
	*
	* <p>The typical use for this method would be when a RPC call is dependent on
	* the results of another RPC. One would call the first RPC (input), create a
	* function that calls another RPC based on input's result, and then call
	* chain on input and that function to get a {@code ListenableFuture} of
	* the result.
	*
	* @param input The future to chain
	* @param function A function to chain the results of the provided future
	* to the results of the returned future. This will be run in the thread
	* that notifies input it is complete.
	* @return A future that holds result of the chain.
	*/
	public static <I, O> ListenableFuture<O> chain(ListenableFuture<I> input,
	Function<? super I, ? extends ListenableFuture<? extends O>> function) {
	return chain(input, function, Executors.sameThreadExecutor());
	}

	/**
	* Creates a new {@code ListenableFuture} that wraps another
	* {@code ListenableFuture}. The result of the new future is the result of
	* the provided function called on the result of the provided future.
	* The resulting future doesn't interrupt when aborted.
	*
	* <p>This version allows an arbitrary executor to be passed in for running
	* the chained Function. When using {@link Executors#sameThreadExecutor}, the
	* thread chained Function executes in will be whichever thread set the
	* result of the input Future, which may be the network thread in the case of
	* RPC-based Futures.
	*
	* @param input The future to chain
	* @param function A function to chain the results of the provided future
	* to the results of the returned future.
	* @param exec Executor to run the function in.
	* @return A future that holds result of the chain.
	*/
	public static <I, O> ListenableFuture<O> chain(ListenableFuture<I> input,
	Function<? super I, ? extends ListenableFuture<? extends O>> function,
	Executor exec) {
	ChainingListenableFuture<I, O> chain =
	new ChainingListenableFuture<I, O>(function, input);
	input.addListener(chain, exec);
	return chain;
	}

	/**
	* Creates a new {@code ListenableFuture} that wraps another
	* {@code ListenableFuture}. The result of the new future is the result of
	* the provided function called on the result of the provided future.
	* The resulting future doesn't interrupt when aborted.
	*
	* <p>An example use of this method is to convert a serializable object
	* returned from an RPC into a POJO.
	*
	* @param future The future to compose
	* @param function A Function to compose the results of the provided future
	* to the results of the returned future. This will be run in the thread
	* that notifies input it is complete.
	* @return A future that holds result of the composition.
	*/
	public static <I, O> ListenableFuture<O> compose(ListenableFuture<I> future,
	final Function<? super I, ? extends O> function) {
	return compose(future, function, Executors.sameThreadExecutor());
	}

	/**
	* Creates a new {@code ListenableFuture} that wraps another
	* {@code ListenableFuture}. The result of the new future is the result of
	* the provided function called on the result of the provided future.
	* The resulting future doesn't interrupt when aborted.
	*
	* <p>An example use of this method is to convert a serializable object
	* returned from an RPC into a POJO.
	*
	* <p>This version allows an arbitrary executor to be passed in for running
	* the chained Function. When using {@link Executors#sameThreadExecutor}, the
	* thread chained Function executes in will be whichever thread set the result
	* of the input Future, which may be the network thread in the case of
	* RPC-based Futures.
	*
	* @param future The future to compose
	* @param function A Function to compose the results of the provided future
	* to the results of the returned future.
	* @param exec Executor to run the function in.
	* @return A future that holds result of the composition.
	* @since 2010.01.04 <b>tentative</b>
	*/
	public static <I, O> ListenableFuture<O> compose(ListenableFuture<I> future,
	final Function<? super I, ? extends O> function, Executor exec) {
	Function<I, ListenableFuture<O>> wrapperFunction
	= new Function<I, ListenableFuture<O>>() {
	/@Override/ public ListenableFuture<O> apply(I input) {
	O output = function.apply(input);
	return immediateFuture(output);
	}
	};
	return chain(future, wrapperFunction, exec);
	}

	/**
	* Creates a new {@code Future} that wraps another {@code Future}.
	* The result of the new future is the result of the provided function called
	* on the result of the provided future.
	*
	* <p>An example use of this method is to convert a Future that produces a
	* handle to an object to a future that produces the object itself.
	*
	* <p>Each call to {@code Future<O>.get(*)} results in a call to
	* {@code Future<I>.get(*)}, but {@code function} is only applied once, so it
	* is assumed that {@code Future<I>.get(*)} is idempotent.
	*
	* <p>When calling {@link Future#get(long, TimeUnit)} on the returned
	* future, the timeout only applies to the future passed in to this method.
	* Any additional time taken by applying {@code function} is not considered.
	*
	* @param future The future to compose
	* @param function A Function to compose the results of the provided future
	* to the results of the returned future. This will be run in the thread
	* that calls one of the varieties of {@code get()}.
	* @return A future that computes result of the composition.
	*/
	public static <I, O> Future<O> compose(final Future<I> future,
	final Function<? super I, ? extends O> function) {

	return new Future<O>() {

	/*
	* Concurrency detail:
	*
	* <p>To preserve the idempotency of calls to this.get(*) calls to the
	* function are only applied once. A lock is required to prevent multiple
	* applications of the function. The calls to future.get(*) are performed
	* outside the lock, as is required to prevent calls to
	* get(long, TimeUnit) to persist beyond their timeout.
	*
	* <p>Calls to future.get() on every call to this.get() also provide
	* the cancellation behavior for this.
	*
	* <p>(Consider: in thread A, call get(), in thread B call get(long,
	* TimeUnit). Thread B may have to wait for Thread A to finish, which
	* would be unacceptable.)
	*
	* <p>Note that each call to Future<O>.get(*) results in a call to
	* Future<I>.get(*), but the function is only applied once, so
	* Future<I>.get(*) is assumed to be idempotent.
	*/

	private final Object lock = new Object();
	private boolean set = false;
	private O value = null;

	/@Override/
	public O get() throws InterruptedException, ExecutionException {
	return apply(future.get());
	}

	/@Override/
	public O get(long timeout, TimeUnit unit) throws InterruptedException,
	ExecutionException, TimeoutException {
	return apply(future.get(timeout, unit));
	}

	private O apply(I raw) {
	synchronized(lock) {
	if (!set) {
	value = function.apply(raw);
	set = true;
	}
	return value;
	}
	}

	/@Override/
	public boolean cancel(boolean mayInterruptIfRunning) {
	return future.cancel(mayInterruptIfRunning);
	}

	/@Override/
	public boolean isCancelled() {
	return future.isCancelled();
	}

	/@Override/
	public boolean isDone() {
	return future.isDone();
	}
	};
	}

	/**
	* An implementation of {@code ListenableFuture} that also implements
	* {@code Runnable} so that it can be used to nest ListenableFutures.
	* Once the passed-in {@code ListenableFuture} is complete, it calls the
	* passed-in {@code Function} to generate the result.
	* The resulting future doesn't interrupt when aborted.
	*
	* <p>If the function throws any checked exceptions, they should be wrapped
	* in a {@code UndeclaredThrowableException} so that this class can get
	* access to the cause.
	*/
	private static class ChainingListenableFuture<I, O>
	extends AbstractListenableFuture<O> implements Runnable {

	private final Function<? super I, ? extends ListenableFuture<? extends O>>
	function;
	private final UninterruptibleFuture<? extends I> inputFuture;

	private ChainingListenableFuture(
	Function<? super I, ? extends ListenableFuture<? extends O>> function,
	ListenableFuture<? extends I> inputFuture) {
	this.function = function;
	this.inputFuture = makeUninterruptible(inputFuture);
	}

	public void run() {
	try {
	I sourceResult;
	try {
	sourceResult = inputFuture.get();
	} catch (CancellationException e) {
	// Cancel this future and return.
	cancel();
	return;
	} catch (ExecutionException e) {
	// Set the cause of the exception as this future's exception
	setException(e.getCause());
	return;
	}

	final ListenableFuture<? extends O> outputFuture =
	function.apply(sourceResult);
	outputFuture.addListener(new Runnable() {
	public void run() {
	try {
	// Here it would have been nice to have had an
	// UninterruptibleListenableFuture, but we don't want to start a
	// combinatorial explosion of interfaces, so we have to make do.
	set(makeUninterruptible(outputFuture).get());
	} catch (ExecutionException e) {
	// Set the cause of the exception as this future's exception
	setException(e.getCause());
	}
	}
	}, Executors.sameThreadExecutor());
	} catch (UndeclaredThrowableException e) {
	// Set the cause of the exception as this future's exception
	setException(e.getCause());
	} catch (RuntimeException e) {
	// This exception is irrelevant in this thread, but useful for the
	// client
	setException(e);
	} catch (Error e) {
	// This seems evil, but the client needs to know an error occured and
	// the error needs to be propagated ASAP.
	setException(e);
	throw e;
	}
	}
	}

	/**
	* A checked future that uses a function to map from exceptions to the
	* appropriate checked type.
	*/
	private static class MappingCheckedFuture<T, E extends Exception> extends
	AbstractCheckedFuture<T, E> {

	final Function<Exception, E> mapper;

	MappingCheckedFuture(ListenableFuture<T> delegate,
	Function<Exception, E> mapper) {
	super(delegate);

	this.mapper = mapper;
	}

	@Override
	protected E mapException(Exception e) {
	return mapper.apply(e);
	}
	}

	/**
	* An adapter to turn a {@link Future} into a {@link ListenableFuture}. This
	* will wait on the future to finish, and when it completes, run the
	* listeners. This implementation will wait on the source future
	* indefinitely, so if the source future never completes, the adapter will
	* never complete either.
	*
	* <p>If the delegate future is interrupted or throws an unexpected unchecked
	* exception, the listeners will not be invoked.
	*/
	private static class ListenableFutureAdapter<T> extends ForwardingFuture<T>
	implements ListenableFuture<T> {

	private static final Executor adapterExecutor =
	java.util.concurrent.Executors.newCachedThreadPool();

	// The execution list to hold our listeners.
	private final ExecutionList executionList = new ExecutionList();

	// This allows us to only start up a thread waiting on the delegate future
	// when the first listener is added.
	private final AtomicBoolean hasListeners = new AtomicBoolean(false);

	// The delegate future.
	private final Future<T> delegate;

	ListenableFutureAdapter(final Future<T> delegate) {
	this.delegate = delegate;
	}

	@Override
	protected Future<T> delegate() {
	return delegate;
	}

	/@Override/
	public void addListener(Runnable listener, Executor exec) {

	// When a listener is first added, we run a task that will wait for
	// the delegate to finish, and when it is done will run the listeners.
	if (!hasListeners.get() && hasListeners.compareAndSet(false, true)) {
	adapterExecutor.execute(new Runnable() {
	/@Override/
	public void run() {
	try {
	delegate.get();
	} catch (CancellationException e) {
	// The task was cancelled, so it is done, run the listeners.
	} catch (InterruptedException e) {
	// This thread was interrupted. This should never happen, so we
	// throw an IllegalStateException.
	throw new IllegalStateException("Adapter thread interrupted!", e);
	} catch (ExecutionException e) {
	// The task caused an exception, so it is done, run the listeners.
	}
	executionList.run();
	}
	});
	}
	executionList.add(listener, exec);
	}
	}
	}