/*
* 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);
}
}
}