This example Java source code file (Futures.java) is included in the alvinalexander.com
"Java Source Code
Warehouse" project. The intent of this project is to help you "Learn
Java by Example" TM.
/*
* Copyright (C) 2006 The Guava Authors
*
* 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 static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static com.google.common.util.concurrent.Uninterruptibles.getUninterruptibly;
import com.google.common.annotations.Beta;
import com.google.common.annotations.GwtCompatible;
import com.google.common.annotations.GwtIncompatible;
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Queues;
import com.google.common.util.concurrent.CollectionFuture.ListFuture;
import com.google.common.util.concurrent.ImmediateFuture.ImmediateCancelledFuture;
import com.google.common.util.concurrent.ImmediateFuture.ImmediateFailedCheckedFuture;
import com.google.common.util.concurrent.ImmediateFuture.ImmediateFailedFuture;
import com.google.common.util.concurrent.ImmediateFuture.ImmediateSuccessfulCheckedFuture;
import com.google.common.util.concurrent.ImmediateFuture.ImmediateSuccessfulFuture;
import com.google.errorprone.annotations.CanIgnoreReturnValue;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import javax.annotation.Nullable;
/**
* Static utility methods pertaining to the {@link Future} interface.
*
* <p>Many of these methods use the {@link ListenableFuture} API; consult the Guava User Guide
* article on <a href="https://github.com/google/guava/wiki/ListenableFutureExplained">
* {@code ListenableFuture}</a>.
*
* @author Kevin Bourrillion
* @author Nishant Thakkar
* @author Sven Mawson
* @since 1.0
*/
@Beta
@GwtCompatible(emulated = true)
public final class Futures extends GwtFuturesCatchingSpecialization {
// A note on memory visibility.
// Many of the utilities in this class (transform, withFallback, withTimeout, asList, combine)
// have two requirements that significantly complicate their design.
// 1. Cancellation should propagate from the returned future to the input future(s).
// 2. The returned futures shouldn't unnecessarily 'pin' their inputs after completion.
//
// A consequence of these requirements is that the delegate futures cannot be stored in
// final fields.
//
// For simplicity the rest of this description will discuss Futures.catching since it is the
// simplest instance, though very similar descriptions apply to many other classes in this file.
//
// In the constructor of AbstractCatchingFuture, the delegate future is assigned to a field
// 'inputFuture'. That field is non-final and non-volatile. There are 2 places where the
// 'inputFuture' field is read and where we will have to consider visibility of the write
// operation in the constructor.
//
// 1. In the listener that performs the callback. In this case it is fine since inputFuture is
// assigned prior to calling addListener, and addListener happens-before any invocation of the
// listener. Notably, this means that 'volatile' is unnecessary to make 'inputFuture' visible
// to the listener.
//
// 2. In done() where we may propagate cancellation to the input. In this case it is _not_ fine.
// There is currently nothing that enforces that the write to inputFuture in the constructor is
// visible to done(). This is because there is no happens before edge between the write and a
// (hypothetical) unsafe read by our caller. Note: adding 'volatile' does not fix this issue,
// it would just add an edge such that if done() observed non-null, then it would also
// definitely observe all earlier writes, but we still have no guarantee that done() would see
// the inital write (just stronger guarantees if it does).
//
// See: http://cs.oswego.edu/pipermail/concurrency-interest/2015-January/013800.html
// For a (long) discussion about this specific issue and the general futility of life.
//
// For the time being we are OK with the problem discussed above since it requires a caller to
// introduce a very specific kind of data-race. And given the other operations performed by these
// methods that involve volatile read/write operations, in practice there is no issue. Also, the
// way in such a visibility issue would surface is most likely as a failure of cancel() to
// propagate to the input. Cancellation propagation is fundamentally racy so this is fine.
//
// Future versions of the JMM may revise safe construction semantics in such a way that we can
// safely publish these objects and we won't need this whole discussion.
// TODO(user,lukes): consider adding volatile to all these fields since in current known JVMs
// that should resolve the issue. This comes at the cost of adding more write barriers to the
// implementations.
private Futures() {}
/**
* Creates a {@link CheckedFuture} out of a normal {@link ListenableFuture} and a {@link Function}
* that maps from {@link Exception} instances into the appropriate checked type.
*
* <p>Warning: We recommend against using {@code CheckedFuture} in new projects. {@code
* CheckedFuture} is difficult to build libraries atop. {@code CheckedFuture} ports of methods
* like {@link Futures#transformAsync} have historically had bugs, and some of these bugs are
* necessary, unavoidable consequences of the {@code CheckedFuture} API. Additionally, {@code
* CheckedFuture} encourages users to take exceptions from one thread and rethrow them in another,
* producing confusing stack traces.
*
* <p>The given mapping function will be applied to an {@link InterruptedException}, a {@link
* CancellationException}, or an {@link ExecutionException}. See {@link Future#get()} for details
* on the exceptions thrown.
*
* @since 9.0 (source-compatible since 1.0)
*/
@GwtIncompatible // TODO
public static <V, X extends Exception> CheckedFuture makeChecked(
ListenableFuture<V> future, Function mapper) {
return new MappingCheckedFuture<V, X>(checkNotNull(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}.
*/
public static <V> ListenableFuture immediateFuture(@Nullable V value) {
if (value == null) {
// This cast is safe because null is assignable to V for all V (i.e. it is covariant)
@SuppressWarnings({"unchecked", "rawtypes"})
ListenableFuture<V> typedNull = (ListenableFuture) ImmediateSuccessfulFuture.NULL;
return typedNull;
}
return new ImmediateSuccessfulFuture<V>(value);
}
/**
* Returns a {@code CheckedFuture} which has its value set immediately upon construction.
*
* <p>The returned {@code Future} can't be cancelled, and its {@code isDone()} method always
* returns {@code true}. Calling {@code get()} or {@code checkedGet()} will immediately return the
* provided value.
*/
@GwtIncompatible // TODO
public static <V, X extends Exception> CheckedFuture immediateCheckedFuture(
@Nullable V value) {
return new ImmediateSuccessfulCheckedFuture<V, X>(value);
}
/**
* Returns a {@code ListenableFuture} which has an exception set immediately upon construction.
*
* <p>The returned {@code Future} can't be cancelled, and its {@code isDone()} method always
* returns {@code true}. Calling {@code get()} will immediately throw the provided {@code
* Throwable} wrapped in an {@code ExecutionException}.
*/
public static <V> ListenableFuture immediateFailedFuture(Throwable throwable) {
checkNotNull(throwable);
return new ImmediateFailedFuture<V>(throwable);
}
/**
* Creates a {@code ListenableFuture} which is cancelled immediately upon construction, so that
* {@code isCancelled()} always returns {@code true}.
*
* @since 14.0
*/
public static <V> ListenableFuture immediateCancelledFuture() {
return new ImmediateCancelledFuture<V>();
}
/**
* Returns a {@code CheckedFuture} which has an exception set immediately upon construction.
*
* <p>The returned {@code Future} can't be cancelled, and its {@code isDone()} method always
* returns {@code true}. Calling {@code get()} will immediately throw the provided {@code
* Exception} wrapped in an {@code ExecutionException}, and calling {@code checkedGet()} will
* throw the provided exception itself.
*/
@GwtIncompatible // TODO
public static <V, X extends Exception> CheckedFuture immediateFailedCheckedFuture(
X exception) {
checkNotNull(exception);
return new ImmediateFailedCheckedFuture<V, X>(exception);
}
/**
* Returns a {@code Future} whose result is taken from the given primary {@code input} or, if the
* primary input fails with the given {@code exceptionType}, from the result provided by the
* {@code fallback}. {@link Function#apply} is not invoked until the primary input has failed, so
* if the primary input succeeds, it is never invoked. If, during the invocation of {@code
* fallback}, an exception is thrown, this exception is used as the result of the output {@code
* Future}.
*
* <p>Usage example:
*
* <pre> {@code
* ListenableFuture<Integer> fetchCounterFuture = ...;
*
* // Falling back to a zero counter in case an exception happens when
* // processing the RPC to fetch counters.
* ListenableFuture<Integer> faultTolerantFuture = Futures.catching(
* fetchCounterFuture, FetchException.class,
* new Function<FetchException, Integer>() {
* public Integer apply(FetchException e) {
* return 0;
* }
* });}</pre>
*
* <p>This overload, which does not accept an executor, uses {@code directExecutor}, a dangerous
* choice in some cases. See the discussion in the {@link ListenableFuture#addListener
* ListenableFuture.addListener} documentation. The documentation's warnings about "lightweight
* listeners" refer here to the work done during {@code Function.apply}.
*
* @param input the primary input {@code Future}
* @param exceptionType the exception type that triggers use of {@code fallback}. The exception
* type is matched against the input's exception. "The input's exception" means the cause of
* the {@link ExecutionException} thrown by {@code input.get()} or, if {@code get()} throws a
* different kind of exception, that exception itself. To avoid hiding bugs and other
* unrecoverable errors, callers should prefer more specific types, avoiding {@code
* Throwable.class} in particular.
* @param fallback the {@link Function} to be called if {@code input} fails with the expected
* exception type. The function's argument is the input's exception. "The input's exception"
* means the cause of the {@link ExecutionException} thrown by {@code input.get()} or, if
* {@code get()} throws a different kind of exception, that exception itself.
* @since 19.0
*/
@Partially.GwtIncompatible("AVAILABLE but requires exceptionType to be Throwable.class")
public static <V, X extends Throwable> ListenableFuture catching(
ListenableFuture<? extends V> input,
Class<X> exceptionType,
Function<? super X, ? extends V> fallback) {
return AbstractCatchingFuture.create(input, exceptionType, fallback);
}
/**
* Returns a {@code Future} whose result is taken from the given primary {@code input} or, if the
* primary input fails with the given {@code exceptionType}, from the result provided by the
* {@code fallback}. {@link Function#apply} is not invoked until the primary input has failed, so
* if the primary input succeeds, it is never invoked. If, during the invocation of {@code
* fallback}, an exception is thrown, this exception is used as the result of the output {@code
* Future}.
*
* <p>Usage example:
*
* <pre> {@code
* ListenableFuture<Integer> fetchCounterFuture = ...;
*
* // Falling back to a zero counter in case an exception happens when
* // processing the RPC to fetch counters.
* ListenableFuture<Integer> faultTolerantFuture = Futures.catching(
* fetchCounterFuture, FetchException.class,
* new Function<FetchException, Integer>() {
* public Integer apply(FetchException e) {
* return 0;
* }
* }, directExecutor());}</pre>
*
* <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
* the discussion in the {@link ListenableFuture#addListener ListenableFuture.addListener}
* documentation. The documentation's warnings about "lightweight listeners" refer here to the
* work done during {@code Function.apply}.
*
* @param input the primary input {@code Future}
* @param exceptionType the exception type that triggers use of {@code fallback}. The exception
* type is matched against the input's exception. "The input's exception" means the cause of
* the {@link ExecutionException} thrown by {@code input.get()} or, if {@code get()} throws a
* different kind of exception, that exception itself. To avoid hiding bugs and other
* unrecoverable errors, callers should prefer more specific types, avoiding {@code
* Throwable.class} in particular.
* @param fallback the {@link Function} to be called if {@code input} fails with the expected
* exception type. The function's argument is the input's exception. "The input's exception"
* means the cause of the {@link ExecutionException} thrown by {@code input.get()} or, if
* {@code get()} throws a different kind of exception, that exception itself.
* @param executor the executor that runs {@code fallback} if {@code input} fails
* @since 19.0
*/
@Partially.GwtIncompatible("AVAILABLE but requires exceptionType to be Throwable.class")
public static <V, X extends Throwable> ListenableFuture catching(
ListenableFuture<? extends V> input,
Class<X> exceptionType,
Function<? super X, ? extends V> fallback,
Executor executor) {
return AbstractCatchingFuture.create(input, exceptionType, fallback, executor);
}
/**
* Returns a {@code Future} whose result is taken from the given primary {@code input} or, if the
* primary input fails with the given {@code exceptionType}, from the result provided by the
* {@code fallback}. {@link AsyncFunction#apply} is not invoked until the primary input has
* failed, so if the primary input succeeds, it is never invoked. If, during the invocation of
* {@code fallback}, an exception is thrown, this exception is used as the result of the output
* {@code Future}.
*
* <p>Usage examples:
*
* <pre> {@code
* ListenableFuture<Integer> fetchCounterFuture = ...;
*
* // Falling back to a zero counter in case an exception happens when
* // processing the RPC to fetch counters.
* ListenableFuture<Integer> faultTolerantFuture = Futures.catchingAsync(
* fetchCounterFuture, FetchException.class,
* new AsyncFunction<FetchException, Integer>() {
* public ListenableFuture<Integer> apply(FetchException e) {
* return immediateFuture(0);
* }
* });}</pre>
*
* <p>The fallback can also choose to propagate the original exception when desired:
*
* <pre> {@code
* ListenableFuture<Integer> fetchCounterFuture = ...;
*
* // Falling back to a zero counter only in case the exception was a
* // TimeoutException.
* ListenableFuture<Integer> faultTolerantFuture = Futures.catchingAsync(
* fetchCounterFuture, FetchException.class,
* new AsyncFunction<FetchException, Integer>() {
* public ListenableFuture<Integer> apply(FetchException e)
* throws FetchException {
* if (omitDataOnFetchFailure) {
* return immediateFuture(0);
* }
* throw e;
* }
* });}</pre>
*
* <p>This overload, which does not accept an executor, uses {@code directExecutor}, a dangerous
* choice in some cases. See the discussion in the {@link ListenableFuture#addListener
* ListenableFuture.addListener} documentation. The documentation's warnings about "lightweight
* listeners" refer here to the work done during {@code AsyncFunction.apply}, not to any work done
* to complete the returned {@code Future}.
*
* @param input the primary input {@code Future}
* @param exceptionType the exception type that triggers use of {@code fallback}. The exception
* type is matched against the input's exception. "The input's exception" means the cause of
* the {@link ExecutionException} thrown by {@code input.get()} or, if {@code get()} throws a
* different kind of exception, that exception itself. To avoid hiding bugs and other
* unrecoverable errors, callers should prefer more specific types, avoiding {@code
* Throwable.class} in particular.
* @param fallback the {@link AsyncFunction} to be called if {@code input} fails with the expected
* exception type. The function's argument is the input's exception. "The input's exception"
* means the cause of the {@link ExecutionException} thrown by {@code input.get()} or, if
* {@code get()} throws a different kind of exception, that exception itself.
* @since 19.0 (similar functionality in 14.0 as {@code withFallback})
*/
@CanIgnoreReturnValue // TODO(kak): @CheckReturnValue
@Partially.GwtIncompatible("AVAILABLE but requires exceptionType to be Throwable.class")
public static <V, X extends Throwable> ListenableFuture catchingAsync(
ListenableFuture<? extends V> input,
Class<X> exceptionType,
AsyncFunction<? super X, ? extends V> fallback) {
return AbstractCatchingFuture.create(input, exceptionType, fallback);
}
/**
* Returns a {@code Future} whose result is taken from the given primary {@code input} or, if the
* primary input fails with the given {@code exceptionType}, from the result provided by the
* {@code fallback}. {@link AsyncFunction#apply} is not invoked until the primary input has
* failed, so if the primary input succeeds, it is never invoked. If, during the invocation of
* {@code fallback}, an exception is thrown, this exception is used as the result of the output
* {@code Future}.
*
* <p>Usage examples:
*
* <pre> {@code
* ListenableFuture<Integer> fetchCounterFuture = ...;
*
* // Falling back to a zero counter in case an exception happens when
* // processing the RPC to fetch counters.
* ListenableFuture<Integer> faultTolerantFuture = Futures.catchingAsync(
* fetchCounterFuture, FetchException.class,
* new AsyncFunction<FetchException, Integer>() {
* public ListenableFuture<Integer> apply(FetchException e) {
* return immediateFuture(0);
* }
* }, directExecutor());}</pre>
*
* <p>The fallback can also choose to propagate the original exception when desired:
*
* <pre> {@code
* ListenableFuture<Integer> fetchCounterFuture = ...;
*
* // Falling back to a zero counter only in case the exception was a
* // TimeoutException.
* ListenableFuture<Integer> faultTolerantFuture = Futures.catchingAsync(
* fetchCounterFuture, FetchException.class,
* new AsyncFunction<FetchException, Integer>() {
* public ListenableFuture<Integer> apply(FetchException e)
* throws FetchException {
* if (omitDataOnFetchFailure) {
* return immediateFuture(0);
* }
* throw e;
* }
* }, directExecutor());}</pre>
*
* <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
* the discussion in the {@link ListenableFuture#addListener ListenableFuture.addListener}
* documentation. The documentation's warnings about "lightweight listeners" refer here to the
* work done during {@code AsyncFunction.apply}, not to any work done to complete the returned
* {@code Future}.
*
* @param input the primary input {@code Future}
* @param exceptionType the exception type that triggers use of {@code fallback}. The exception
* type is matched against the input's exception. "The input's exception" means the cause of
* the {@link ExecutionException} thrown by {@code input.get()} or, if {@code get()} throws a
* different kind of exception, that exception itself. To avoid hiding bugs and other
* unrecoverable errors, callers should prefer more specific types, avoiding {@code
* Throwable.class} in particular.
* @param fallback the {@link AsyncFunction} to be called if {@code input} fails with the expected
* exception type. The function's argument is the input's exception. "The input's exception"
* means the cause of the {@link ExecutionException} thrown by {@code input.get()} or, if
* {@code get()} throws a different kind of exception, that exception itself.
* @param executor the executor that runs {@code fallback} if {@code input} fails
* @since 19.0 (similar functionality in 14.0 as {@code withFallback})
*/
@CanIgnoreReturnValue // TODO(kak): @CheckReturnValue
@Partially.GwtIncompatible("AVAILABLE but requires exceptionType to be Throwable.class")
public static <V, X extends Throwable> ListenableFuture catchingAsync(
ListenableFuture<? extends V> input,
Class<X> exceptionType,
AsyncFunction<? super X, ? extends V> fallback,
Executor executor) {
return AbstractCatchingFuture.create(input, exceptionType, fallback, executor);
}
/**
* Returns a future that delegates to another but will finish early (via a {@link
* TimeoutException} wrapped in an {@link ExecutionException}) if the specified duration expires.
*
* <p>The delegate future is interrupted and cancelled if it times out.
*
* @param delegate The future to delegate to.
* @param time when to timeout the future
* @param unit the time unit of the time parameter
* @param scheduledExecutor The executor service to enforce the timeout.
*
* @since 19.0
*/
@GwtIncompatible // java.util.concurrent.ScheduledExecutorService
public static <V> ListenableFuture withTimeout(
ListenableFuture<V> delegate,
long time,
TimeUnit unit,
ScheduledExecutorService scheduledExecutor) {
return TimeoutFuture.create(delegate, time, unit, scheduledExecutor);
}
/**
* Returns a new {@code Future} whose result is asynchronously derived from the result of the
* given {@code Future}. If the given {@code Future} fails, the returned {@code Future} fails with
* the same exception (and the function is not invoked).
*
* <p>More precisely, the returned {@code Future} takes its result from a {@code Future} produced
* by applying the given {@code AsyncFunction} to the result of the original {@code Future}.
* Example usage:
*
* <pre> {@code
* ListenableFuture<RowKey> rowKeyFuture = indexService.lookUp(query);
* AsyncFunction<RowKey, QueryResult> queryFunction =
* new AsyncFunction<RowKey, QueryResult>() {
* public ListenableFuture<QueryResult> apply(RowKey rowKey) {
* return dataService.read(rowKey);
* }
* };
* ListenableFuture<QueryResult> queryFuture =
* transformAsync(rowKeyFuture, queryFunction);}</pre>
*
* <p>This overload, which does not accept an executor, uses {@code directExecutor}, a dangerous
* choice in some cases. See the discussion in the {@link ListenableFuture#addListener
* ListenableFuture.addListener} documentation. The documentation's warnings about "lightweight
* listeners" refer here to the work done during {@code AsyncFunction.apply}, not to any work done
* to complete the returned {@code Future}.
*
* <p>The returned {@code Future} attempts to keep its cancellation state in sync with that of the
* input future and that of the future returned by the function. That is, if the returned {@code
* Future} is cancelled, it will attempt to cancel the other two, and if either of the other two
* is cancelled, the returned {@code Future} will receive a callback in which it will attempt to
* cancel itself.
*
* @param input The future to transform
* @param function A function to transform the result of the input future to the result of the
* output future
* @return A future that holds result of the function (if the input succeeded) or the original
* input's failure (if not)
* @since 19.0 (in 11.0 as {@code transform})
*/
public static <I, O> ListenableFuture transformAsync(
ListenableFuture<I> input, AsyncFunction function) {
return AbstractTransformFuture.create(input, function);
}
/**
* Returns a new {@code Future} whose result is asynchronously derived from the result of the
* given {@code Future}. If the given {@code Future} fails, the returned {@code Future} fails with
* the same exception (and the function is not invoked).
*
* <p>More precisely, the returned {@code Future} takes its result from a {@code Future} produced
* by applying the given {@code AsyncFunction} to the result of the original {@code Future}.
* Example usage:
*
* <pre> {@code
* ListenableFuture<RowKey> rowKeyFuture = indexService.lookUp(query);
* AsyncFunction<RowKey, QueryResult> queryFunction =
* new AsyncFunction<RowKey, QueryResult>() {
* public ListenableFuture<QueryResult> apply(RowKey rowKey) {
* return dataService.read(rowKey);
* }
* };
* ListenableFuture<QueryResult> queryFuture =
* transformAsync(rowKeyFuture, queryFunction, executor);}</pre>
*
* <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
* the discussion in the {@link ListenableFuture#addListener ListenableFuture.addListener}
* documentation. The documentation's warnings about "lightweight listeners" refer here to the
* work done during {@code AsyncFunction.apply}, not to any work done to complete the returned
* {@code Future}.
*
* <p>The returned {@code Future} attempts to keep its cancellation state in sync with that of the
* input future and that of the future returned by the chain function. That is, if the returned
* {@code Future} is cancelled, it will attempt to cancel the other two, and if either of the
* other two is cancelled, the returned {@code Future} will receive a callback in which it will
* attempt to cancel itself.
*
* @param input The future to transform
* @param function A function to transform the result of the input future to the result of the
* output future
* @param executor Executor to run the function in.
* @return A future that holds result of the function (if the input succeeded) or the original
* input's failure (if not)
* @since 19.0 (in 11.0 as {@code transform})
*/
public static <I, O> ListenableFuture transformAsync(
ListenableFuture<I> input,
AsyncFunction<? super I, ? extends O> function,
Executor executor) {
return AbstractTransformFuture.create(input, function, executor);
}
/**
* Returns a new {@code Future} whose result is derived from the result of the given {@code
* Future}. If {@code input} fails, the returned {@code Future} fails with the same exception (and
* the function is not invoked). Example usage:
*
* <pre> {@code
* ListenableFuture<QueryResult> queryFuture = ...;
* Function<QueryResult, List rowsFunction =
* new Function<QueryResult, List() {
* public List<Row> apply(QueryResult queryResult) {
* return queryResult.getRows();
* }
* };
* ListenableFuture<List rowsFuture =
* transform(queryFuture, rowsFunction);}</pre>
*
* <p>This overload, which does not accept an executor, uses {@code directExecutor}, a dangerous
* choice in some cases. See the discussion in the {@link ListenableFuture#addListener
* ListenableFuture.addListener} documentation. The documentation's warnings about "lightweight
* listeners" refer here to the work done during {@code Function.apply}.
*
* <p>The returned {@code Future} attempts to keep its cancellation state in sync with that of the
* input future. That is, if the returned {@code Future} is cancelled, it will attempt to cancel
* the input, and if the input is cancelled, the returned {@code Future} will receive a callback
* in which it will attempt to cancel itself.
*
* <p>An example use of this method is to convert a serializable object returned from an RPC into
* a POJO.
*
* @param input The future to transform
* @param function A Function to transform 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 transformation.
* @since 9.0 (in 1.0 as {@code compose})
*/
public static <I, O> ListenableFuture transform(
ListenableFuture<I> input, Function function) {
return AbstractTransformFuture.create(input, function);
}
/**
* Returns a new {@code Future} whose result is derived from the result of the given {@code
* Future}. If {@code input} fails, the returned {@code Future} fails with the same exception (and
* the function is not invoked). Example usage:
*
* <pre> {@code
* ListenableFuture<QueryResult> queryFuture = ...;
* Function<QueryResult, List rowsFunction =
* new Function<QueryResult, List() {
* public List<Row> apply(QueryResult queryResult) {
* return queryResult.getRows();
* }
* };
* ListenableFuture<List rowsFuture =
* transform(queryFuture, rowsFunction, executor);}</pre>
*
* <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
* the discussion in the {@link ListenableFuture#addListener ListenableFuture.addListener}
* documentation. The documentation's warnings about "lightweight listeners" refer here to the
* work done during {@code Function.apply}.
*
* <p>The returned {@code Future} attempts to keep its cancellation state in sync with that of the
* input future. That is, if the returned {@code Future} is cancelled, it will attempt to cancel
* the input, and if the input is cancelled, the returned {@code Future} will receive a callback
* in which it will attempt to cancel itself.
*
* <p>An example use of this method is to convert a serializable object returned from an RPC into
* a POJO.
*
* @param input The future to transform
* @param function A Function to transform the results of the provided future to the results of
* the returned future.
* @param executor Executor to run the function in.
* @return A future that holds result of the transformation.
* @since 9.0 (in 2.0 as {@code compose})
*/
public static <I, O> ListenableFuture transform(
ListenableFuture<I> input, Function function, Executor executor) {
return AbstractTransformFuture.create(input, function, executor);
}
/**
* Like {@link #transform(ListenableFuture, Function)} except that the transformation {@code
* function} is invoked on each call to {@link Future#get() get()} on the returned future.
*
* <p>The returned {@code Future} reflects the input's cancellation state directly, and any
* attempt to cancel the returned Future is likewise passed through to the input Future.
*
* <p>Note that calls to {@linkplain Future#get(long, TimeUnit) timed get} only apply the timeout
* to the execution of the underlying {@code Future}, <em>not to the execution of the
* transformation function.
*
* <p>The primary audience of this method is callers of {@code transform} who don't have a {@code
* ListenableFuture} available and do not mind repeated, lazy function evaluation.
*
* @param input The future to transform
* @param function A Function to transform the results of the provided future to the results of
* the returned future.
* @return A future that returns the result of the transformation.
* @since 10.0
*/
@GwtIncompatible // TODO
public static <I, O> Future lazyTransform(
final Future<I> input, final Function function) {
checkNotNull(input);
checkNotNull(function);
return new Future<O>() {
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
return input.cancel(mayInterruptIfRunning);
}
@Override
public boolean isCancelled() {
return input.isCancelled();
}
@Override
public boolean isDone() {
return input.isDone();
}
@Override
public O get() throws InterruptedException, ExecutionException {
return applyTransformation(input.get());
}
@Override
public O get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
return applyTransformation(input.get(timeout, unit));
}
private O applyTransformation(I input) throws ExecutionException {
try {
return function.apply(input);
} catch (Throwable t) {
throw new ExecutionException(t);
}
}
};
}
/**
* Returns a new {@code ListenableFuture} whose result is the product of calling {@code get()} on
* the {@code Future} nested within the given {@code Future}, effectively chaining the futures one
* after the other. Example:
*
* <pre> {@code
* SettableFuture<ListenableFuture nested = SettableFuture.create();
* ListenableFuture<String> dereferenced = dereference(nested);}
*
* <p>Most users will not need this method. To create a {@code Future} that completes with the
* result of another {@code Future}, create a {@link SettableFuture}, and call {@link
* SettableFuture#setFuture setFuture(otherFuture)} on it.
*
* <p>{@code dereference} has the same cancellation and execution semantics as {@link
* #transformAsync(ListenableFuture, AsyncFunction)}, in that the returned {@code Future}
* attempts to keep its cancellation state in sync with both the input {@code Future} and the
* nested {@code Future}. The transformation is very lightweight and therefore takes place in
* the same thread (either the thread that called {@code dereference}, or the thread in which
* the dereferenced future completes).
*
* @param nested The nested future to transform.
* @return A future that holds result of the inner future.
* @since 13.0
*/
@SuppressWarnings({"rawtypes", "unchecked"})
public static <V> ListenableFuture dereference(
ListenableFuture<? extends ListenableFuture nested) {
return transformAsync((ListenableFuture) nested, (AsyncFunction) DEREFERENCER);
}
/**
* Helper {@code Function} for {@link #dereference}.
*/
private static final AsyncFunction<ListenableFuture DEREFERENCER =
new AsyncFunction<ListenableFuture() {
@Override
public ListenableFuture<Object> apply(ListenableFuture
