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}