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