001/*
002 * Copyright (C) 2006 The Guava Authors
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
005 * in compliance with the License. You may obtain a copy of the License at
006 *
007 * http://www.apache.org/licenses/LICENSE-2.0
008 *
009 * Unless required by applicable law or agreed to in writing, software distributed under the License
010 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
011 * or implied. See the License for the specific language governing permissions and limitations under
012 * the License.
013 */
014
015package com.google.common.util.concurrent;
016
017import static com.google.common.base.Preconditions.checkNotNull;
018import static com.google.common.util.concurrent.Internal.toNanosSaturated;
019
020import com.google.common.annotations.GwtCompatible;
021import com.google.common.annotations.GwtIncompatible;
022import com.google.common.annotations.J2ktIncompatible;
023import com.google.common.base.Function;
024import com.google.errorprone.annotations.CanIgnoreReturnValue;
025import com.google.errorprone.annotations.DoNotMock;
026import java.time.Duration;
027import java.util.concurrent.ExecutionException;
028import java.util.concurrent.Executor;
029import java.util.concurrent.ScheduledExecutorService;
030import java.util.concurrent.TimeUnit;
031import java.util.concurrent.TimeoutException;
032import org.checkerframework.checker.nullness.qual.Nullable;
033
034/**
035 * A {@link ListenableFuture} that supports fluent chains of operations. For example:
036 *
037 * <pre>{@code
038 * ListenableFuture<Boolean> adminIsLoggedIn =
039 *     FluentFuture.from(usersDatabase.getAdminUser())
040 *         .transform(User::getId, directExecutor())
041 *         .transform(ActivityService::isLoggedIn, threadPool)
042 *         .catching(RpcException.class, e -> false, directExecutor());
043 * }</pre>
044 *
045 * <h3>Alternatives</h3>
046 *
047 * <h4>Frameworks</h4>
048 *
049 * <p>When chaining together a graph of asynchronous operations, you will often find it easier to
050 * use a framework. Frameworks automate the process, often adding features like monitoring,
051 * debugging, and cancellation. Examples of frameworks include:
052 *
053 * <ul>
054 *   <li><a href="https://dagger.dev/producers.html">Dagger Producers</a>
055 * </ul>
056 *
057 * <h4>{@link java.util.concurrent.CompletableFuture} / {@link java.util.concurrent.CompletionStage}
058 * </h4>
059 *
060 * <p>Users of {@code CompletableFuture} will likely want to continue using {@code
061 * CompletableFuture}. {@code FluentFuture} is targeted at people who use {@code ListenableFuture},
062 * who can't use Java 8, or who want an API more focused than {@code CompletableFuture}. (If you
063 * need to adapt between {@code CompletableFuture} and {@code ListenableFuture}, consider <a
064 * href="https://github.com/lukas-krecan/future-converter">Future Converter</a>.)
065 *
066 * <h3>Extension</h3>
067 *
068 * If you want a class like {@code FluentFuture} but with extra methods, we recommend declaring your
069 * own subclass of {@link ListenableFuture}, complete with a method like {@link #from} to adapt an
070 * existing {@code ListenableFuture}, implemented atop a {@link ForwardingListenableFuture} that
071 * forwards to that future and adds the desired methods.
072 *
073 * @since 23.0
074 */
075@DoNotMock("Use FluentFuture.from(Futures.immediate*Future) or SettableFuture")
076@GwtCompatible(emulated = true)
077@ElementTypesAreNonnullByDefault
078public abstract class FluentFuture<V extends @Nullable Object>
079    extends GwtFluentFutureCatchingSpecialization<V> {
080
081  /**
082   * A less abstract subclass of AbstractFuture. This can be used to optimize setFuture by ensuring
083   * that {@link #get} calls exactly the implementation of {@link AbstractFuture#get}.
084   */
085  abstract static class TrustedFuture<V extends @Nullable Object> extends FluentFuture<V>
086      implements AbstractFuture.Trusted<V> {
087    @CanIgnoreReturnValue
088    @Override
089    @ParametricNullness
090    public final V get() throws InterruptedException, ExecutionException {
091      return super.get();
092    }
093
094    @CanIgnoreReturnValue
095    @Override
096    @ParametricNullness
097    public final V get(long timeout, TimeUnit unit)
098        throws InterruptedException, ExecutionException, TimeoutException {
099      return super.get(timeout, unit);
100    }
101
102    @Override
103    public final boolean isDone() {
104      return super.isDone();
105    }
106
107    @Override
108    public final boolean isCancelled() {
109      return super.isCancelled();
110    }
111
112    @Override
113    public final void addListener(Runnable listener, Executor executor) {
114      super.addListener(listener, executor);
115    }
116
117    @CanIgnoreReturnValue
118    @Override
119    public final boolean cancel(boolean mayInterruptIfRunning) {
120      return super.cancel(mayInterruptIfRunning);
121    }
122  }
123
124  FluentFuture() {}
125
126  /**
127   * Converts the given {@code ListenableFuture} to an equivalent {@code FluentFuture}.
128   *
129   * <p>If the given {@code ListenableFuture} is already a {@code FluentFuture}, it is returned
130   * directly. If not, it is wrapped in a {@code FluentFuture} that delegates all calls to the
131   * original {@code ListenableFuture}.
132   */
133  public static <V extends @Nullable Object> FluentFuture<V> from(ListenableFuture<V> future) {
134    return future instanceof FluentFuture
135        ? (FluentFuture<V>) future
136        : new ForwardingFluentFuture<V>(future);
137  }
138
139  /**
140   * Simply returns its argument.
141   *
142   * @deprecated no need to use this
143   * @since 28.0
144   */
145  @Deprecated
146  public static <V extends @Nullable Object> FluentFuture<V> from(FluentFuture<V> future) {
147    return checkNotNull(future);
148  }
149
150  /**
151   * Returns a {@code Future} whose result is taken from this {@code Future} or, if this {@code
152   * Future} fails with the given {@code exceptionType}, from the result provided by the {@code
153   * fallback}. {@link Function#apply} is not invoked until the primary input has failed, so if the
154   * primary input succeeds, it is never invoked. If, during the invocation of {@code fallback}, an
155   * exception is thrown, this exception is used as the result of the output {@code Future}.
156   *
157   * <p>Usage example:
158   *
159   * <pre>{@code
160   * // Falling back to a zero counter in case an exception happens when processing the RPC to fetch
161   * // counters.
162   * ListenableFuture<Integer> faultTolerantFuture =
163   *     fetchCounters().catching(FetchException.class, x -> 0, directExecutor());
164   * }</pre>
165   *
166   * <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
167   * the discussion in the {@link #addListener} documentation. All its warnings about heavyweight
168   * listeners are also applicable to heavyweight functions passed to this method.
169   *
170   * <p>This method is similar to {@link java.util.concurrent.CompletableFuture#exceptionally}. It
171   * can also serve some of the use cases of {@link java.util.concurrent.CompletableFuture#handle}
172   * and {@link java.util.concurrent.CompletableFuture#handleAsync} when used along with {@link
173   * #transform}.
174   *
175   * @param exceptionType the exception type that triggers use of {@code fallback}. The exception
176   *     type is matched against the input's exception. "The input's exception" means the cause of
177   *     the {@link ExecutionException} thrown by {@code input.get()} or, if {@code get()} throws a
178   *     different kind of exception, that exception itself. To avoid hiding bugs and other
179   *     unrecoverable errors, callers should prefer more specific types, avoiding {@code
180   *     Throwable.class} in particular.
181   * @param fallback the {@link Function} to be called if the input fails with the expected
182   *     exception type. The function's argument is the input's exception. "The input's exception"
183   *     means the cause of the {@link ExecutionException} thrown by {@code this.get()} or, if
184   *     {@code get()} throws a different kind of exception, that exception itself.
185   * @param executor the executor that runs {@code fallback} if the input fails
186   */
187  @J2ktIncompatible
188  @Partially.GwtIncompatible("AVAILABLE but requires exceptionType to be Throwable.class")
189  public final <X extends Throwable> FluentFuture<V> catching(
190      Class<X> exceptionType, Function<? super X, ? extends V> fallback, Executor executor) {
191    return (FluentFuture<V>) Futures.catching(this, exceptionType, fallback, executor);
192  }
193
194  /**
195   * Returns a {@code Future} whose result is taken from this {@code Future} or, if this {@code
196   * Future} fails with the given {@code exceptionType}, from the result provided by the {@code
197   * fallback}. {@link AsyncFunction#apply} is not invoked until the primary input has failed, so if
198   * the primary input succeeds, it is never invoked. If, during the invocation of {@code fallback},
199   * an exception is thrown, this exception is used as the result of the output {@code Future}.
200   *
201   * <p>Usage examples:
202   *
203   * <pre>{@code
204   * // Falling back to a zero counter in case an exception happens when processing the RPC to fetch
205   * // counters.
206   * ListenableFuture<Integer> faultTolerantFuture =
207   *     fetchCounters().catchingAsync(
208   *         FetchException.class, x -> immediateFuture(0), directExecutor());
209   * }</pre>
210   *
211   * <p>The fallback can also choose to propagate the original exception when desired:
212   *
213   * <pre>{@code
214   * // Falling back to a zero counter only in case the exception was a
215   * // TimeoutException.
216   * ListenableFuture<Integer> faultTolerantFuture =
217   *     fetchCounters().catchingAsync(
218   *         FetchException.class,
219   *         e -> {
220   *           if (omitDataOnFetchFailure) {
221   *             return immediateFuture(0);
222   *           }
223   *           throw e;
224   *         },
225   *         directExecutor());
226   * }</pre>
227   *
228   * <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
229   * the discussion in the {@link #addListener} documentation. All its warnings about heavyweight
230   * listeners are also applicable to heavyweight functions passed to this method. (Specifically,
231   * {@code directExecutor} functions should avoid heavyweight operations inside {@code
232   * AsyncFunction.apply}. Any heavyweight operations should occur in other threads responsible for
233   * completing the returned {@code Future}.)
234   *
235   * <p>This method is similar to {@link java.util.concurrent.CompletableFuture#exceptionally}. It
236   * can also serve some of the use cases of {@link java.util.concurrent.CompletableFuture#handle}
237   * and {@link java.util.concurrent.CompletableFuture#handleAsync} when used along with {@link
238   * #transform}.
239   *
240   * @param exceptionType the exception type that triggers use of {@code fallback}. The exception
241   *     type is matched against the input's exception. "The input's exception" means the cause of
242   *     the {@link ExecutionException} thrown by {@code this.get()} or, if {@code get()} throws a
243   *     different kind of exception, that exception itself. To avoid hiding bugs and other
244   *     unrecoverable errors, callers should prefer more specific types, avoiding {@code
245   *     Throwable.class} in particular.
246   * @param fallback the {@link AsyncFunction} to be called if the input fails with the expected
247   *     exception type. The function's argument is the input's exception. "The input's exception"
248   *     means the cause of the {@link ExecutionException} thrown by {@code input.get()} or, if
249   *     {@code get()} throws a different kind of exception, that exception itself.
250   * @param executor the executor that runs {@code fallback} if the input fails
251   */
252  @J2ktIncompatible
253  @Partially.GwtIncompatible("AVAILABLE but requires exceptionType to be Throwable.class")
254  public final <X extends Throwable> FluentFuture<V> catchingAsync(
255      Class<X> exceptionType, AsyncFunction<? super X, ? extends V> fallback, Executor executor) {
256    return (FluentFuture<V>) Futures.catchingAsync(this, exceptionType, fallback, executor);
257  }
258
259  /**
260   * Returns a future that delegates to this future but will finish early (via a {@link
261   * TimeoutException} wrapped in an {@link ExecutionException}) if the specified timeout expires.
262   * If the timeout expires, not only will the output future finish, but also the input future
263   * ({@code this}) will be cancelled and interrupted.
264   *
265   * @param timeout when to time out the future
266   * @param scheduledExecutor The executor service to enforce the timeout.
267   * @since 28.0
268   */
269  @J2ktIncompatible
270  @GwtIncompatible // ScheduledExecutorService
271  public final FluentFuture<V> withTimeout(
272      Duration timeout, ScheduledExecutorService scheduledExecutor) {
273    return withTimeout(toNanosSaturated(timeout), TimeUnit.NANOSECONDS, scheduledExecutor);
274  }
275
276  /**
277   * Returns a future that delegates to this future but will finish early (via a {@link
278   * TimeoutException} wrapped in an {@link ExecutionException}) if the specified timeout expires.
279   * If the timeout expires, not only will the output future finish, but also the input future
280   * ({@code this}) will be cancelled and interrupted.
281   *
282   * @param timeout when to time out the future
283   * @param unit the time unit of the time parameter
284   * @param scheduledExecutor The executor service to enforce the timeout.
285   */
286  @J2ktIncompatible
287  @GwtIncompatible // ScheduledExecutorService
288  @SuppressWarnings("GoodTime") // should accept a java.time.Duration
289  public final FluentFuture<V> withTimeout(
290      long timeout, TimeUnit unit, ScheduledExecutorService scheduledExecutor) {
291    return (FluentFuture<V>) Futures.withTimeout(this, timeout, unit, scheduledExecutor);
292  }
293
294  /**
295   * Returns a new {@code Future} whose result is asynchronously derived from the result of this
296   * {@code Future}. If the input {@code Future} fails, the returned {@code Future} fails with the
297   * same exception (and the function is not invoked).
298   *
299   * <p>More precisely, the returned {@code Future} takes its result from a {@code Future} produced
300   * by applying the given {@code AsyncFunction} to the result of the original {@code Future}.
301   * Example usage:
302   *
303   * <pre>{@code
304   * FluentFuture<RowKey> rowKeyFuture = FluentFuture.from(indexService.lookUp(query));
305   * ListenableFuture<QueryResult> queryFuture =
306   *     rowKeyFuture.transformAsync(dataService::readFuture, executor);
307   * }</pre>
308   *
309   * <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
310   * the discussion in the {@link #addListener} documentation. All its warnings about heavyweight
311   * listeners are also applicable to heavyweight functions passed to this method. (Specifically,
312   * {@code directExecutor} functions should avoid heavyweight operations inside {@code
313   * AsyncFunction.apply}. Any heavyweight operations should occur in other threads responsible for
314   * completing the returned {@code Future}.)
315   *
316   * <p>The returned {@code Future} attempts to keep its cancellation state in sync with that of the
317   * input future and that of the future returned by the chain function. That is, if the returned
318   * {@code Future} is cancelled, it will attempt to cancel the other two, and if either of the
319   * other two is cancelled, the returned {@code Future} will receive a callback in which it will
320   * attempt to cancel itself.
321   *
322   * <p>This method is similar to {@link java.util.concurrent.CompletableFuture#thenCompose} and
323   * {@link java.util.concurrent.CompletableFuture#thenComposeAsync}. It can also serve some of the
324   * use cases of {@link java.util.concurrent.CompletableFuture#handle} and {@link
325   * java.util.concurrent.CompletableFuture#handleAsync} when used along with {@link #catching}.
326   *
327   * @param function A function to transform the result of this future to the result of the output
328   *     future
329   * @param executor Executor to run the function in.
330   * @return A future that holds result of the function (if the input succeeded) or the original
331   *     input's failure (if not)
332   */
333  public final <T extends @Nullable Object> FluentFuture<T> transformAsync(
334      AsyncFunction<? super V, T> function, Executor executor) {
335    return (FluentFuture<T>) Futures.transformAsync(this, function, executor);
336  }
337
338  /**
339   * Returns a new {@code Future} whose result is derived from the result of this {@code Future}. If
340   * this input {@code Future} fails, the returned {@code Future} fails with the same exception (and
341   * the function is not invoked). Example usage:
342   *
343   * <pre>{@code
344   * ListenableFuture<List<Row>> rowsFuture =
345   *     queryFuture.transform(QueryResult::getRows, executor);
346   * }</pre>
347   *
348   * <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
349   * the discussion in the {@link #addListener} documentation. All its warnings about heavyweight
350   * listeners are also applicable to heavyweight functions passed to this method.
351   *
352   * <p>The returned {@code Future} attempts to keep its cancellation state in sync with that of the
353   * input future. That is, if the returned {@code Future} is cancelled, it will attempt to cancel
354   * the input, and if the input is cancelled, the returned {@code Future} will receive a callback
355   * in which it will attempt to cancel itself.
356   *
357   * <p>An example use of this method is to convert a serializable object returned from an RPC into
358   * a POJO.
359   *
360   * <p>This method is similar to {@link java.util.concurrent.CompletableFuture#thenApply} and
361   * {@link java.util.concurrent.CompletableFuture#thenApplyAsync}. It can also serve some of the
362   * use cases of {@link java.util.concurrent.CompletableFuture#handle} and {@link
363   * java.util.concurrent.CompletableFuture#handleAsync} when used along with {@link #catching}.
364   *
365   * @param function A Function to transform the results of this future to the results of the
366   *     returned future.
367   * @param executor Executor to run the function in.
368   * @return A future that holds result of the transformation.
369   */
370  public final <T extends @Nullable Object> FluentFuture<T> transform(
371      Function<? super V, T> function, Executor executor) {
372    return (FluentFuture<T>) Futures.transform(this, function, executor);
373  }
374
375  /**
376   * Registers separate success and failure callbacks to be run when this {@code Future}'s
377   * computation is {@linkplain java.util.concurrent.Future#isDone() complete} or, if the
378   * computation is already complete, immediately.
379   *
380   * <p>The callback is run on {@code executor}. There is no guaranteed ordering of execution of
381   * callbacks, but any callback added through this method is guaranteed to be called once the
382   * computation is complete.
383   *
384   * <p>Example:
385   *
386   * <pre>{@code
387   * future.addCallback(
388   *     new FutureCallback<QueryResult>() {
389   *       public void onSuccess(QueryResult result) {
390   *         storeInCache(result);
391   *       }
392   *       public void onFailure(Throwable t) {
393   *         reportError(t);
394   *       }
395   *     }, executor);
396   * }</pre>
397   *
398   * <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
399   * the discussion in the {@link #addListener} documentation. All its warnings about heavyweight
400   * listeners are also applicable to heavyweight callbacks passed to this method.
401   *
402   * <p>For a more general interface to attach a completion listener, see {@link #addListener}.
403   *
404   * <p>This method is similar to {@link java.util.concurrent.CompletableFuture#whenComplete} and
405   * {@link java.util.concurrent.CompletableFuture#whenCompleteAsync}. It also serves the use case
406   * of {@link java.util.concurrent.CompletableFuture#thenAccept} and {@link
407   * java.util.concurrent.CompletableFuture#thenAcceptAsync}.
408   *
409   * @param callback The callback to invoke when this {@code Future} is completed.
410   * @param executor The executor to run {@code callback} when the future completes.
411   */
412  public final void addCallback(FutureCallback<? super V> callback, Executor executor) {
413    Futures.addCallback(this, callback, executor);
414  }
415}